From ec13570b0bc6c21e98faddd97e39848f394825af Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 11:41:41 +0300 Subject: [PATCH 1/9] feat: support schema.name-mapping.default --- tea/common/iceberg_json.cpp | 12 +- tea/common/utils.cpp | 1 + tea/gpext/tea_reader.cpp | 7 +- tea/metadata/access_iceberg.cpp | 13 +- tea/metadata/access_iceberg.h | 9 ++ tea/reader.cpp | 4 +- tea/reader.h | 2 + tea/samovar/planner.cpp | 7 + tea/samovar/planner.h | 2 + tea/samovar/proto/samovar.proto | 3 + tea/samovar/utils.cpp | 9 ++ tea/smoke_test/CMakeLists.txt | 1 + tea/smoke_test/name_mapping_default_test.cpp | 134 ++++++++++++++++++ tea/smoke_test/test_base.h | 13 ++ tea/test_utils/metadata.h | 25 ++++ vendor/CMakeLists.txt | 2 + ...01-scan-metadata-schema-name-mapping.patch | 16 +++ 17 files changed, 255 insertions(+), 5 deletions(-) create mode 100644 tea/smoke_test/name_mapping_default_test.cpp create mode 100644 vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch diff --git a/tea/common/iceberg_json.cpp b/tea/common/iceberg_json.cpp index e733db45..e4dc2b04 100644 --- a/tea/common/iceberg_json.cpp +++ b/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(Extract(value, kSchemaField)); auto partitions = Extract>(value, kPartitionsField); - return iceberg::ice_tea::ScanMetadata{.schema = std::move(schema), .partitions = std::move(partitions)}; + std::optional schema_name_mapping; + if (value.HasMember(kSchemaNameMappingField.data())) { + schema_name_mapping = Extract(value, kSchemaNameMappingField); + } + return iceberg::ice_tea::ScanMetadata{.schema = std::move(schema), + .partitions = std::move(partitions), + .schema_name_mapping = std::move(schema_name_mapping)}; } template <> diff --git a/tea/common/utils.cpp b/tea/common/utils.cpp index 28422be8..72047fc8 100644 --- a/tea/common/utils.cpp +++ b/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) { diff --git a/tea/gpext/tea_reader.cpp b/tea/gpext/tea_reader.cpp index acfbc327..0307407b 100644 --- a/tea/gpext/tea_reader.cpp +++ b/tea/gpext/tea_reader.cpp @@ -971,6 +971,8 @@ std::shared_ptr SamovarMakePlan(TeaContextPtr t { std::shared_ptr schema = tea::GetSchemaForSnapshot(table_metadata, config.snapshot_ref); + std::optional schema_name_mapping = + tea::meta::access::GetSchemaNameMappingDefault(*table_metadata); TEA_LOG("Samovar: getting manifest files"); std::deque manifest_files_queue = @@ -987,8 +989,8 @@ std::shared_ptr 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 +1026,7 @@ std::shared_ptr 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); diff --git a/tea/metadata/access_iceberg.cpp b/tea/metadata/access_iceberg.cpp index 792efede..a3e7b7cd 100644 --- a/tea/metadata/access_iceberg.cpp +++ b/tea/metadata/access_iceberg.cpp @@ -102,6 +102,14 @@ std::string GetIcebergTableLocation(const Config& config, TableId table_id) { return table->Location(); } +std::optional GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata) { + if (auto it = table_metadata.properties.find(std::string(kSchemaNameMappingDefaultProperty)); + it != table_metadata.properties.end()) { + return it->second; + } + return std::nullopt; +} + namespace { class EmptyIcebergStream : public iceberg::ice_tea::IcebergEntriesStream { public: @@ -177,7 +185,10 @@ std::pair 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()) { diff --git a/tea/metadata/access_iceberg.h b/tea/metadata/access_iceberg.h index 272daa72..0bfc0a42 100644 --- a/tea/metadata/access_iceberg.h +++ b/tea/metadata/access_iceberg.h @@ -1,7 +1,9 @@ #pragma once #include +#include #include +#include #include #include "iceberg/common/fs/filesystem_provider.h" @@ -14,6 +16,10 @@ namespace tea::meta::access { +// Iceberg table property holding the default JSON name mapping. See +// https://iceberg.apache.org/spec/#column-projection +inline constexpr std::string_view kSchemaNameMappingDefaultProperty = "schema.name-mapping.default"; + std::pair FromIceberg( const Config& config, TableId table_id, iceberg::filter::NodePtr filter, std::shared_ptr fs_provider, int64_t timestamp_to_timestamptz_shift_us, @@ -29,4 +35,7 @@ std::pair FromIcebergWithLocation( std::string GetIcebergTableLocation(const Config& config, TableId table_id); +// Returns the `schema.name-mapping.default` property value, or std::nullopt if absent. +std::optional GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata); + } // namespace tea::meta::access diff --git a/tea/reader.cpp b/tea/reader.cpp index 6a2fd644..eb3f4b97 100644 --- a/tea/reader.cpp +++ b/tea/reader.cpp @@ -579,6 +579,7 @@ class LowercaseRenamingStream : public iceberg::IcebergStream { arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilter& filter, bool postfilter_on_gp) { schema_ = meta.GetDeletes().schema; + schema_name_mapping_ = meta.GetDeletes().schema_name_mapping; iceberg::Ensure(schema_ != nullptr, std::string(__PRETTY_FUNCTION__) + ": schema is nullptr"); // empty table @@ -699,7 +700,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(stream_); return arrow::Status::OK(); diff --git a/tea/reader.h b/tea/reader.h index f92e4352..091e94cd 100644 --- a/tea/reader.h +++ b/tea/reader.h @@ -7,6 +7,7 @@ #include #include +#include #include #include #include @@ -98,6 +99,7 @@ class Reader { std::shared_ptr entries_stream_; std::shared_ptr schema_; + std::optional schema_name_mapping_; mutable bool at_least_one_matching_row_in_file_ = true; mutable bool potential_row_group_filter_logged_ = false; diff --git a/tea/samovar/planner.cpp b/tea/samovar/planner.cpp index ce5347ba..48b20d29 100644 --- a/tea/samovar/planner.cpp +++ b/tea/samovar/planner.cpp @@ -76,6 +76,7 @@ std::shared_ptr MakeSamovarDataClient(const SamovarConfig& co } arrow::Result FillSamovarWithManifests(const Config& config, std::shared_ptr schema, + std::optional schema_name_mapping, std::deque manifests, int segment_count, std::shared_ptr samovar_client) { PlannerStats stats; @@ -83,6 +84,9 @@ arrow::Result FillSamovarWithManifests(const Config& config, std:: samovar::ScanMetadata result; *result.mutable_schema() = IcebergSchemaToTeapotSchema(schema); + if (schema_name_mapping.has_value()) { + result.set_schema_name_mapping(*schema_name_mapping); + } std::vector samovar_manifests = ConvertToSamovarManifestLists(manifests); @@ -398,6 +402,9 @@ arrow::Result> 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(config, samovar_client); diff --git a/tea/samovar/planner.h b/tea/samovar/planner.h index 3aee48f9..ca440769 100644 --- a/tea/samovar/planner.h +++ b/tea/samovar/planner.h @@ -5,6 +5,7 @@ #include #include +#include #include #include @@ -28,6 +29,7 @@ arrow::Result FillSamovar(const Config& config, iceberg::ice_tea:: std::shared_ptr samovar_client); arrow::Result FillSamovarWithManifests(const Config& config, std::shared_ptr schema, + std::optional schema_name_mapping, std::deque, int segment_count, std::shared_ptr samovar_client); diff --git a/tea/samovar/proto/samovar.proto b/tea/samovar/proto/samovar.proto index 8388c7ff..1cf475b2 100644 --- a/tea/samovar/proto/samovar.proto +++ b/tea/samovar/proto/samovar.proto @@ -55,6 +55,9 @@ message ScanMetadata { bool use_distributed_metadata_processing = 3; bool scan_already_finished = 4; + + // JSON name mapping from the `schema.name-mapping.default` table property. + optional string schema_name_mapping = 5; } message AnnotatedDataEntry { diff --git a/tea/samovar/utils.cpp b/tea/samovar/utils.cpp index a1217e3e..76f80138 100644 --- a/tea/samovar/utils.cpp +++ b/tea/samovar/utils.cpp @@ -72,6 +72,9 @@ samovar::ScanMetadata ConvertIcebergScanMetaToSamovarRepresentation(iceberg::ice samovar::ScanMetadata result; *result.mutable_schema() = IcebergSchemaToTeapotSchema(scan_metadata.schema); + if (scan_metadata.schema_name_mapping.has_value()) { + result.set_schema_name_mapping(*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 +143,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 +481,9 @@ void SendManifestLists(const std::shared_ptr 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]; diff --git a/tea/smoke_test/CMakeLists.txt b/tea/smoke_test/CMakeLists.txt index f3dd4225..f916a4d1 100644 --- a/tea/smoke_test/CMakeLists.txt +++ b/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 diff --git a/tea/smoke_test/name_mapping_default_test.cpp b/tea/smoke_test/name_mapping_default_test.cpp new file mode 100644 index 00000000..41d7915c --- /dev/null +++ b/tea/smoke_test/name_mapping_default_test.cpp @@ -0,0 +1,134 @@ +// Tests for the Iceberg table property `schema.name-mapping.default`. +// +// When data files are added to a table without embedded Parquet field-ids +// (e.g. via add_files or a migrated table), Iceberg assigns field-ids to the +// physical columns by name using the JSON name mapping stored in the +// `schema.name-mapping.default` table property. +// +// These tests write data files with NO embedded field-ids and rely purely on +// the name mapping to resolve the Iceberg schema field-ids to Parquet columns. +// +// NOTE: this suite FAILS until Tea plumbs the property through to +// iceberg::IcebergScanBuilder::MakeIcebergStream (which currently receives +// std::nullopt for schema_name_mapping). See the implementation plan. + +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "iceberg/schema.h" +#include "iceberg/test_utils/assertions.h" +#include "iceberg/type.h" + +#include "tea/smoke_test/environment.h" +#include "tea/smoke_test/pq.h" +#include "tea/smoke_test/test_base.h" +#include "tea/test_utils/column.h" +#include "tea/test_utils/metadata.h" + +namespace tea { +namespace { + +// Parquet field-id sentinel meaning "no field-id embedded in the file". +constexpr int kNoFieldId = -1; + +class NameMappingDefaultTest : public TeaTest { + public: + void SetUp() override { + TeaTest::SetUp(); + // The name mapping is only meaningful for the Iceberg metadata path (it is + // carried to segments via the Samovar ScanMetadata message). Skip for the + // mock Teapot metadata path. + if (Environment::GetMetadataType() != MetadataType::kIceberg) { + GTEST_SKIP() << "schema.name-mapping.default only applies to the Iceberg metadata path"; + } + } + + protected: + static std::shared_ptr MakeIcebergSchema() { + using iceberg::types::NestedField; + using iceberg::types::PrimitiveType; + + std::vector fields; + fields.push_back(NestedField{.name = "col1", + .field_id = 1, + .is_required = false, + .type = std::make_shared(iceberg::TypeID::kString)}); + fields.push_back(NestedField{.name = "col2", + .field_id = 2, + .is_required = false, + .type = std::make_shared(iceberg::TypeID::kInt)}); + fields.push_back(NestedField{.name = "col3", + .field_id = 3, + .is_required = false, + .type = std::make_shared(iceberg::TypeID::kLong)}); + return std::make_shared(0, std::move(fields)); + } +}; + +// Data files have no field-ids; field-ids are resolved by matching the Parquet +// column names against the `names` in the name mapping. +TEST_F(NameMappingDefaultTest, ResolvesFieldIdsByName) { + std::string s0 = "a"; + std::string s1 = "b"; + + // field_id = kNoFieldId => the Parquet file carries no field-ids. + auto col1 = MakeStringColumn("col1", kNoFieldId, std::vector{&s0, &s1}); + auto col2 = MakeInt32Column("col2", kNoFieldId, OptionalVector{10, 20}); + auto col3 = MakeInt64Column("col3", kNoFieldId, OptionalVector{100, 200}); + + ASSIGN_OR_FAIL(auto data_path, state_->WriteFile({col1, col2, col3})); + ASSERT_OK(state_->AddDataFiles({data_path})); + + // The Iceberg schema field-ids come from table metadata, not the data file. + state_->SetSchema(MakeIcebergSchema()); + state_->SetProperties({{"schema.name-mapping.default", + R"([{"field-id":1,"names":["col1"]},)" + R"({"field-id":2,"names":["col2"]},)" + R"({"field-id":3,"names":["col3"]}])"}}); + + ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"}, + GreenplumColumnInfo{.name = "col2", .type = "int4"}, + GreenplumColumnInfo{.name = "col3", .type = "int8"}})); + + ASSIGN_OR_FAIL(auto result, pq::TableScanQuery(kDefaultTableName, "col1, col2, col3").Run(*conn_)); + + pq::ScanResult expected_result({"col1", "col2", "col3"}, {{"a", "10", "100"}, {"b", "20", "200"}}); + EXPECT_EQ(result, expected_result); +} + +// The name mapping can map a physical Parquet column name to an Iceberg column +// whose logical name differs (i.e. the column was renamed after the file was +// written). Resolution must follow the mapping, not the Iceberg column name. +TEST_F(NameMappingDefaultTest, PhysicalNameDiffersFromIcebergName) { + std::string s0 = "a"; + std::string s1 = "b"; + + // Physical Parquet names are old_*; Iceberg schema uses col*. + auto col1 = MakeStringColumn("old_col1", kNoFieldId, std::vector{&s0, &s1}); + auto col2 = MakeInt32Column("old_col2", kNoFieldId, OptionalVector{10, 20}); + auto col3 = MakeInt64Column("old_col3", kNoFieldId, OptionalVector{100, 200}); + + ASSIGN_OR_FAIL(auto data_path, state_->WriteFile({col1, col2, col3})); + ASSERT_OK(state_->AddDataFiles({data_path})); + + state_->SetSchema(MakeIcebergSchema()); + state_->SetProperties({{"schema.name-mapping.default", + R"([{"field-id":1,"names":["old_col1"]},)" + R"({"field-id":2,"names":["old_col2"]},)" + R"({"field-id":3,"names":["old_col3"]}])"}}); + + ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"}, + GreenplumColumnInfo{.name = "col2", .type = "int4"}, + GreenplumColumnInfo{.name = "col3", .type = "int8"}})); + + ASSIGN_OR_FAIL(auto result, pq::TableScanQuery(kDefaultTableName, "col1, col2, col3").Run(*conn_)); + + pq::ScanResult expected_result({"col1", "col2", "col3"}, {{"a", "10", "100"}, {"b", "20", "200"}}); + EXPECT_EQ(result, expected_result); +} + +} // namespace +} // namespace tea diff --git a/tea/smoke_test/test_base.h b/tea/smoke_test/test_base.h index 13df0de8..45c8a065 100644 --- a/tea/smoke_test/test_base.h +++ b/tea/smoke_test/test_base.h @@ -163,6 +163,19 @@ class TestState { return table_creator_->CreateTable(column_infos, table_name, location); } + // Override the Iceberg schema used in the table metadata (see IMetadataWriter::SetSchema). + void SetSchema(std::shared_ptr schema, const TableName& table_name = kDefaultTableName) { + BuildMetadataWriterIfNecessary(table_name); + metadata_writer_.at(table_name)->SetSchema(std::move(schema)); + } + + // Set Iceberg table properties written into the table metadata (see IMetadataWriter::SetProperties). + void SetProperties(std::map properties, + const TableName& table_name = kDefaultTableName) { + BuildMetadataWriterIfNecessary(table_name); + metadata_writer_.at(table_name)->SetProperties(std::move(properties)); + } + void SetFileWriter(std::shared_ptr file_writer) { file_writer_ = file_writer; } void SetMetadataWriterBuilder(std::shared_ptr metadata_writer_builder) { metadata_writer_.clear(); diff --git a/tea/test_utils/metadata.h b/tea/test_utils/metadata.h index 1e0f9861..570ce732 100644 --- a/tea/test_utils/metadata.h +++ b/tea/test_utils/metadata.h @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -43,6 +44,16 @@ class IMetadataWriter { const std::vector& field_ids) = 0; virtual arrow::Result Finalize() = 0; + // Override the Iceberg table schema instead of deriving it from the field-ids + // embedded in the written data files. Needed to model tables whose data files + // have no embedded field-ids (e.g. add_files / migrated tables), where the + // schema field-ids come from the table metadata, not the Parquet file. + virtual void SetSchema(std::shared_ptr /*schema*/) {} + + // Set Iceberg table properties (e.g. schema.name-mapping.default) to be + // written into the table metadata. + virtual void SetProperties(std::map /*properties*/) {} + virtual ~IMetadataWriter() = default; }; @@ -297,7 +308,17 @@ class IcebergMetadataWriter : public IMetadataWriter { " for column " + column->name()); } + void SetSchema(std::shared_ptr schema) override { forced_schema_ = std::move(schema); } + + void SetProperties(std::map properties) override { + table_properties_ = std::move(properties); + } + arrow::Result> GetSchema() const { + if (forced_schema_) { + return forced_schema_; + } + if (!some_data_path_.has_value()) { return arrow::Status::ExecutionError("No data file to extract schema"); } @@ -379,6 +400,7 @@ class IcebergMetadataWriter : public IMetadataWriter { std::make_shared(iceberg::SortOrder{.order_id = 0, .fields = {}})); builder.default_sort_order_id = 0; builder.partition_specs = {std::make_shared(partition_spec_)}; + builder.properties = table_properties_; std::shared_ptr meta = builder.Build(); { @@ -415,6 +437,9 @@ class IcebergMetadataWriter : public IMetadataWriter { const std::string profile_; const iceberg::PartitionSpec partition_spec_; + std::shared_ptr forced_schema_; + std::map table_properties_; + uint64_t data_files_count_ = 0; uint64_t delete_files_count_ = 0; }; diff --git a/vendor/CMakeLists.txt b/vendor/CMakeLists.txt index ef2c3839..2101c3d0 100644 --- a/vendor/CMakeLists.txt +++ b/vendor/CMakeLists.txt @@ -31,6 +31,8 @@ FetchContent_Declare( EXCLUDE_FROM_ALL GIT_REPOSITORY ${GITHUB}/lithium-tech/iceberg-cxx.git GIT_TAG 3e148ac96d07e9bd3a07dbe9a6ae2efdc6559f2a + PATCH_COMMAND git reset --hard HEAD + && git apply ${CMAKE_CURRENT_SOURCE_DIR}/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch ) FetchContent_MakeAvailable(googletest hiredis) diff --git a/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch new file mode 100644 index 00000000..675b8a67 --- /dev/null +++ b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch @@ -0,0 +1,16 @@ +diff --git a/iceberg/tea_scan.h b/iceberg/tea_scan.h +index 87b0e71..6b658ca 100644 +--- a/iceberg/tea_scan.h ++++ b/iceberg/tea_scan.h +@@ -98,6 +98,11 @@ struct ScanMetadata { + std::shared_ptr schema; + std::vector partitions; + ++ // JSON name mapping from the `schema.name-mapping.default` table property. Used ++ // to assign field-ids to data files that have no embedded field-ids (e.g. files ++ // added via add_files or migrated tables). Empty when the property is absent. ++ std::optional schema_name_mapping; ++ + bool operator==(const ScanMetadata& scan_meta) const = default; + }; + From 927a46b569ae371bdb499ddc08a252b8130da351 Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 12:06:04 +0300 Subject: [PATCH 2/9] clang-format --- tea/gpext/tea_reader.cpp | 3 +-- tea/smoke_test/name_mapping_default_test.cpp | 14 ++++++-------- tea/smoke_test/test_base.h | 3 +-- 3 files changed, 8 insertions(+), 12 deletions(-) diff --git a/tea/gpext/tea_reader.cpp b/tea/gpext/tea_reader.cpp index 0307407b..790fd73e 100644 --- a/tea/gpext/tea_reader.cpp +++ b/tea/gpext/tea_reader.cpp @@ -971,8 +971,7 @@ std::shared_ptr SamovarMakePlan(TeaContextPtr t { std::shared_ptr schema = tea::GetSchemaForSnapshot(table_metadata, config.snapshot_ref); - std::optional schema_name_mapping = - tea::meta::access::GetSchemaNameMappingDefault(*table_metadata); + std::optional schema_name_mapping = tea::meta::access::GetSchemaNameMappingDefault(*table_metadata); TEA_LOG("Samovar: getting manifest files"); std::deque manifest_files_queue = diff --git a/tea/smoke_test/name_mapping_default_test.cpp b/tea/smoke_test/name_mapping_default_test.cpp index 41d7915c..cf37ce31 100644 --- a/tea/smoke_test/name_mapping_default_test.cpp +++ b/tea/smoke_test/name_mapping_default_test.cpp @@ -84,10 +84,9 @@ TEST_F(NameMappingDefaultTest, ResolvesFieldIdsByName) { // The Iceberg schema field-ids come from table metadata, not the data file. state_->SetSchema(MakeIcebergSchema()); - state_->SetProperties({{"schema.name-mapping.default", - R"([{"field-id":1,"names":["col1"]},)" - R"({"field-id":2,"names":["col2"]},)" - R"({"field-id":3,"names":["col3"]}])"}}); + state_->SetProperties({{"schema.name-mapping.default", R"([{"field-id":1,"names":["col1"]},)" + R"({"field-id":2,"names":["col2"]},)" + R"({"field-id":3,"names":["col3"]}])"}}); ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"}, GreenplumColumnInfo{.name = "col2", .type = "int4"}, @@ -115,10 +114,9 @@ TEST_F(NameMappingDefaultTest, PhysicalNameDiffersFromIcebergName) { ASSERT_OK(state_->AddDataFiles({data_path})); state_->SetSchema(MakeIcebergSchema()); - state_->SetProperties({{"schema.name-mapping.default", - R"([{"field-id":1,"names":["old_col1"]},)" - R"({"field-id":2,"names":["old_col2"]},)" - R"({"field-id":3,"names":["old_col3"]}])"}}); + state_->SetProperties({{"schema.name-mapping.default", R"([{"field-id":1,"names":["old_col1"]},)" + R"({"field-id":2,"names":["old_col2"]},)" + R"({"field-id":3,"names":["old_col3"]}])"}}); ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"}, GreenplumColumnInfo{.name = "col2", .type = "int4"}, diff --git a/tea/smoke_test/test_base.h b/tea/smoke_test/test_base.h index 45c8a065..1eb7211c 100644 --- a/tea/smoke_test/test_base.h +++ b/tea/smoke_test/test_base.h @@ -170,8 +170,7 @@ class TestState { } // Set Iceberg table properties written into the table metadata (see IMetadataWriter::SetProperties). - void SetProperties(std::map properties, - const TableName& table_name = kDefaultTableName) { + void SetProperties(std::map properties, const TableName& table_name = kDefaultTableName) { BuildMetadataWriterIfNecessary(table_name); metadata_writer_.at(table_name)->SetProperties(std::move(properties)); } From 427280628ebec8e7577f9c2ba7a7eeb04d21dc46 Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 14:31:55 +0300 Subject: [PATCH 3/9] refactor --- tea/metadata/access_iceberg.cpp | 3 +-- tea/metadata/access_iceberg.h | 5 ----- tea/metadata/planner.h | 1 + tea/reader.cpp | 4 ++-- tea/reader.h | 2 -- tea/samovar/proto/samovar.proto | 1 - tea/smoke_test/test_base.h | 11 +++++++++-- tea/test_utils/metadata.h | 11 ++--------- 8 files changed, 15 insertions(+), 23 deletions(-) diff --git a/tea/metadata/access_iceberg.cpp b/tea/metadata/access_iceberg.cpp index a3e7b7cd..06388f99 100644 --- a/tea/metadata/access_iceberg.cpp +++ b/tea/metadata/access_iceberg.cpp @@ -103,8 +103,7 @@ std::string GetIcebergTableLocation(const Config& config, TableId table_id) { } std::optional GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata) { - if (auto it = table_metadata.properties.find(std::string(kSchemaNameMappingDefaultProperty)); - it != table_metadata.properties.end()) { + if (auto it = table_metadata.properties.find("schema.name-mapping.default"); it != table_metadata.properties.end()) { return it->second; } return std::nullopt; diff --git a/tea/metadata/access_iceberg.h b/tea/metadata/access_iceberg.h index 0bfc0a42..2bb6a2fa 100644 --- a/tea/metadata/access_iceberg.h +++ b/tea/metadata/access_iceberg.h @@ -16,10 +16,6 @@ namespace tea::meta::access { -// Iceberg table property holding the default JSON name mapping. See -// https://iceberg.apache.org/spec/#column-projection -inline constexpr std::string_view kSchemaNameMappingDefaultProperty = "schema.name-mapping.default"; - std::pair FromIceberg( const Config& config, TableId table_id, iceberg::filter::NodePtr filter, std::shared_ptr fs_provider, int64_t timestamp_to_timestamptz_shift_us, @@ -35,7 +31,6 @@ std::pair FromIcebergWithLocation( std::string GetIcebergTableLocation(const Config& config, TableId table_id); -// Returns the `schema.name-mapping.default` property value, or std::nullopt if absent. std::optional GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata); } // namespace tea::meta::access diff --git a/tea/metadata/planner.h b/tea/metadata/planner.h index 7557c1eb..c086c9c8 100644 --- a/tea/metadata/planner.h +++ b/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 GetStream() { return data_meta_stream_; } diff --git a/tea/reader.cpp b/tea/reader.cpp index eb3f4b97..4faac1c8 100644 --- a/tea/reader.cpp +++ b/tea/reader.cpp @@ -579,7 +579,7 @@ class LowercaseRenamingStream : public iceberg::IcebergStream { arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilter& filter, bool postfilter_on_gp) { schema_ = meta.GetDeletes().schema; - schema_name_mapping_ = meta.GetDeletes().schema_name_mapping; + auto schema_name_mapping = meta.GetDeletes().schema_name_mapping; iceberg::Ensure(schema_ != nullptr, std::string(__PRETTY_FUNCTION__) + ": schema is nullptr"); // empty table @@ -700,7 +700,7 @@ 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, schema_name_mapping_, + rg_filter, ice_filter, *schema_, std::move(field_ids_to_retrieve), file_reader_provider, schema_name_mapping, logger_); stream_ = std::make_shared(stream_); diff --git a/tea/reader.h b/tea/reader.h index 091e94cd..f92e4352 100644 --- a/tea/reader.h +++ b/tea/reader.h @@ -7,7 +7,6 @@ #include #include -#include #include #include #include @@ -99,7 +98,6 @@ class Reader { std::shared_ptr entries_stream_; std::shared_ptr schema_; - std::optional schema_name_mapping_; mutable bool at_least_one_matching_row_in_file_ = true; mutable bool potential_row_group_filter_logged_ = false; diff --git a/tea/samovar/proto/samovar.proto b/tea/samovar/proto/samovar.proto index 1cf475b2..4ad29e99 100644 --- a/tea/samovar/proto/samovar.proto +++ b/tea/samovar/proto/samovar.proto @@ -56,7 +56,6 @@ message ScanMetadata { bool use_distributed_metadata_processing = 3; bool scan_already_finished = 4; - // JSON name mapping from the `schema.name-mapping.default` table property. optional string schema_name_mapping = 5; } diff --git a/tea/smoke_test/test_base.h b/tea/smoke_test/test_base.h index 1eb7211c..9ab4e06f 100644 --- a/tea/smoke_test/test_base.h +++ b/tea/smoke_test/test_base.h @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -78,6 +79,14 @@ class TeapotMetadataWriter : public IMetadataWriter { Options{.profile = Environment::GetProfile()})); } + void SetSchema(std::shared_ptr) override { + throw std::runtime_error("Internal error in test. SetSchema is not supported for TeapotMetadataWriter"); + } + + void SetProperties(std::map) override { + throw std::runtime_error("Internal error in test. SetSchema is not supported for TeapotMetadataWriter"); + } + private: const TableName table_name_; std::vector fragments_; @@ -163,13 +172,11 @@ class TestState { return table_creator_->CreateTable(column_infos, table_name, location); } - // Override the Iceberg schema used in the table metadata (see IMetadataWriter::SetSchema). void SetSchema(std::shared_ptr schema, const TableName& table_name = kDefaultTableName) { BuildMetadataWriterIfNecessary(table_name); metadata_writer_.at(table_name)->SetSchema(std::move(schema)); } - // Set Iceberg table properties written into the table metadata (see IMetadataWriter::SetProperties). void SetProperties(std::map properties, const TableName& table_name = kDefaultTableName) { BuildMetadataWriterIfNecessary(table_name); metadata_writer_.at(table_name)->SetProperties(std::move(properties)); diff --git a/tea/test_utils/metadata.h b/tea/test_utils/metadata.h index 570ce732..c2500eb5 100644 --- a/tea/test_utils/metadata.h +++ b/tea/test_utils/metadata.h @@ -44,15 +44,8 @@ class IMetadataWriter { const std::vector& field_ids) = 0; virtual arrow::Result Finalize() = 0; - // Override the Iceberg table schema instead of deriving it from the field-ids - // embedded in the written data files. Needed to model tables whose data files - // have no embedded field-ids (e.g. add_files / migrated tables), where the - // schema field-ids come from the table metadata, not the Parquet file. - virtual void SetSchema(std::shared_ptr /*schema*/) {} - - // Set Iceberg table properties (e.g. schema.name-mapping.default) to be - // written into the table metadata. - virtual void SetProperties(std::map /*properties*/) {} + virtual void SetSchema(std::shared_ptr) = 0; + virtual void SetProperties(std::map) = 0; virtual ~IMetadataWriter() = default; }; From d4823667f00cbbb40fb192f69298fbe43e6f5c77 Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 14:33:54 +0300 Subject: [PATCH 4/9] Add todo --- vendor/CMakeLists.txt | 1 + .../iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch | 3 --- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/vendor/CMakeLists.txt b/vendor/CMakeLists.txt index 2101c3d0..e47283e6 100644 --- a/vendor/CMakeLists.txt +++ b/vendor/CMakeLists.txt @@ -34,6 +34,7 @@ FetchContent_Declare( PATCH_COMMAND git reset --hard HEAD && git apply ${CMAKE_CURRENT_SOURCE_DIR}/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch ) +# TODO(gmusya): support schema.name-mapping.default in GetScanMetadata and add this patch to iceberg-cxx FetchContent_MakeAvailable(googletest hiredis) diff --git a/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch index 675b8a67..eb39a457 100644 --- a/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch +++ b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch @@ -6,9 +6,6 @@ index 87b0e71..6b658ca 100644 std::shared_ptr schema; std::vector partitions; -+ // JSON name mapping from the `schema.name-mapping.default` table property. Used -+ // to assign field-ids to data files that have no embedded field-ids (e.g. files -+ // added via add_files or migrated tables). Empty when the property is absent. + std::optional schema_name_mapping; + bool operator==(const ScanMetadata& scan_meta) const = default; From 3dde571c4040187e273612a5fefab05faccc49cd Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 14:47:36 +0300 Subject: [PATCH 5/9] Update --- tea/smoke_test/name_mapping_default_test.cpp | 53 +++++++++---------- ...01-scan-metadata-schema-name-mapping.patch | 2 +- 2 files changed, 26 insertions(+), 29 deletions(-) diff --git a/tea/smoke_test/name_mapping_default_test.cpp b/tea/smoke_test/name_mapping_default_test.cpp index cf37ce31..7cd98f0f 100644 --- a/tea/smoke_test/name_mapping_default_test.cpp +++ b/tea/smoke_test/name_mapping_default_test.cpp @@ -1,17 +1,3 @@ -// Tests for the Iceberg table property `schema.name-mapping.default`. -// -// When data files are added to a table without embedded Parquet field-ids -// (e.g. via add_files or a migrated table), Iceberg assigns field-ids to the -// physical columns by name using the JSON name mapping stored in the -// `schema.name-mapping.default` table property. -// -// These tests write data files with NO embedded field-ids and rely purely on -// the name mapping to resolve the Iceberg schema field-ids to Parquet columns. -// -// NOTE: this suite FAILS until Tea plumbs the property through to -// iceberg::IcebergScanBuilder::MakeIcebergStream (which currently receives -// std::nullopt for schema_name_mapping). See the implementation plan. - #include #include #include @@ -31,19 +17,15 @@ namespace tea { namespace { -// Parquet field-id sentinel meaning "no field-id embedded in the file". constexpr int kNoFieldId = -1; class NameMappingDefaultTest : public TeaTest { public: void SetUp() override { - TeaTest::SetUp(); - // The name mapping is only meaningful for the Iceberg metadata path (it is - // carried to segments via the Samovar ScanMetadata message). Skip for the - // mock Teapot metadata path. if (Environment::GetMetadataType() != MetadataType::kIceberg) { GTEST_SKIP() << "schema.name-mapping.default only applies to the Iceberg metadata path"; } + TeaTest::SetUp(); } protected: @@ -68,13 +50,10 @@ class NameMappingDefaultTest : public TeaTest { } }; -// Data files have no field-ids; field-ids are resolved by matching the Parquet -// column names against the `names` in the name mapping. TEST_F(NameMappingDefaultTest, ResolvesFieldIdsByName) { std::string s0 = "a"; std::string s1 = "b"; - // field_id = kNoFieldId => the Parquet file carries no field-ids. auto col1 = MakeStringColumn("col1", kNoFieldId, std::vector{&s0, &s1}); auto col2 = MakeInt32Column("col2", kNoFieldId, OptionalVector{10, 20}); auto col3 = MakeInt64Column("col3", kNoFieldId, OptionalVector{100, 200}); @@ -82,7 +61,6 @@ TEST_F(NameMappingDefaultTest, ResolvesFieldIdsByName) { ASSIGN_OR_FAIL(auto data_path, state_->WriteFile({col1, col2, col3})); ASSERT_OK(state_->AddDataFiles({data_path})); - // The Iceberg schema field-ids come from table metadata, not the data file. state_->SetSchema(MakeIcebergSchema()); state_->SetProperties({{"schema.name-mapping.default", R"([{"field-id":1,"names":["col1"]},)" R"({"field-id":2,"names":["col2"]},)" @@ -98,14 +76,10 @@ TEST_F(NameMappingDefaultTest, ResolvesFieldIdsByName) { EXPECT_EQ(result, expected_result); } -// The name mapping can map a physical Parquet column name to an Iceberg column -// whose logical name differs (i.e. the column was renamed after the file was -// written). Resolution must follow the mapping, not the Iceberg column name. -TEST_F(NameMappingDefaultTest, PhysicalNameDiffersFromIcebergName) { +TEST_F(NameMappingDefaultTest, ParquetNameDiffersFromIcebergName) { std::string s0 = "a"; std::string s1 = "b"; - // Physical Parquet names are old_*; Iceberg schema uses col*. auto col1 = MakeStringColumn("old_col1", kNoFieldId, std::vector{&s0, &s1}); auto col2 = MakeInt32Column("old_col2", kNoFieldId, OptionalVector{10, 20}); auto col3 = MakeInt64Column("old_col3", kNoFieldId, OptionalVector{100, 200}); @@ -128,5 +102,28 @@ TEST_F(NameMappingDefaultTest, PhysicalNameDiffersFromIcebergName) { EXPECT_EQ(result, expected_result); } +TEST_F(NameMappingDefaultTest, NoFieldIdNoSchemaNameMapping) { + std::string s0 = "a"; + std::string s1 = "b"; + + auto col1 = MakeStringColumn("col1", kNoFieldId, std::vector{&s0, &s1}); + auto col2 = MakeInt32Column("col2", kNoFieldId, OptionalVector{10, 20}); + auto col3 = MakeInt64Column("col3", kNoFieldId, OptionalVector{100, 200}); + + ASSIGN_OR_FAIL(auto data_path, state_->WriteFile({col1, col2, col3})); + ASSERT_OK(state_->AddDataFiles({data_path})); + + state_->SetSchema(MakeIcebergSchema()); + + ASSIGN_OR_FAIL(auto defer, state_->CreateTable({GreenplumColumnInfo{.name = "col1", .type = "text"}, + GreenplumColumnInfo{.name = "col2", .type = "int4"}, + GreenplumColumnInfo{.name = "col3", .type = "int8"}})); + + ASSIGN_OR_FAIL(auto result, pq::TableScanQuery(kDefaultTableName, "col1, col2, col3").Run(*conn_)); + + pq::ScanResult expected_result({"col1", "col2", "col3"}, {{"", "", ""}, {"", "", ""}}); + EXPECT_EQ(result, expected_result); +} + } // namespace } // namespace tea diff --git a/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch index eb39a457..d9305226 100644 --- a/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch +++ b/vendor/iceberg-cxx/0001-scan-metadata-schema-name-mapping.patch @@ -2,7 +2,7 @@ diff --git a/iceberg/tea_scan.h b/iceberg/tea_scan.h index 87b0e71..6b658ca 100644 --- a/iceberg/tea_scan.h +++ b/iceberg/tea_scan.h -@@ -98,6 +98,11 @@ struct ScanMetadata { +@@ -98,6 +98,8 @@ struct ScanMetadata { std::shared_ptr schema; std::vector partitions; From 9fb03a05855db44d8b865cea78a5461a9ba517bd Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 15:46:24 +0300 Subject: [PATCH 6/9] improve common GetDeletes --- tea/reader.cpp | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tea/reader.cpp b/tea/reader.cpp index 4faac1c8..236dade6 100644 --- a/tea/reader.cpp +++ b/tea/reader.cpp @@ -578,8 +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; - auto schema_name_mapping = meta.GetDeletes().schema_name_mapping; + 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 @@ -604,8 +605,8 @@ arrow::Status Reader::Plan(meta::PlannedMeta meta, const Reader::SerializedFilte auto equality_deletes = std::make_shared(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()) { From a3c9d7adc2030966af19b100c062c2c42563756c Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 15:46:57 +0300 Subject: [PATCH 7/9] Add AttachSchema --- tea/samovar/planner.cpp | 5 +---- tea/samovar/utils.cpp | 6 ++---- tea/samovar/utils.h | 9 +++++++++ 3 files changed, 12 insertions(+), 8 deletions(-) diff --git a/tea/samovar/planner.cpp b/tea/samovar/planner.cpp index 48b20d29..93184717 100644 --- a/tea/samovar/planner.cpp +++ b/tea/samovar/planner.cpp @@ -83,10 +83,7 @@ arrow::Result FillSamovarWithManifests(const Config& config, std:: std::optional timer = ScopedTimerTicks(stats.plan_duration); samovar::ScanMetadata result; - *result.mutable_schema() = IcebergSchemaToTeapotSchema(schema); - if (schema_name_mapping.has_value()) { - result.set_schema_name_mapping(*schema_name_mapping); - } + AttachSchema(result, schema, schema_name_mapping); std::vector samovar_manifests = ConvertToSamovarManifestLists(manifests); diff --git a/tea/samovar/utils.cpp b/tea/samovar/utils.cpp index 76f80138..777ce4ae 100644 --- a/tea/samovar/utils.cpp +++ b/tea/samovar/utils.cpp @@ -71,10 +71,8 @@ samovar::ScanMetadata ConvertIcebergScanMetaToSamovarRepresentation(iceberg::ice }; samovar::ScanMetadata result; - *result.mutable_schema() = IcebergSchemaToTeapotSchema(scan_metadata.schema); - if (scan_metadata.schema_name_mapping.has_value()) { - result.set_schema_name_mapping(*scan_metadata.schema_name_mapping); - } + 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) { diff --git a/tea/samovar/utils.h b/tea/samovar/utils.h index 9f923356..26f707d1 100644 --- a/tea/samovar/utils.h +++ b/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 client, const std::vector& 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 schema, + const std::optional& 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 From 4633158270cabf8d797675a8ad4a647b1c278fa3 Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 15:48:22 +0300 Subject: [PATCH 8/9] Refactor GetSchemaNameMappingDefault --- tea/metadata/access_iceberg.cpp | 9 +++++++-- tea/metadata/access_iceberg.h | 3 +++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/tea/metadata/access_iceberg.cpp b/tea/metadata/access_iceberg.cpp index 06388f99..872ff974 100644 --- a/tea/metadata/access_iceberg.cpp +++ b/tea/metadata/access_iceberg.cpp @@ -102,13 +102,18 @@ std::string GetIcebergTableLocation(const Config& config, TableId table_id) { return table->Location(); } -std::optional GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata) { - if (auto it = table_metadata.properties.find("schema.name-mapping.default"); it != table_metadata.properties.end()) { +std::optional 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 GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata) { + return GetMetadataProperty(table_metadata, "schema.name-mapping.default"); +} + namespace { class EmptyIcebergStream : public iceberg::ice_tea::IcebergEntriesStream { public: diff --git a/tea/metadata/access_iceberg.h b/tea/metadata/access_iceberg.h index 2bb6a2fa..66a8621c 100644 --- a/tea/metadata/access_iceberg.h +++ b/tea/metadata/access_iceberg.h @@ -31,6 +31,9 @@ std::pair FromIcebergWithLocation( std::string GetIcebergTableLocation(const Config& config, TableId table_id); +std::optional GetMetadataProperty(const iceberg::TableMetadataV2& table_metadata, + const std::string& property_name); + std::optional GetSchemaNameMappingDefault(const iceberg::TableMetadataV2& table_metadata); } // namespace tea::meta::access From c4c8ba905493a62f8a8785c081a0f3acbd4189c1 Mon Sep 17 00:00:00 2001 From: German Perov Date: Tue, 7 Jul 2026 16:02:42 +0300 Subject: [PATCH 9/9] Add symmetric AttachSchema --- tea/samovar/planner.cpp | 7 ++----- tea/samovar/utils.cpp | 5 +---- tea/samovar/utils.h | 7 +++++++ 3 files changed, 10 insertions(+), 9 deletions(-) diff --git a/tea/samovar/planner.cpp b/tea/samovar/planner.cpp index 93184717..7fe8f0bc 100644 --- a/tea/samovar/planner.cpp +++ b/tea/samovar/planner.cpp @@ -362,7 +362,7 @@ arrow::Result> FromSamovar( auto response = samovar_client->GetPlannedMetadata(); if (response.scan_already_finished()) { iceberg::ice_tea::ScanMetadata metadata; - metadata.schema = TeapotSchemaToIcebergSchema(response.schema()); + AttachSchema(metadata, response); auto sched = std::make_shared(); @@ -398,10 +398,7 @@ arrow::Result> FromSamovar( samovar_client->WaitForManifestsQueue(); iceberg::ice_tea::ScanMetadata metadata; - metadata.schema = TeapotSchemaToIcebergSchema(response.schema()); - if (response.has_schema_name_mapping()) { - metadata.schema_name_mapping = response.schema_name_mapping(); - } + AttachSchema(metadata, response); // process tasks from the entries queue auto sched = std::make_shared(config, samovar_client); diff --git a/tea/samovar/utils.cpp b/tea/samovar/utils.cpp index 777ce4ae..5417e77a 100644 --- a/tea/samovar/utils.cpp +++ b/tea/samovar/utils.cpp @@ -140,10 +140,7 @@ iceberg::ice_tea::ScanMetadata ConvertSamovarRepresentationToScanMeta(const samo result.partitions.push_back(std::move(ice_partition)); } - result.schema = TeapotSchemaToIcebergSchema(scan_metadata.schema()); - if (scan_metadata.has_schema_name_mapping()) { - result.schema_name_mapping = scan_metadata.schema_name_mapping(); - } + AttachSchema(result, scan_metadata); return result; } diff --git a/tea/samovar/utils.h b/tea/samovar/utils.h index 26f707d1..dddf073f 100644 --- a/tea/samovar/utils.h +++ b/tea/samovar/utils.h @@ -55,6 +55,13 @@ void SendDataEntries(const std::shared_ptr client, const std::vector& additional_data_entries, const std::string& queue_id, std::chrono::seconds ttl_seconds, uint32_t batch_size); +inline void AttachSchema(iceberg::ice_tea::ScanMetadata& result, const samovar::ScanMetadata& source) { + result.schema = TeapotSchemaToIcebergSchema(source.schema()); + if (source.has_schema_name_mapping()) { + result.schema_name_mapping = source.schema_name_mapping(); + } +} + inline void AttachSchema(samovar::ScanMetadata& result, std::shared_ptr schema, const std::optional& schema_name_mapping) { *result.mutable_schema() = IcebergSchemaToTeapotSchema(schema);