feat: support schema.name-mapping.default

tea/common/iceberg_json.cpp

@@ -31,6 +31,7 @@ constexpr std::string_view kNameField = "name";
 constexpr std::string_view kTypeField = "type";
 constexpr std::string_view kFieldsField = "type";
 constexpr std::string_view kSchemaField = "schema";
+constexpr std::string_view kSchemaNameMappingField = "schema_name_mapping";
 constexpr std::string_view kScanMetadata = "scan_metadata";
 constexpr std::string_view kScanMetadataIdentifier = "scan_metadata_identifier";
 constexpr std::string_view kPartitionsField = "partitions";
@@ -204,6 +205,9 @@ rapidjson::Value Serializer::Serialize(iceberg::ice_tea::ScanMetadata&& scan_met
   rapidjson::Value result(rapidjson::kObjectType);
   AddMember(result, kSchemaField, std::move(*scan_metadata.schema));
   AddMember(result, kPartitionsField, std::move(scan_metadata.partitions));
+  if (scan_metadata.schema_name_mapping.has_value()) {
+    AddMember(result, kSchemaNameMappingField, std::move(*scan_metadata.schema_name_mapping));
+  }
   return result;
 }
 
@@ -375,7 +379,13 @@ iceberg::ice_tea::ScanMetadata Deserialize(const rapidjson::Value& value) {
 
   auto schema = std::make_shared<iceberg::Schema>(Extract<iceberg::Schema>(value, kSchemaField));
   auto partitions = Extract<std::vector<iceberg::ice_tea::ScanMetadata::Partition>>(value, kPartitionsField);
-  return iceberg::ice_tea::ScanMetadata{.schema = std::move(schema), .partitions = std::move(partitions)};
+  std::optional<std::string> schema_name_mapping;
+  if (value.HasMember(kSchemaNameMappingField.data())) {
+    schema_name_mapping = Extract<std::string>(value, kSchemaNameMappingField);
+  }
+  return iceberg::ice_tea::ScanMetadata{.schema = std::move(schema),
+                                        .partitions = std::move(partitions),
+                                        .schema_name_mapping = std::move(schema_name_mapping)};
 }
 
 template <>

tea/common/utils.cpp

@@ -914,6 +914,7 @@ iceberg::ice_tea::ScanMetadata SplitPartitionsAndFilter(iceberg::ice_tea::ScanMe
   const auto& partitions = scan_metadata.partitions;
   iceberg::ice_tea::ScanMetadata result;
   result.schema = scan_metadata.schema;
+  result.schema_name_mapping = scan_metadata.schema_name_mapping;
 
   int iterator = 0;
   for (const auto& partition : partitions) {

tea/gpext/tea_reader.cpp

@@ -971,6 +971,7 @@ std::shared_ptr<tea::samovar::SingleQueueClient> SamovarMakePlan(TeaContextPtr t
 
     {
       std::shared_ptr<iceberg::Schema> schema = tea::GetSchemaForSnapshot(table_metadata, config.snapshot_ref);
+      std::optional<std::string> schema_name_mapping = tea::meta::access::GetSchemaNameMappingDefault(*table_metadata);
       TEA_LOG("Samovar: getting manifest files");
 
       std::deque<iceberg::ManifestFile> manifest_files_queue =
@@ -987,8 +988,8 @@ std::shared_ptr<tea::samovar::SingleQueueClient> SamovarMakePlan(TeaContextPtr t
             CreateSamovarClient(tea_ctx, queue_name, query_scans_count_key, segment_id, segment_count,
                                 tea::samovar::SamovarRole::kFollower);
         TEA_LOG("Samovar: filling manifests queue");
-        auto maybe_stats = tea::samovar::FillSamovarWithManifests(get::Config(tea_ctx), schema, manifest_files_queue,
-                                                                  segment_count, samovar_client);
+        auto maybe_stats = tea::samovar::FillSamovarWithManifests(get::Config(tea_ctx), schema, schema_name_mapping,
+                                                                  manifest_files_queue, segment_count, samovar_client);
         get::PlannerStats(tea_ctx).Combine(iceberg::ValueSafe(maybe_stats));
 
         return samovar_client;
@@ -1024,6 +1025,7 @@ std::shared_ptr<tea::samovar::SingleQueueClient> SamovarMakePlan(TeaContextPtr t
 
         iceberg::ice_tea::ScanMetadata all_meta =
             iceberg::ValueSafe(iceberg::ice_tea::GetScanMetadata(*entries_stream, *table_metadata, schema, logger));
+        all_meta.schema_name_mapping = schema_name_mapping;
 
         ValidateAllMetadata(get::Config(tea_ctx), all_meta);
 

tea/metadata/access_iceberg.cpp

@@ -102,6 +102,18 @@ std::string GetIcebergTableLocation(const Config& config, TableId table_id) {
   return table->Location();
 }
 
+std::optional<std::string> GetMetadataProperty(const iceberg::TableMetadataV2& table_metadata,
+                                               const std::string& property_name) {
+  if (auto it = table_metadata.properties.find(property_name); it != table_metadata.properties.end()) {
+    return it->second;
+  }
+  return std::nullopt;
+}
+
+std::optional<std::string> GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata) {
+  return GetMetadataProperty(table_metadata, "schema.name-mapping.default");
+}
+
 namespace {
 class EmptyIcebergStream : public iceberg::ice_tea::IcebergEntriesStream {
  public:
@@ -177,7 +189,10 @@ std::pair<iceberg::ice_tea::ScanMetadata, PlannerStats> FromIcebergWithLocation(
       }
     }
 
-    return iceberg::ice_tea::GetScanMetadata(*entries_stream, *table_metadata, scan_schema, logger);
+    ARROW_ASSIGN_OR_RAISE(auto scan_metadata,
+                          iceberg::ice_tea::GetScanMetadata(*entries_stream, *table_metadata, scan_schema, logger));
+    scan_metadata.schema_name_mapping = GetSchemaNameMappingDefault(*table_metadata);
+    return scan_metadata;
   }();
 
   if (result.ok()) {

tea/metadata/access_iceberg.h

@@ -1,7 +1,9 @@
 #pragma once
 
 #include <memory>
+#include <optional>
 #include <string>
+#include <string_view>
 #include <utility>
 
 #include "iceberg/common/fs/filesystem_provider.h"
@@ -29,4 +31,9 @@ std::pair<iceberg::ice_tea::ScanMetadata, PlannerStats> FromIcebergWithLocation(
 
 std::string GetIcebergTableLocation(const Config& config, TableId table_id);
 
+std::optional<std::string> GetMetadataProperty(const iceberg::TableMetadataV2& table_metadata,
+                                               const std::string& property_name);
+
+std::optional<std::string> GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata);
+
 }  // namespace tea::meta::access

tea/metadata/planner.h

@@ -13,6 +13,7 @@ namespace tea::meta {
 
 class PlannedMeta {
  public:
+  // this field also contains schema and schema_name_mapping. TODO(gmusya): refactor and rename
   iceberg::ice_tea::ScanMetadata& GetDeletes() { return metadata_; }
   std::shared_ptr<AnnotatedDataEntryStream> GetStream() { return data_meta_stream_; }
 

tea/reader.cpp

@@ -578,7 +578,9 @@ class LowercaseRenamingStream : public iceberg::IcebergStream {
 };
 
 arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilter& filter, bool postfilter_on_gp) {
-  schema_ = meta.GetDeletes().schema;
+  const auto& deletes_with_schema = meta.GetDeletes();
+  schema_ = deletes_with_schema.schema;
+  auto schema_name_mapping = deletes_with_schema.schema_name_mapping;
   iceberg::Ensure(schema_ != nullptr, std::string(__PRETTY_FUNCTION__) + ": schema is nullptr");
 
   // empty table
@@ -603,8 +605,8 @@ arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilte
   auto equality_deletes = std::make_shared<iceberg::EqualityDeletes>(iceberg::EqualityDeletes{});
   iceberg::EqualityDeleteHandler::Config equality_delete_config = MakeIcebergEqualityDeleteConfig(config_);
 
-  for (size_t partition_id = 0; partition_id < meta.GetDeletes().partitions.size(); ++partition_id) {
-    auto& partition = meta.GetDeletes().partitions.at(partition_id);
+  for (size_t partition_id = 0; partition_id < deletes_with_schema.partitions.size(); ++partition_id) {
+    auto& partition = deletes_with_schema.partitions.at(partition_id);
     for (size_t layer_id = 0; layer_id < partition.size(); ++layer_id) {
       auto& layer = partition[layer_id];
       if (!layer.positional_delete_entries_.empty()) {
@@ -699,7 +701,8 @@ arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilte
 
   stream_ = iceberg::IcebergScanBuilder::MakeIcebergStream(
       meta_stream, std::move(positional_deletes), std::move(equality_deletes), std::move(equality_delete_config),
-      rg_filter, ice_filter, *schema_, std::move(field_ids_to_retrieve), file_reader_provider, std::nullopt, logger_);
+      rg_filter, ice_filter, *schema_, std::move(field_ids_to_retrieve), file_reader_provider, schema_name_mapping,
+      logger_);
 
   stream_ = std::make_shared<LowercaseRenamingStream>(stream_);
   return arrow::Status::OK();

tea/samovar/planner.cpp

@@ -76,13 +76,14 @@ std::shared_ptr<SingleQueueClient> MakeSamovarDataClient(const SamovarConfig& co
 }
 
 arrow::Result<PlannerStats> FillSamovarWithManifests(const Config& config, std::shared_ptr<iceberg::Schema> schema,
+                                                     std::optional<std::string> schema_name_mapping,
                                                      std::deque<iceberg::ManifestFile> manifests, int segment_count,
                                                      std::shared_ptr<SingleQueueClient> samovar_client) {
   PlannerStats stats;
   std::optional<ScopedTimerTicks> timer = ScopedTimerTicks(stats.plan_duration);
 
   samovar::ScanMetadata result;
-  *result.mutable_schema() = IcebergSchemaToTeapotSchema(schema);
+  AttachSchema(result, schema, schema_name_mapping);
 
   std::vector<samovar::ManifestList> samovar_manifests = ConvertToSamovarManifestLists(manifests);
 
@@ -398,6 +399,9 @@ arrow::Result<std::pair<meta::PlannedMeta, PlannerStats>> FromSamovar(
 
     iceberg::ice_tea::ScanMetadata metadata;
     metadata.schema = TeapotSchemaToIcebergSchema(response.schema());
+    if (response.has_schema_name_mapping()) {
+      metadata.schema_name_mapping = response.schema_name_mapping();
+    }
 
     // process tasks from the entries queue
     auto sched = std::make_shared<SamovarMetadataScheduler>(config, samovar_client);

tea/samovar/planner.h

@@ -5,6 +5,7 @@
 
 #include <deque>
 #include <memory>
+#include <optional>
 #include <string>
 #include <utility>
 
@@ -28,6 +29,7 @@ arrow::Result<PlannerStats> FillSamovar(const Config& config, iceberg::ice_tea::
                                         std::shared_ptr<SingleQueueClient> samovar_client);
 
 arrow::Result<PlannerStats> FillSamovarWithManifests(const Config& config, std::shared_ptr<iceberg::Schema> schema,
+                                                     std::optional<std::string> schema_name_mapping,
                                                      std::deque<iceberg::ManifestFile>, int segment_count,
                                                      std::shared_ptr<SingleQueueClient> samovar_client);
 

tea/samovar/proto/samovar.proto

@@ -55,6 +55,8 @@ message ScanMetadata {
 
   bool use_distributed_metadata_processing = 3;
   bool scan_already_finished = 4;
+
+  optional string schema_name_mapping = 5;
 }
 
 message AnnotatedDataEntry {

tea/samovar/utils.cpp

@@ -71,7 +71,8 @@ samovar::ScanMetadata ConvertIcebergScanMetaToSamovarRepresentation(iceberg::ice
   };
 
   samovar::ScanMetadata result;
-  *result.mutable_schema() = IcebergSchemaToTeapotSchema(scan_metadata.schema);
+  AttachSchema(result, scan_metadata.schema, scan_metadata.schema_name_mapping);
+
   for (const auto& partition : scan_metadata.partitions) {
     auto* samovar_partition = result.add_partitions();
     for (const auto& layer : partition) {
@@ -140,6 +141,9 @@ iceberg::ice_tea::ScanMetadata ConvertSamovarRepresentationToScanMeta(const samo
   }
 
   result.schema = TeapotSchemaToIcebergSchema(scan_metadata.schema());
+  if (scan_metadata.has_schema_name_mapping()) {
+    result.schema_name_mapping = scan_metadata.schema_name_mapping();
+  }
 
   return result;
 }
@@ -475,6 +479,9 @@ void SendManifestLists(const std::shared_ptr<ISamovarClient> client,
 samovar::ScanMetadata ClearDataEntries(const samovar::ScanMetadata& scan_metadata) {
   samovar::ScanMetadata result;
   *result.mutable_schema() = scan_metadata.schema();
+  if (scan_metadata.has_schema_name_mapping()) {
+    result.set_schema_name_mapping(scan_metadata.schema_name_mapping());
+  }
   result.set_use_distributed_metadata_processing(scan_metadata.use_distributed_metadata_processing());
   for (int i = 0; i < scan_metadata.partitions_size(); ++i) {
     const auto& partition = scan_metadata.partitions()[i];

tea/samovar/utils.h

@@ -11,6 +11,7 @@
 #include "iceberg/tea_scan.h"
 
 #include "tea/common/config.h"
+#include "tea/common/utils.h"
 #include "tea/metadata/metadata.h"
 #include "tea/samovar/network_layer/samovar_client.h"
 #include "tea/samovar/proto/samovar.pb.h"
@@ -54,6 +55,14 @@ void SendDataEntries(const std::shared_ptr<ISamovarClient> client,
                      const std::vector<samovar::AnnotatedDataEntry>& additional_data_entries,
                      const std::string& queue_id, std::chrono::seconds ttl_seconds, uint32_t batch_size);
 
+inline void AttachSchema(samovar::ScanMetadata& result, std::shared_ptr<iceberg::Schema> schema,
+                         const std::optional<std::string>& schema_name_mapping) {
+  *result.mutable_schema() = IcebergSchemaToTeapotSchema(schema);
+  if (schema_name_mapping.has_value()) {
+    result.set_schema_name_mapping(*schema_name_mapping);
+  }
+}
+
 samovar::ScanMetadata ClearDataEntries(const samovar::ScanMetadata& scan_metadata);
 
 }  // namespace tea::samovar

tea/smoke_test/CMakeLists.txt

@@ -63,6 +63,7 @@ set(smoke_test_sources
   real_parquet_test.cpp
   row_group_filter_test.cpp
   mapping_test.cpp
+  name_mapping_default_test.cpp
   self_join_test.cpp
   large_metadata_test.cpp
   mismatching_types_test.cpp