feat: support manifest reader

- add manifest reader
- refactor of arrow array parser
- add basic case

Author: xiao.dong

src/iceberg/avro/avro_schema_util.cc

@@ -58,8 +58,7 @@ ::avro::LogicalType GetMapLogicalType() {
   std::call_once(flag, []() {
     // Register the map logical type with the avro custom logical type registry.
     // See https://github.com/apache/avro/pull/3326 for details.
-    ::avro::CustomLogicalTypeRegistry::instance().registerType(
-        "map", [](const std::string&) { return std::make_shared<MapLogicalType>(); });
+    RegisterLogicalTypes();
   });
   return ::avro::LogicalType(std::make_shared<MapLogicalType>());
 }
@@ -73,6 +72,11 @@ ::avro::CustomAttributes GetAttributesWithFieldId(int32_t field_id) {
 
 }  // namespace
 
+void RegisterLogicalTypes() {
+  ::avro::CustomLogicalTypeRegistry::instance().registerType(
+      "map", [](const std::string&) { return std::make_shared<MapLogicalType>(); });
+}
+
 std::string ToString(const ::avro::NodePtr& node) {
   std::stringstream ss;
   ss << *node;

src/iceberg/avro/avro_schema_util_internal.h

@@ -144,4 +144,6 @@ std::string ToString(const ::avro::LogicalType::Type& logical_type);
 /// \return True if the node has a map logical type, false otherwise.
 bool HasMapLogicalType(const ::avro::NodePtr& node);
 
+void RegisterLogicalTypes();
+
 }  // namespace iceberg::avro

src/iceberg/file_format.h

@@ -27,6 +27,7 @@
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
+#include "iceberg/util/string_utils.h"
 
 namespace iceberg {
 
@@ -56,10 +57,11 @@ ICEBERG_EXPORT inline std::string_view ToString(FileFormatType format_type) {
 /// \brief Convert a string to a FileFormatType
 ICEBERG_EXPORT constexpr Result<FileFormatType> FileFormatTypeFromString(
     std::string_view str) noexcept {
-  if (str == "parquet") return FileFormatType::kParquet;
-  if (str == "avro") return FileFormatType::kAvro;
-  if (str == "orc") return FileFormatType::kOrc;
-  if (str == "puffin") return FileFormatType::kPuffin;
+  auto lower = internal::StringUtils::to_lower(str);
+  if (lower == "parquet") return FileFormatType::kParquet;
+  if (lower == "avro") return FileFormatType::kAvro;
+  if (lower == "orc") return FileFormatType::kOrc;
+  if (lower == "puffin") return FileFormatType::kPuffin;
   return InvalidArgument("Invalid file format type: {}", str);
 }
 

src/iceberg/manifest_entry.cc

@@ -40,7 +40,7 @@ std::shared_ptr<StructType> DataFile::Type(std::shared_ptr<StructType> partition
       kContent,
       kFilePath,
       kFileFormat,
-      SchemaField::MakeRequired(102, "partition", std::move(partition_type)),
+      SchemaField::MakeRequired(102, kPartitionField, std::move(partition_type)),
       kRecordCount,
       kFileSize,
       kColumnSizes,
@@ -68,7 +68,7 @@ std::shared_ptr<StructType> ManifestEntry::TypeFromDataFileType(
     std::shared_ptr<StructType> datafile_type) {
   return std::make_shared<StructType>(std::vector<SchemaField>{
       kStatus, kSnapshotId, kSequenceNumber, kFileSequenceNumber,
-      SchemaField::MakeRequired(2, "data_file", std::move(datafile_type))});
+      SchemaField::MakeRequired(2, kDataFileField, std::move(datafile_type))});
 }
 
 }  // namespace iceberg

src/iceberg/manifest_entry.h

@@ -182,7 +182,8 @@ struct ICEBERG_EXPORT DataFile {
   inline static const SchemaField kFilePath = SchemaField::MakeRequired(
       100, "file_path", iceberg::string(), "Location URI with FS scheme");
   inline static const SchemaField kFileFormat = SchemaField::MakeRequired(
-      101, "file_format", iceberg::int32(), "File format name: avro, orc, or parquet");
+      101, "file_format", iceberg::string(), "File format name: avro, orc, or parquet");
+  inline static const std::string kPartitionField = "partition";
   inline static const SchemaField kRecordCount = SchemaField::MakeRequired(
       103, "record_count", iceberg::int64(), "Number of records in the file");
   inline static const SchemaField kFileSize = SchemaField::MakeRequired(
@@ -299,6 +300,7 @@ struct ICEBERG_EXPORT ManifestEntry {
       SchemaField::MakeOptional(3, "sequence_number", iceberg::int64());
   inline static const SchemaField kFileSequenceNumber =
       SchemaField::MakeOptional(4, "file_sequence_number", iceberg::int64());
+  inline static const std::string kDataFileField = "data_file";
 
   bool operator==(const ManifestEntry& other) const;
 

src/iceberg/manifest_reader_internal.cc

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <algorithm>
+#include <string>
+
+namespace iceberg::internal {
+
+class StringUtils {
+ public:
+  static std::string to_lower(std::string_view str) {
+    std::string input(str);
+    std::transform(input.begin(), input.end(), input.begin(),
+                   [](char c) { return std::tolower(c); });
+    return input;
+  }
+
+  static std::string to_upper(std::string_view str) {
+    std::string input(str);
+    std::transform(input.begin(), input.end(), input.begin(),
+                   [](char c) { return std::toupper(c); });
+    return input;
+  }
+};
+
+}  // namespace iceberg::internal

src/iceberg/util/string_utils.h

@@ -78,6 +78,7 @@ if(ICEBERG_BUILD_BUNDLE)
                          avro_schema_test.cc
                          avro_stream_test.cc
                          manifest_list_reader_test.cc
+                         manifest_reader_test.cc
                          test_common.cc)
   target_link_libraries(avro_test PRIVATE iceberg_bundle_static GTest::gtest_main
                                           GTest::gmock)

test/CMakeLists.txt

@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/manifest_reader.h"
+
+#include <arrow/filesystem/localfs.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/avro/avro_reader.h"
+#include "iceberg/avro/avro_schema_util_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/schema.h"
+#include "temp_file_test_base.h"
+#include "test_common.h"
+
+namespace iceberg {
+
+class ManifestReaderTest : public TempFileTestBase {
+ protected:
+  static void SetUpTestSuite() { avro::AvroReader::Register(); }
+
+  void SetUp() override {
+    TempFileTestBase::SetUp();
+    local_fs_ = std::make_shared<::arrow::fs::LocalFileSystem>();
+    file_io_ = std::make_shared<iceberg::arrow::ArrowFileSystemFileIO>(local_fs_);
+
+    avro::RegisterLogicalTypes();
+  }
+
+  std::vector<ManifestEntry> prepare_manifest_entries() {
+    std::vector<ManifestEntry> manifest_entries;
+    std::string test_dir_prefix = "/tmp/db/db/iceberg_test/data/";
+    std::vector<std::string> paths = {
+        "order_ts_hour=2021-01-27-00/"
+        "00000-2-d5ae78b7-4449-45ec-adb7-c0e9c0bdb714-0-00001.parquet",
+        "order_ts_hour=2024-01-27-00/"
+        "00000-2-d5ae78b7-4449-45ec-adb7-c0e9c0bdb714-0-00002.parquet",
+        "order_ts_hour=2023-01-26-00/"
+        "00000-2-d5ae78b7-4449-45ec-adb7-c0e9c0bdb714-0-00003.parquet",
+        "order_ts_hour=2021-01-26-00/"
+        "00000-2-d5ae78b7-4449-45ec-adb7-c0e9c0bdb714-0-00004.parquet"};
+    std::vector<int64_t> partitions = {447696, 473976, 465192, 447672};
+    std::vector<std::map<int32_t, std::vector<uint8_t>>> bounds = {
+        {{1, {0xd2, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {2, {'.', 0x16, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {3, {0x12, 0xe2}},
+         {4, {0xc0, 'y', 0xe7, 0x98, 0xd6, 0xb9, 0x05, 0x00}}},
+        {{1, {0xd2, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {2, {'.', 0x16, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {3, {0x12, 0xe3}},
+         {4, {0xc0, 0x19, '#', '=', 0xe2, 0x0f, 0x06, 0x00}}},
+        {{1, {'{', 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {2, {0xc8, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {3, {0x0e, '"'}},
+         {4, {0xc0, 0xd9, '7', 0x93, 0x1f, 0xf3, 0x05, 0x00}}},
+        {{1, {'{', 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {2, {0xc8, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}},
+         {3, {0x0e, '!'}},
+         {4, {0xc0, 0x19, 0x10, '{', 0xc2, 0xb9, 0x05, 0x00}}},
+    };
+    for (int i = 0; i < 4; ++i) {
+      ManifestEntry entry;
+      entry.status = ManifestStatus::kAdded;
+      entry.snapshot_id = 6387266376565973956;
+      entry.data_file = std::make_shared<DataFile>();
+      entry.data_file->file_path = test_dir_prefix + paths[i];
+      entry.data_file->file_format = FileFormatType::kParquet;
+      entry.data_file->partition.emplace_back(Literal::Int(partitions[i]));
+      entry.data_file->record_count = 1;
+      entry.data_file->file_size_in_bytes = 1375;
+      entry.data_file->column_sizes = {{1, 49}, {2, 49}, {3, 49}, {4, 49}};
+      entry.data_file->value_counts = {{1, 1}, {2, 1}, {3, 1}, {4, 1}};
+      entry.data_file->null_value_counts = {{1, 0}, {2, 0}, {3, 0}, {4, 0}};
+      entry.data_file->split_offsets = {4};
+      entry.data_file->sort_order_id = 0;
+      entry.data_file->upper_bounds = bounds[i];
+      entry.data_file->lower_bounds = bounds[i];
+      manifest_entries.emplace_back(entry);
+    }
+    return manifest_entries;
+  }
+
+  std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs_;
+  std::shared_ptr<FileIO> file_io_;
+};
+
+TEST_F(ManifestReaderTest, BasicTest) {
+  iceberg::SchemaField partition_field(1000, "order_ts_hour", iceberg::int32(), true);
+  auto partition_schema =
+      std::make_shared<Schema>(std::vector<SchemaField>({partition_field}));
+  std::string path = GetResourcePath("56357cd7-391f-4df8-aa24-e7e667da8870-m4.avro");
+  auto manifest_reader_result =
+      ManifestReader::MakeReader(path, file_io_, partition_schema);
+  ASSERT_EQ(manifest_reader_result.has_value(), true)
+      << manifest_reader_result.error().message;
+  auto manifest_reader = std::move(manifest_reader_result.value());
+  auto read_result = manifest_reader->Entries();
+  ASSERT_EQ(read_result.has_value(), true) << read_result.error().message;
+
+  auto expected_entries = prepare_manifest_entries();
+  ASSERT_EQ(read_result.value(), expected_entries);
+}
+
+}  // namespace iceberg