AVRO-4174: [C++] Support read and write key-value metadata (#3468)

* AVRO-4174: [C++] Support read and write key-value metadata

* refine docstring

Author: wgtmac

lang/c++/impl/DataFile.cc

@@ -66,30 +66,33 @@ const size_t zlibBufGrowSize = 128 * 1024;
 } // namespace
 
 DataFileWriterBase::DataFileWriterBase(const char *filename, const ValidSchema &schema, size_t syncInterval,
-                                       Codec codec) : filename_(filename),
-                                                      schema_(schema),
-                                                      encoderPtr_(binaryEncoder()),
-                                                      syncInterval_(syncInterval),
-                                                      codec_(codec),
-                                                      stream_(fileOutputStream(filename)),
-                                                      buffer_(memoryOutputStream()),
-                                                      sync_(makeSync()),
-                                                      objectCount_(0),
-                                                      lastSync_(0) {
+                                       Codec codec, const Metadata &metadata) : filename_(filename),
+                                                                                schema_(schema),
+                                                                                encoderPtr_(binaryEncoder()),
+                                                                                syncInterval_(syncInterval),
+                                                                                codec_(codec),
+                                                                                stream_(fileOutputStream(filename)),
+                                                                                buffer_(memoryOutputStream()),
+                                                                                sync_(makeSync()),
+                                                                                objectCount_(0),
+                                                                                metadata_(metadata),
+                                                                                lastSync_(0) {
     init(schema, syncInterval, codec);
 }
 
 DataFileWriterBase::DataFileWriterBase(std::unique_ptr<OutputStream> outputStream,
-                                       const ValidSchema &schema, size_t syncInterval, Codec codec) : filename_(),
-                                                                                                      schema_(schema),
-                                                                                                      encoderPtr_(binaryEncoder()),
-                                                                                                      syncInterval_(syncInterval),
-                                                                                                      codec_(codec),
-                                                                                                      stream_(std::move(outputStream)),
-                                                                                                      buffer_(memoryOutputStream()),
-                                                                                                      sync_(makeSync()),
-                                                                                                      objectCount_(0),
-                                                                                                      lastSync_(0) {
+                                       const ValidSchema &schema, size_t syncInterval,
+                                       Codec codec, const Metadata &metadata) : filename_(),
+                                                                                schema_(schema),
+                                                                                encoderPtr_(binaryEncoder()),
+                                                                                syncInterval_(syncInterval),
+                                                                                codec_(codec),
+                                                                                stream_(std::move(outputStream)),
+                                                                                buffer_(memoryOutputStream()),
+                                                                                sync_(makeSync()),
+                                                                                objectCount_(0),
+                                                                                metadata_(metadata),
+                                                                                lastSync_(0) {
     init(schema, syncInterval, codec);
 }
 

lang/c++/include/avro/DataFile.hh

@@ -54,6 +54,22 @@ const int SyncSize = 16;
  */
 typedef std::array<uint8_t, SyncSize> DataFileSync;
 
+/**
+ * Avro files may include arbitrary user-specified metadata.
+ * File metadata is written as if defined by the following map schema:
+ *
+ * `{"type": "map", "values": "bytes"}`
+ *
+ * All metadata properties that start with "avro." are reserved.
+ * The following file metadata properties are currently used:
+ *
+ * - `avro.schema` contains the schema of objects stored in the file, as JSON data (required).
+ * - `avro.codec`, the name of the compression codec used to compress blocks, as a string.
+ *   Implementations are required to support the following codecs: "null" and "deflate".
+ *   If codec is absent, it is assumed to be "null". See avro.codecs for implementation details.
+ */
+typedef std::map<std::string, std::vector<uint8_t>> Metadata;
+
 /**
  * Type-independent portion of DataFileWriter.
  *  At any given point in time, at most one file can be written using
@@ -71,8 +87,6 @@ class AVRO_DECL DataFileWriterBase {
     const DataFileSync sync_;
     int64_t objectCount_;
 
-    typedef std::map<std::string, std::vector<uint8_t>> Metadata;
-
     Metadata metadata_;
     int64_t lastSync_;
 
@@ -119,9 +133,11 @@ public:
      * Constructs a data file writer with the given sync interval and name.
      */
     DataFileWriterBase(const char *filename, const ValidSchema &schema,
-                       size_t syncInterval, Codec codec = NULL_CODEC);
+                       size_t syncInterval, Codec codec = NULL_CODEC,
+                       const Metadata &metadata = {});
     DataFileWriterBase(std::unique_ptr<OutputStream> outputStream,
-                       const ValidSchema &schema, size_t syncInterval, Codec codec);
+                       const ValidSchema &schema, size_t syncInterval, Codec codec,
+                       const Metadata &metadata = {});
 
     DataFileWriterBase(const DataFileWriterBase &) = delete;
     DataFileWriterBase &operator=(const DataFileWriterBase &) = delete;
@@ -156,10 +172,12 @@ public:
      * Constructs a new data file.
      */
     DataFileWriter(const char *filename, const ValidSchema &schema,
-                   size_t syncInterval = 16 * 1024, Codec codec = NULL_CODEC) : base_(new DataFileWriterBase(filename, schema, syncInterval, codec)) {}
+                   size_t syncInterval = 16 * 1024, Codec codec = NULL_CODEC,
+                   const Metadata &metadata = {}) : base_(std::make_unique<DataFileWriterBase>(filename, schema, syncInterval, codec, metadata)) {}
 
     DataFileWriter(std::unique_ptr<OutputStream> outputStream, const ValidSchema &schema,
-                   size_t syncInterval = 16 * 1024, Codec codec = NULL_CODEC) : base_(new DataFileWriterBase(std::move(outputStream), schema, syncInterval, codec)) {}
+                   size_t syncInterval = 16 * 1024, Codec codec = NULL_CODEC,
+                   const Metadata &metadata = {}) : base_(std::make_unique<DataFileWriterBase>(std::move(outputStream), schema, syncInterval, codec, metadata)) {}
 
     DataFileWriter(const DataFileWriter &) = delete;
     DataFileWriter &operator=(const DataFileWriter &) = delete;
@@ -212,7 +230,6 @@ class AVRO_DECL DataFileReaderBase {
     ValidSchema dataSchema_;
     DecoderPtr dataDecoder_;
     std::unique_ptr<InputStream> dataStream_;
-    typedef std::map<std::string, std::vector<uint8_t>> Metadata;
 
     Metadata metadata_;
     DataFileSync sync_{};
@@ -306,6 +323,11 @@ public:
      * Return the last synchronization point before our current position.
      */
     int64_t previousSync() const;
+
+    /**
+     * Return the metadata for the data file.
+     */
+    const Metadata &metadata() const { return metadata_; }
 };
 
 /**
@@ -421,6 +443,11 @@ public:
      * Return the last synchronization point before our current position.
      */
     int64_t previousSync() { return base_->previousSync(); }
+
+    /**
+     * Return the metadata for the data file.
+     */
+    const Metadata &metadata() const { return base_->metadata(); }
 };
 
 } // namespace avro

lang/c++/test/DataFileTests.cc

@@ -716,6 +716,60 @@ class DataFileTest {
         ComplexDouble unused;
         BOOST_CHECK_NO_THROW(df.write(unused)); // write has not effect on closed stream
     }
+
+    void testMetadata() {
+        avro::Metadata customMetadata;
+        std::string key1 = "author";
+        std::string value1 = "test-user";
+        customMetadata[key1] = std::vector<uint8_t>(value1.begin(), value1.end());
+
+        std::string key2 = "version";
+        std::string value2 = "1.0.0";
+        customMetadata[key2] = std::vector<uint8_t>(value2.begin(), value2.end());
+
+        std::string key3 = "description";
+        std::string value3 = "Test file with custom metadata";
+        customMetadata[key3] = std::vector<uint8_t>(value3.begin(), value3.end());
+
+        // Write data with custom metadata
+        {
+            avro::DataFileWriter<ComplexInteger> df(filename, writerSchema, 100, avro::NULL_CODEC, customMetadata);
+            int64_t re = 10;
+            int64_t im = 20;
+            for (int i = 0; i < 5; ++i, re += 5, im += 10) {
+                ComplexInteger c(re, im);
+                df.write(c);
+            }
+            df.close();
+        }
+
+        // Read and verify metadata
+        {
+            avro::DataFileReader<ComplexInteger> df(filename, writerSchema);
+            const avro::Metadata &readMetadata = df.metadata();
+
+            // Check that our custom metadata is present
+            auto it1 = readMetadata.find(key1);
+            BOOST_CHECK(it1 != readMetadata.end());
+            BOOST_CHECK_EQUAL(std::string(it1->second.begin(), it1->second.end()), value1);
+
+            auto it2 = readMetadata.find(key2);
+            BOOST_CHECK(it2 != readMetadata.end());
+            BOOST_CHECK_EQUAL(std::string(it2->second.begin(), it2->second.end()), value2);
+
+            auto it3 = readMetadata.find(key3);
+            BOOST_CHECK(it3 != readMetadata.end());
+            BOOST_CHECK_EQUAL(std::string(it3->second.begin(), it3->second.end()), value3);
+
+            // Check that standard metadata is also present
+            auto schemaIt = readMetadata.find("avro.schema");
+            BOOST_CHECK(schemaIt != readMetadata.end());
+
+            auto codecIt = readMetadata.find("avro.codec");
+            BOOST_CHECK(codecIt != readMetadata.end());
+            BOOST_CHECK_EQUAL(std::string(codecIt->second.begin(), codecIt->second.end()), "null");
+        }
+    }
 };
 
 void addReaderTests(test_suite *ts, const shared_ptr<DataFileTest> &t) {
@@ -1082,6 +1136,79 @@ void testReadRecordEfficientlyUsingLastSyncZstdCodec() {
 }
 #endif
 
+void testMetadataWithCodec(avro::Codec codec) {
+    const char *filename = "test_metadata_codec.df";
+    avro::ValidSchema schema = avro::compileJsonSchemaFromString(sch);
+
+    avro::Metadata customMetadata;
+    std::string key1 = "test.key1";
+    std::string value1 = "test-value-1";
+    customMetadata[key1] = std::vector<uint8_t>(value1.begin(), value1.end());
+
+    std::string key2 = "test.key2";
+    std::string value2 = "test-value-2-with-special-chars: !@#$%^&*()";
+    customMetadata[key2] = std::vector<uint8_t>(value2.begin(), value2.end());
+
+    // Write data with custom metadata
+    {
+        avro::DataFileWriter<ComplexInteger> writer(filename, schema, 100, codec, customMetadata);
+        for (int i = 0; i < 10; ++i) {
+            ComplexInteger c(i * 2, i * 3);
+            writer.write(c);
+        }
+        writer.close();
+    }
+
+    // Read and verify metadata
+    {
+        avro::DataFileReader<ComplexInteger> reader(filename, schema);
+        const avro::Metadata &readMetadata = reader.metadata();
+
+        // Verify custom metadata
+        auto it1 = readMetadata.find(key1);
+        BOOST_CHECK(it1 != readMetadata.end());
+        BOOST_CHECK_EQUAL(std::string(it1->second.begin(), it1->second.end()), value1);
+
+        auto it2 = readMetadata.find(key2);
+        BOOST_CHECK(it2 != readMetadata.end());
+        BOOST_CHECK_EQUAL(std::string(it2->second.begin(), it2->second.end()), value2);
+
+        // Verify standard metadata
+        auto schemaIt = readMetadata.find("avro.schema");
+        BOOST_CHECK(schemaIt != readMetadata.end());
+
+        auto codecIt = readMetadata.find("avro.codec");
+        BOOST_CHECK(codecIt != readMetadata.end());
+    }
+
+    // Clean up
+    std::filesystem::remove(filename);
+}
+
+void testMetadataWithNullCodec() {
+    BOOST_TEST_CHECKPOINT(__func__);
+    testMetadataWithCodec(avro::NULL_CODEC);
+}
+
+void testMetadataWithDeflateCodec() {
+    BOOST_TEST_CHECKPOINT(__func__);
+    testMetadataWithCodec(avro::DEFLATE_CODEC);
+}
+
+#ifdef SNAPPY_CODEC_AVAILABLE
+void testMetadataWithSnappyCodec() {
+    BOOST_TEST_CHECKPOINT(__func__);
+    testMetadataWithCodec(avro::SNAPPY_CODEC);
+}
+#endif
+
+#ifdef ZSTD_CODEC_AVAILABLE
+void testMetadataWithZstdCodec() {
+    BOOST_TEST_CHECKPOINT(__func__);
+    testMetadataWithCodec(avro::ZSTD_CODEC);
+}
+#endif
+
 test_suite *
 init_unit_test_suite(int, char *[]) {
     {
@@ -1232,6 +1359,13 @@ init_unit_test_suite(int, char *[]) {
         ts->add(BOOST_CLASS_TEST_CASE(&DataFileTest::testCleanup, t));
         boost::unit_test::framework::master_test_suite().add(ts);
     }
+    {
+        auto *ts = BOOST_TEST_SUITE("DataFile tests: test15.df");
+        shared_ptr<DataFileTest> t(new DataFileTest("test15.df", sch, isch));
+        ts->add(BOOST_CLASS_TEST_CASE(&DataFileTest::testMetadata, t));
+        ts->add(BOOST_CLASS_TEST_CASE(&DataFileTest::testCleanup, t));
+        boost::unit_test::framework::master_test_suite().add(ts);
+    }
 
     boost::unit_test::framework::master_test_suite().add(BOOST_TEST_CASE(&testSkipStringNullCodec));
     boost::unit_test::framework::master_test_suite().add(BOOST_TEST_CASE(&testSkipStringDeflateCodec));
@@ -1260,5 +1394,14 @@ init_unit_test_suite(int, char *[]) {
     boost::unit_test::framework::master_test_suite().add(BOOST_TEST_CASE(&testReadRecordEfficientlyUsingLastSyncZstdCodec));
 #endif
 
+    boost::unit_test::framework::master_test_suite().add(BOOST_TEST_CASE(&testMetadataWithNullCodec));
+    boost::unit_test::framework::master_test_suite().add(BOOST_TEST_CASE(&testMetadataWithDeflateCodec));
+#ifdef SNAPPY_CODEC_AVAILABLE
+    boost::unit_test::framework::master_test_suite().add(BOOST_TEST_CASE(&testMetadataWithSnappyCodec));
+#endif
+#ifdef ZSTD_CODEC_AVAILABLE
+    boost::unit_test::framework::master_test_suite().add(BOOST_TEST_CASE(&testMetadataWithZstdCodec));
+#endif
+
     return nullptr;
 }