feat: add demo for registering table and planning files (#205)

Below is an output of the demo to read and plan files for a V1 table:
```
» ./demo_example /Users/gangwu/Projects/iceberg-warehouse table_v1_test /Users/gangwu/Projects/iceberg-warehouse/db/table_v1_test/metadata/v3.metadata.json
Scan tasks: 
 - /Users/gangwu/Projects/iceberg-warehouse/db/table_v1_test/data/00000-12-297cb609-fb4b-4e6c-a7ee-0a9bf62d356f-0-00001.parquet
 - /Users/gangwu/Projects/iceberg-warehouse/db/table_v1_test/data/00000-0-9052e5ed-641e-4f02-873f-5a87510b7451-0-00001.parquet
```

Author: wgtmac

example/demo_example.cc

@@ -19,18 +19,63 @@
 
 #include <iostream>
 
+#include "iceberg/arrow/arrow_file_io.h"
 #include "iceberg/avro/avro_register.h"
-#include "iceberg/file_reader.h"
+#include "iceberg/catalog/in_memory_catalog.h"
 #include "iceberg/parquet/parquet_register.h"
+#include "iceberg/table.h"
+#include "iceberg/table_scan.h"
+
+int main(int argc, char** argv) {
+  if (argc != 4) {
+    std::cerr << "Usage: " << argv[0]
+              << " <warehouse_location> <table_name> <table_location>" << std::endl;
+    return 0;
+  }
+
+  const std::string warehouse_location = argv[1];
+  const std::string table_name = argv[2];
+  const std::string table_location = argv[3];
+  const std::unordered_map<std::string, std::string> properties;
 
-int main() {
   iceberg::avro::RegisterAll();
   iceberg::parquet::RegisterAll();
-  auto open_result = iceberg::ReaderFactoryRegistry::Open(
-      iceberg::FileFormatType::kAvro, {.path = "non-existing-file.avro"});
-  if (!open_result.has_value()) {
-    std::cerr << "Failed to open avro file" << std::endl;
+
+  auto catalog = iceberg::InMemoryCatalog::Make("test", iceberg::arrow::MakeLocalFileIO(),
+                                                warehouse_location, properties);
+
+  auto register_result = catalog->RegisterTable({.name = table_name}, table_location);
+  if (!register_result.has_value()) {
+    std::cerr << "Failed to register table: " << register_result.error().message
+              << std::endl;
     return 1;
   }
+
+  auto load_result = catalog->LoadTable({.name = table_name});
+  if (!load_result.has_value()) {
+    std::cerr << "Failed to load table: " << load_result.error().message << std::endl;
+    return 1;
+  }
+
+  auto table = std::move(load_result.value());
+  auto scan_result = table->NewScan()->Build();
+  if (!scan_result.has_value()) {
+    std::cerr << "Failed to build scan: " << scan_result.error().message << std::endl;
+    return 1;
+  }
+
+  auto scan = std::move(scan_result.value());
+  auto plan_result = scan->PlanFiles();
+  if (!plan_result.has_value()) {
+    std::cerr << "Failed to plan files: " << plan_result.error().message << std::endl;
+    return 1;
+  }
+
+  std::cout << "Scan tasks: " << std::endl;
+  auto scan_tasks = std::move(plan_result.value());
+  for (const auto& scan_task : scan_tasks) {
+    std::cout << " - " << scan_task->data_file()->file_path << std::endl;
+  }
+
   return 0;
 }

src/iceberg/avro.h renamed to src/iceberg/arrow/arrow_file_io.h

@@ -19,16 +19,15 @@
 
 #pragma once
 
-#include <string>
+#include <memory>
 
-#include "iceberg/iceberg_export.h"
+#include "iceberg/file_io.h"
+#include "iceberg/iceberg_bundle_export.h"
 
-namespace iceberg {
+namespace iceberg::arrow {
 
-class ICEBERG_EXPORT Avro {
- public:
-  virtual ~Avro() = default;
-  virtual std::string print() const = 0;
-};
+ICEBERG_BUNDLE_EXPORT std::unique_ptr<FileIO> MakeMockFileIO();
 
-}  // namespace iceberg
+ICEBERG_BUNDLE_EXPORT std::unique_ptr<FileIO> MakeLocalFileIO();
+
+}  // namespace iceberg::arrow

src/iceberg/arrow/arrow_fs_file_io.cc

@@ -23,6 +23,7 @@
 #include <arrow/filesystem/mockfs.h>
 
 #include "iceberg/arrow/arrow_error_transform_internal.h"
+#include "iceberg/arrow/arrow_file_io.h"
 #include "iceberg/arrow/arrow_fs_file_io_internal.h"
 
 namespace iceberg::arrow {
@@ -80,4 +81,12 @@ std::unique_ptr<FileIO> ArrowFileSystemFileIO::MakeLocalFileIO() {
       std::make_shared<::arrow::fs::LocalFileSystem>());
 }
 
+std::unique_ptr<FileIO> MakeMockFileIO() {
+  return ArrowFileSystemFileIO::MakeMockFileIO();
+}
+
+std::unique_ptr<FileIO> MakeLocalFileIO() {
+  return ArrowFileSystemFileIO::MakeLocalFileIO();
+}
+
 }  // namespace iceberg::arrow

src/iceberg/avro/constants.h renamed to src/iceberg/avro/avro_constants.h

@@ -30,9 +30,9 @@
 #include <avro/Types.hh>
 #include <avro/ValidSchema.hh>
 
+#include "iceberg/avro/avro_constants.h"
 #include "iceberg/avro/avro_register.h"
 #include "iceberg/avro/avro_schema_util_internal.h"
-#include "iceberg/avro/constants.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/name_mapping.h"
 #include "iceberg/schema.h"

src/iceberg/avro/avro_schema_util.cc

@@ -314,6 +314,13 @@ Result<std::string> InMemoryNamespace::GetTableMetadataLocation(
   return it->second;
 }
 
+std::shared_ptr<InMemoryCatalog> InMemoryCatalog::Make(
+    std::string const& name, std::shared_ptr<FileIO> const& file_io,
+    std::string const& warehouse_location,
+    std::unordered_map<std::string, std::string> const& properties) {
+  return std::make_shared<InMemoryCatalog>(name, file_io, warehouse_location, properties);
+}
+
 InMemoryCatalog::InMemoryCatalog(
     std::string const& name, std::shared_ptr<FileIO> const& file_io,
     std::string const& warehouse_location,

src/iceberg/catalog/in_memory_catalog.cc

@@ -44,6 +44,11 @@ class ICEBERG_EXPORT InMemoryCatalog
                   std::unordered_map<std::string, std::string> const& properties);
   ~InMemoryCatalog() override;
 
+  static std::shared_ptr<InMemoryCatalog> Make(
+      std::string const& name, std::shared_ptr<FileIO> const& file_io,
+      std::string const& warehouse_location,
+      std::unordered_map<std::string, std::string> const& properties);
+
   std::string_view name() const override;
 
   Status CreateNamespace(

src/iceberg/catalog/in_memory_catalog.h

@@ -745,9 +745,9 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
   std::unordered_map<std::string, std::string> summary;
   if (summary_json.has_value()) {
     for (const auto& [key, value] : summary_json->items()) {
-      if (!kValidSnapshotSummaryFields.contains(key)) {
-        return JsonParseError("Invalid snapshot summary field: {}", key);
-      }
+      // if (!kValidSnapshotSummaryFields.contains(key)) {
+      //   return JsonParseError("Invalid snapshot summary field: {}", key);
+      // }
       if (!value.is_string()) {
         return JsonParseError("Invalid snapshot summary field value: {}",
                               SafeDumpJson(value));

src/iceberg/json_internal.cc

@@ -225,7 +225,9 @@ TEST(JsonInternalTest, Snapshot) {
   TestJsonConversion(snapshot, expected_json);
 }
 
-TEST(JsonInternalTest, SnapshotFromJsonWithInvalidSummary) {
+// FIXME: disable it for now since Iceberg Spark plugin generates
+// custom summary keys.
+TEST(JsonInternalTest, DISABLED_SnapshotFromJsonWithInvalidSummary) {
   nlohmann::json invalid_json =
       R"({"snapshot-id":1234567890,
           "parent-snapshot-id":9876543210,