From 342ee9591af093770683f53ce6912a4adf2a5df3 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Wed, 28 Oct 2026 18:47:17 +0800 Subject: [PATCH 1/9] feat(read): materialize partition columns from file scan tasks --- src/iceberg/avro/avro_data_util.cc | 27 +++- src/iceberg/avro/avro_direct_decoder.cc | 3 +- src/iceberg/avro/avro_reader.cc | 5 +- src/iceberg/avro/avro_schema_util.cc | 61 +++++---- src/iceberg/avro/avro_schema_util_internal.h | 4 +- src/iceberg/catalog/rest/json_serde.cc | 9 +- src/iceberg/catalog/rest/rest_table_scan.cc | 9 +- src/iceberg/data/file_scan_task_reader.cc | 83 ++++++++++-- src/iceberg/file_reader.h | 16 +++ src/iceberg/json_serde.cc | 2 +- src/iceberg/manifest/manifest_group.cc | 3 +- src/iceberg/parquet/parquet_data_util.cc | 15 ++- src/iceberg/parquet/parquet_reader.cc | 11 +- src/iceberg/parquet/parquet_schema_util.cc | 14 +- .../parquet/parquet_schema_util_internal.h | 4 +- src/iceberg/schema_util.cc | 46 +++++++ src/iceberg/schema_util.h | 5 + src/iceberg/table_scan.cc | 17 ++- src/iceberg/table_scan.h | 14 +- src/iceberg/test/avro_test.cc | 25 ++++ .../test/file_scan_task_reader_test.cc | 123 ++++++++++++++++++ src/iceberg/test/manifest_group_test.cc | 4 + src/iceberg/test/parquet_test.cc | 19 +++ src/iceberg/test/rest_json_serde_test.cc | 2 + 24 files changed, 450 insertions(+), 71 deletions(-) diff --git a/src/iceberg/avro/avro_data_util.cc b/src/iceberg/avro/avro_data_util.cc index 39f2948b9f..5689187eae 100644 --- a/src/iceberg/avro/avro_data_util.cc +++ b/src/iceberg/avro/avro_data_util.cc @@ -94,7 +94,8 @@ Status AppendStructToBuilder(const ::avro::NodePtr& avro_node, metadata_context, field_builder)); } else if (field_projection.kind == FieldProjection::Kind::kNull) { ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull()); - } else if (field_projection.kind == FieldProjection::Kind::kDefault) { + } else if (field_projection.kind == FieldProjection::Kind::kDefault || + field_projection.kind == FieldProjection::Kind::kConstant) { ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(field_projection, field_builder)); } else if (field_projection.kind == FieldProjection::Kind::kMetadata) { int32_t field_id = expected_field.field_id(); @@ -476,7 +477,8 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node, return {}; } - if (projection.kind == FieldProjection::Kind::kDefault) { + if (projection.kind == FieldProjection::Kind::kDefault || + projection.kind == FieldProjection::Kind::kConstant) { return AppendDefaultToBuilder(projection, array_builder); } @@ -593,7 +595,13 @@ Status PrepareStructDefaultScalars(std::span projections, auto& field_projection = projections[i]; auto* field_builder = struct_builder->field_builder(static_cast(i)); - if (field_projection.kind == FieldProjection::Kind::kDefault) { + if (field_projection.kind == FieldProjection::Kind::kDefault || + field_projection.kind == FieldProjection::Kind::kConstant) { + if (field_builder->type()->id() == ::arrow::Type::STRUCT) { + ICEBERG_RETURN_UNEXPECTED( + PrepareStructDefaultScalars(field_projection.children, field_builder)); + continue; + } // Get-or-create the single Avro attributes container: another Avro attribute may // have created it already, so guard on `default_scalar` being unset rather than on // the container's presence (otherwise a pre-existing container would skip @@ -640,6 +648,19 @@ Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* bui Status AppendDefaultToBuilder(const FieldProjection& projection, ::arrow::ArrayBuilder* builder) { + if (projection.kind == FieldProjection::Kind::kNull) { + ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendNull()); + return {}; + } + if (builder->type()->id() == ::arrow::Type::STRUCT) { + auto* struct_builder = internal::checked_cast<::arrow::StructBuilder*>(builder); + ICEBERG_ARROW_RETURN_NOT_OK(struct_builder->Append()); + for (size_t i = 0; i < projection.children.size(); ++i) { + ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder( + projection.children[i], struct_builder->field_builder(static_cast(i)))); + } + return {}; + } // Avro projections carry a single attributes type, so once one is attached it is an // AvroExtraAttributes; use checked_cast instead of a per-row dynamic_cast. if (projection.attributes != nullptr) { diff --git a/src/iceberg/avro/avro_direct_decoder.cc b/src/iceberg/avro/avro_direct_decoder.cc index 465bc4b240..7bff2768de 100644 --- a/src/iceberg/avro/avro_direct_decoder.cc +++ b/src/iceberg/avro/avro_direct_decoder.cc @@ -216,7 +216,8 @@ Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder& auto* field_builder = struct_builder->field_builder(static_cast(proj_idx)); if (field_projection.kind == FieldProjection::Kind::kNull) { ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull()); - } else if (field_projection.kind == FieldProjection::Kind::kDefault) { + } else if (field_projection.kind == FieldProjection::Kind::kDefault || + field_projection.kind == FieldProjection::Kind::kConstant) { ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder(field_projection, field_builder)); } else if (field_projection.kind == FieldProjection::Kind::kMetadata) { int32_t field_id = expected_field.field_id(); diff --git a/src/iceberg/avro/avro_reader.cc b/src/iceberg/avro/avro_reader.cc index f333a38cbc..95f7d4e471 100644 --- a/src/iceberg/avro/avro_reader.cc +++ b/src/iceberg/avro/avro_reader.cc @@ -329,8 +329,9 @@ class AvroReader::Impl { } // Project the read schema on top of the file schema. - ICEBERG_ASSIGN_OR_RAISE(projection_, Project(*read_schema_, file_schema.root(), - /*prune_source=*/false)); + ICEBERG_ASSIGN_OR_RAISE(projection_, + Project(*read_schema_, file_schema.root(), + /*prune_source=*/false, options.constants)); ICEBERG_RETURN_UNEXPECTED(backend_->InitWithSchema(file_schema, options.split)); diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc index 94d0a6116b..819b4e2641 100644 --- a/src/iceberg/avro/avro_schema_util.cc +++ b/src/iceberg/avro/avro_schema_util.cc @@ -685,12 +685,13 @@ Status UnwrapUnion(const ::avro::NodePtr& node, ::avro::NodePtr* result) { // Forward declaration Result ProjectNested(const Type& expected_type, - const ::avro::NodePtr& avro_node, - bool prune_source); + const ::avro::NodePtr& avro_node, bool prune_source, + const ColumnConstants& constants); Result ProjectField(const SchemaField& expected_field, const ::avro::NodePtr& avro_node, - size_t source_index, bool prune_source) { + size_t source_index, bool prune_source, + const ColumnConstants& constants) { const Type& expected_type = *expected_field.type(); ::avro::NodePtr field_node; ICEBERG_RETURN_UNEXPECTED(UnwrapUnion(avro_node, &field_node)); @@ -707,8 +708,8 @@ Result ProjectField(const SchemaField& expected_field, } if (expected_type.is_nested()) { - ICEBERG_ASSIGN_OR_RAISE(projection, - ProjectNested(expected_type, field_node, prune_source)); + ICEBERG_ASSIGN_OR_RAISE( + projection, ProjectNested(expected_type, field_node, prune_source, constants)); } else { ICEBERG_RETURN_UNEXPECTED(ValidateAvroSchemaEvolution(expected_type, field_node)); } @@ -718,8 +719,8 @@ Result ProjectField(const SchemaField& expected_field, } Result ProjectStruct(const StructType& struct_type, - const ::avro::NodePtr& avro_node, - bool prune_source) { + const ::avro::NodePtr& avro_node, bool prune_source, + const ColumnConstants& constants) { if (avro_node->type() != ::avro::AVRO_RECORD) { return InvalidSchema("Expected AVRO_RECORD type, but got {}", ToString(avro_node)); } @@ -752,9 +753,13 @@ Result ProjectStruct(const StructType& struct_type, FieldProjection child_projection; if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) { + ICEBERG_ASSIGN_OR_RAISE( + child_projection, + ProjectField(expected_field, iter->second.field_node, iter->second.local_index, + prune_source, constants)); + } else if (auto constant = constants.find(field_id); constant != constants.end()) { ICEBERG_ASSIGN_OR_RAISE(child_projection, - ProjectField(expected_field, iter->second.field_node, - iter->second.local_index, prune_source)); + ProjectConstant(expected_field, constant->second)); } else if (MetadataColumns::IsMetadataColumn(field_id)) { child_projection.kind = FieldProjection::Kind::kMetadata; } else if (expected_field.initial_default() != nullptr) { @@ -778,7 +783,8 @@ Result ProjectStruct(const StructType& struct_type, } Result ProjectList(const ListType& list_type, - const ::avro::NodePtr& avro_node, bool prune_source) { + const ::avro::NodePtr& avro_node, bool prune_source, + const ColumnConstants& constants) { if (avro_node->type() != ::avro::AVRO_ARRAY) { return InvalidSchema("Expected AVRO_ARRAY type, but got {}", ToString(avro_node)); } @@ -795,9 +801,10 @@ Result ProjectList(const ListType& list_type, } FieldProjection element_projection; - ICEBERG_ASSIGN_OR_RAISE(element_projection, - ProjectField(expected_element_field, avro_node->leafAt(0), - /*source_index*/ size_t{0}, prune_source)); + ICEBERG_ASSIGN_OR_RAISE( + element_projection, + ProjectField(expected_element_field, avro_node->leafAt(0), + /*source_index*/ size_t{0}, prune_source, constants)); FieldProjection result; result.children.emplace_back(std::move(element_projection)); @@ -805,7 +812,8 @@ Result ProjectList(const ListType& list_type, } Result ProjectMap(const MapType& map_type, - const ::avro::NodePtr& avro_node, bool prune_source) { + const ::avro::NodePtr& avro_node, bool prune_source, + const ColumnConstants& constants) { const auto& expected_key_field = map_type.key(); const auto& expected_value_field = map_type.value(); @@ -854,9 +862,9 @@ Result ProjectMap(const MapType& map_type, for (size_t i = 0; i < map_node->leaves(); ++i) { FieldProjection sub_projection; const auto& expected_sub_field = map_type.fields()[i]; - ICEBERG_ASSIGN_OR_RAISE( - sub_projection, - ProjectField(expected_sub_field, map_node->leafAt(i), i, prune_source)); + ICEBERG_ASSIGN_OR_RAISE(sub_projection, + ProjectField(expected_sub_field, map_node->leafAt(i), i, + prune_source, constants)); result.children.emplace_back(std::move(sub_projection)); } @@ -864,8 +872,8 @@ Result ProjectMap(const MapType& map_type, } Result ProjectNested(const Type& expected_type, - const ::avro::NodePtr& avro_node, - bool prune_source) { + const ::avro::NodePtr& avro_node, bool prune_source, + const ColumnConstants& constants) { if (!expected_type.is_nested()) { return InvalidSchema("Expected a nested type, but got {}", expected_type); } @@ -873,13 +881,13 @@ Result ProjectNested(const Type& expected_type, switch (expected_type.type_id()) { case TypeId::kStruct: return ProjectStruct(internal::checked_cast(expected_type), - avro_node, prune_source); + avro_node, prune_source, constants); case TypeId::kList: return ProjectList(internal::checked_cast(expected_type), - avro_node, prune_source); + avro_node, prune_source, constants); case TypeId::kMap: return ProjectMap(internal::checked_cast(expected_type), avro_node, - prune_source); + prune_source, constants); default: return InvalidSchema("Unsupported nested type: {}", expected_type); } @@ -894,10 +902,11 @@ bool HasMapLogicalType(const ::avro::NodePtr& node) { } Result Project(const Schema& expected_schema, - const ::avro::NodePtr& avro_node, bool prune_source) { - ICEBERG_ASSIGN_OR_RAISE( - auto field_projection, - ProjectNested(static_cast(expected_schema), avro_node, prune_source)); + const ::avro::NodePtr& avro_node, bool prune_source, + const ColumnConstants& constants) { + ICEBERG_ASSIGN_OR_RAISE(auto field_projection, + ProjectNested(static_cast(expected_schema), + avro_node, prune_source, constants)); return SchemaProjection{std::move(field_projection.children)}; } diff --git a/src/iceberg/avro/avro_schema_util_internal.h b/src/iceberg/avro/avro_schema_util_internal.h index 342b119a53..7cc0012b5a 100644 --- a/src/iceberg/avro/avro_schema_util_internal.h +++ b/src/iceberg/avro/avro_schema_util_internal.h @@ -142,9 +142,11 @@ class HasIdVisitor { /// \param expected_schema The Iceberg Schema that defines the expected structure. /// \param avro_node The Avro node to read data from. /// \param prune_source Whether the source schema can be pruned. +/// \param constants File-level values used for fields absent from the file. /// \return The schema projection result. Result Project(const Schema& expected_schema, - const ::avro::NodePtr& avro_node, bool prune_source); + const ::avro::NodePtr& avro_node, bool prune_source, + const ColumnConstants& constants = {}); std::string ToString(const ::avro::NodePtr& node); std::string ToString(const ::avro::LogicalType& logical_type); diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc index 7b862e7293..46edba4ab0 100644 --- a/src/iceberg/catalog/rest/json_serde.cc +++ b/src/iceberg/catalog/rest/json_serde.cc @@ -231,9 +231,16 @@ Result>> FileScanTasksFromJson( ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json)); } + std::shared_ptr spec; + if (data_file.partition_spec_id.has_value()) { + auto it = partition_spec_by_id.find(*data_file.partition_spec_id); + if (it != partition_spec_by_id.end()) { + spec = it->second; + } + } auto task = FileScanTask::MakeSplit(std::make_shared(std::move(data_file)), start, length, std::move(task_delete_files), - std::move(residual_filter)); + std::move(residual_filter), std::move(spec)); if (!task) { return JsonParseError("Invalid FileScanTask range: {}", task.error().message); } diff --git a/src/iceberg/catalog/rest/rest_table_scan.cc b/src/iceberg/catalog/rest/rest_table_scan.cc index e7dae0f492..b884875a74 100644 --- a/src/iceberg/catalog/rest/rest_table_scan.cc +++ b/src/iceberg/catalog/rest/rest_table_scan.cc @@ -278,9 +278,12 @@ class RestFileScanTaskStream final : public FileScanTaskStream { auto task = std::move(buffer_[buffer_pos_++]); if (ignore_residuals_ || !task->residual_filter()) { // The server may leave row filtering to the client. - return std::make_shared( - task->data_file(), task->delete_files(), - ignore_residuals_ ? True::Instance() : filter_); + ICEBERG_ASSIGN_OR_RAISE( + auto adjusted, + FileScanTask::MakeSplit( + task->data_file(), task->start(), task->length(), task->delete_files(), + ignore_residuals_ ? True::Instance() : filter_, task->spec())); + return adjusted; } return task; } diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 8c55e31b2b..9de45a39d8 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -30,23 +30,75 @@ #include "iceberg/data/delete_filter.h" #include "iceberg/file_reader.h" #include "iceberg/manifest/manifest_entry.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/partition_spec.h" #include "iceberg/schema.h" #include "iceberg/schema_internal.h" #include "iceberg/table_scan.h" +#include "iceberg/transform.h" #include "iceberg/util/macros.h" namespace iceberg { namespace { -ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr io, - std::shared_ptr projection, - std::shared_ptr filter, - std::shared_ptr name_mapping, - ReaderProperties properties, - std::optional first_row_id, - std::optional data_sequence_number) { +Result ConstantsOf(const FileScanTask& task, const Schema& projection) { + const auto& file = *task.data_file(); + const auto& spec = task.spec(); + ICEBERG_ASSIGN_OR_RAISE(auto partition_field, + projection.FindFieldById(MetadataColumns::kPartitionColumnId)); + ICEBERG_ASSIGN_OR_RAISE(auto spec_field, + projection.FindFieldById(MetadataColumns::kSpecIdColumnId)); + ColumnConstants constants; + auto spec_id = file.partition_spec_id; + if (spec) { + if (spec_id.has_value() && *spec_id != spec->spec_id()) { + return InvalidArgument("Data file spec ID {} does not match task spec ID {}", + *spec_id, spec->spec_id()); + } + spec_id = spec->spec_id(); + if (file.partition.num_fields() != spec->fields().size()) { + return InvalidArgument( + "Partition value count {} does not match spec field count {}", + file.partition.num_fields(), spec->fields().size()); + } + StructConstant partition; + for (size_t i = 0; i < spec->fields().size(); ++i) { + const auto& field = spec->fields()[i]; + const auto& value = file.partition.values()[i]; + if (field.transform() == nullptr) { + return InvalidArgument("Missing transform for partition field {}", + field.field_id()); + } + if (field.transform()->transform_type() == TransformType::kIdentity) { + constants.emplace(field.source_id(), value); + } + if (partition_field.has_value()) { + partition.emplace(field.field_id(), value); + } + } + if (partition_field.has_value()) { + constants.emplace(MetadataColumns::kPartitionColumnId, std::move(partition)); + } + } else if (file.partition.num_fields() != 0 || partition_field.has_value()) { + return InvalidArgument("A partition spec is required to read partition values"); + } + if (spec_field.has_value()) { + if (!spec_id.has_value()) { + return InvalidArgument("A partition spec ID is required to read _spec_id"); + } + constants.emplace(MetadataColumns::kSpecIdColumnId, Literal::Int(*spec_id)); + } + return constants; +} + +Result MakeReaderOptions( + const FileScanTask& task, std::shared_ptr io, + std::shared_ptr projection, std::shared_ptr filter, + std::shared_ptr name_mapping, ReaderProperties properties, + std::optional first_row_id, std::optional data_sequence_number) { const auto& data_file = *task.data_file(); + ICEBERG_ASSIGN_OR_RAISE(auto constants, ConstantsOf(task, *projection)); return ReaderOptions{ .path = data_file.file_path, .length = static_cast(data_file.file_size_in_bytes), @@ -61,6 +113,7 @@ ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptrfirst_row_id, data_file->data_sequence_number); + ICEBERG_ASSIGN_OR_RAISE( + auto options, + MakeReaderOptions(task, io_, projected_schema_, task.residual_filter(), + name_mapping_, properties_, data_file->first_row_id, + data_file->data_sequence_number)); ICEBERG_ASSIGN_OR_RAISE( auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options)); return MakeArrowArrayStream(std::move(reader)); @@ -215,9 +270,11 @@ class FileScanTaskReader::Impl { ProjectionContext::Make(*required_schema, *projected_schema_, project_batch_function)); - auto options = MakeReaderOptions(task, io_, required_schema, task.residual_filter(), - name_mapping_, properties_, data_file->first_row_id, - data_file->data_sequence_number); + ICEBERG_ASSIGN_OR_RAISE( + auto options, + MakeReaderOptions(task, io_, required_schema, task.residual_filter(), + name_mapping_, properties_, data_file->first_row_id, + data_file->data_sequence_number)); ICEBERG_ASSIGN_OR_RAISE(auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options)); diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index e08b2df494..6c536b7b4b 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -26,8 +26,11 @@ #include #include #include +#include +#include #include "iceberg/arrow_c_data.h" +#include "iceberg/expression/literal.h" #include "iceberg/file_format.h" #include "iceberg/result.h" #include "iceberg/type_fwd.h" @@ -92,6 +95,14 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase { const std::unordered_map& properties); }; +/// A non-null struct constant, keyed by child field ID. Children are primitive +/// literals, including typed nulls. Projected children absent from the map are null. +using StructConstant = std::unordered_map; + +/// A file-level primitive or struct constant, independent of the file format. +using ColumnConstant = std::variant; +using ColumnConstants = std::unordered_map; + /// \brief Options for creating a reader. struct ICEBERG_EXPORT ReaderOptions { /// \brief The path to the file to read. @@ -116,6 +127,11 @@ struct ICEBERG_EXPORT ReaderOptions { std::optional data_sequence_number; /// \brief Format-specific or implementation-specific properties. ReaderProperties properties; + + /// File-level values keyed by field ID, used when physical columns are absent. + /// Constants precede initial defaults and optional-field nulls. + /// Example: `options.constants.emplace(2, Literal::String("CN"));` + ColumnConstants constants; }; /// \brief Factory function to create a reader of a specific file format. diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc index 37f6c08c94..6174db7e28 100644 --- a/src/iceberg/json_serde.cc +++ b/src/iceberg/json_serde.cc @@ -2551,7 +2551,7 @@ Result> FileScanTaskFromJson(const nlohmann::json& auto task = FileScanTask::MakeSplit(std::make_shared(std::move(data_file)), start, length, std::move(delete_files), - std::move(residual_filter)); + std::move(residual_filter), std::move(shared_spec)); if (!task) { return JsonParseError("Invalid FileScanTask range: {}", task.error().message); } diff --git a/src/iceberg/manifest/manifest_group.cc b/src/iceberg/manifest/manifest_group.cc index 036a71bdcb..883fd770ef 100644 --- a/src/iceberg/manifest/manifest_group.cc +++ b/src/iceberg/manifest/manifest_group.cc @@ -320,7 +320,8 @@ class ManifestGroup::FilePlanningStream final : public FileScanTaskStream { residuals->ResidualFor(value.data_file->partition)); return std::optional>{std::make_shared( - std::move(value.data_file), std::move(delete_files), std::move(residual))}; + std::move(value.data_file), std::move(delete_files), std::move(residual), + group_->specs_by_id_.at(spec_id))}; } } diff --git a/src/iceberg/parquet/parquet_data_util.cc b/src/iceberg/parquet/parquet_data_util.cc index 73597cbfb8..dc43c5358e 100644 --- a/src/iceberg/parquet/parquet_data_util.cc +++ b/src/iceberg/parquet/parquet_data_util.cc @@ -251,7 +251,20 @@ Result> ProjectStructArray( ICEBERG_ASSIGN_OR_RAISE( projected_array, MakeNullArray(output_arrow_type, struct_array->length(), pool)); - } else if (field_projection.kind == FieldProjection::Kind::kDefault) { + } else if (field_projection.kind == FieldProjection::Kind::kConstant && + projected_field.type()->type_id() == TypeId::kStruct) { + // An empty source struct supplies the batch length; children come entirely + // from the constant projection and retain the requested field IDs/order. + auto source = std::make_shared<::arrow::StructArray>( + ::arrow::struct_({}), struct_array->length(), ::arrow::ArrayVector{}); + ICEBERG_ASSIGN_OR_RAISE( + projected_array, + ProjectNestedArray( + source, output_arrow_type, + internal::checked_cast(*projected_field.type()), + field_projection.children, metadata_context, row_positions, pool)); + } else if (field_projection.kind == FieldProjection::Kind::kDefault || + field_projection.kind == FieldProjection::Kind::kConstant) { ICEBERG_ASSIGN_OR_RAISE( projected_array, arrow::MakeDefaultArray(std::get(field_projection.from), diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 549a3d3aed..55aa91e9a2 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -62,7 +62,8 @@ Result> OpenInputStream( Result BuildProjection(::parquet::arrow::FileReader* reader, const Schema& read_schema, - const NameMapping* name_mapping) { + const NameMapping* name_mapping, + const ColumnConstants& constants) { auto metadata = reader->parquet_reader()->metadata(); ::parquet::arrow::SchemaManifest schema_manifest; @@ -79,7 +80,8 @@ Result BuildProjection(::parquet::arrow::FileReader* reader, } // Leverage SchemaManifest to project the schema - ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, schema_manifest)); + ICEBERG_ASSIGN_OR_RAISE(auto projection, + Project(read_schema, schema_manifest, constants)); return projection; } @@ -304,8 +306,9 @@ class ParquetReader::Impl { ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options)); // Project read schema onto the Parquet file schema - ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_, - options.name_mapping.get())); + ICEBERG_ASSIGN_OR_RAISE( + projection_, BuildProjection(reader_.get(), *read_schema_, + options.name_mapping.get(), options.constants)); metadata_context_ = {.file_path = options.path, .first_row_id = options.first_row_id, .data_sequence_number = options.data_sequence_number}; diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 2a230fe4cc..720dfb862f 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -508,7 +508,9 @@ namespace { class ProjectionBuilder { public: - explicit ProjectionBuilder(const ::parquet::SchemaDescriptor* descr) : descr_(descr) {} + ProjectionBuilder(const ::parquet::SchemaDescriptor* descr, + const ColumnConstants& constants) + : descr_(descr), constants_(constants) {} Result Project( const Type& nested_type, @@ -596,6 +598,10 @@ class ProjectionBuilder { const auto& parquet_field = iter->second.parquet_field; ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectField(field, parquet_field, iter->second.local_index)); + } else if (auto constant = constants_.find(field_id); + constant != constants_.end()) { + ICEBERG_ASSIGN_OR_RAISE(child_projection, + ProjectConstant(field, constant->second)); } else if (MetadataColumns::IsMetadataColumn(field_id)) { child_projection.kind = FieldProjection::Kind::kMetadata; } else if (field.initial_default() != nullptr) { @@ -723,6 +729,7 @@ class ProjectionBuilder { } const ::parquet::SchemaDescriptor* descr_; + const ColumnConstants& constants_; }; void CollectColumnIds(const FieldProjection& field_projection, @@ -788,8 +795,9 @@ void ApplyNameMappingToField(const NameMapping& mapping, std::vector Project(const Schema& expected_schema, - const ::parquet::arrow::SchemaManifest& parquet_schema) { - ProjectionBuilder builder(parquet_schema.descr); + const ::parquet::arrow::SchemaManifest& parquet_schema, + const ColumnConstants& constants) { + ProjectionBuilder builder(parquet_schema.descr, constants); ICEBERG_ASSIGN_OR_RAISE(auto field_projection, builder.Project(static_cast(expected_schema), parquet_schema.schema_fields)); diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h index 6612de0569..bc350e11fd 100644 --- a/src/iceberg/parquet/parquet_schema_util_internal.h +++ b/src/iceberg/parquet/parquet_schema_util_internal.h @@ -47,10 +47,12 @@ struct ParquetExtraAttributes : public FieldProjection::ExtraAttributes { /// /// \param expected_schema The Iceberg Schema that defines the expected structure. /// \param parquet_schema The Parquet schema to read data from. +/// \param constants File-level values used for fields absent from the file. /// \return The schema projection result with column indices of projected Parquet columns /// specified via ParquetExtraAttributes. Result Project(const Schema& expected_schema, - const ::parquet::arrow::SchemaManifest& parquet_schema); + const ::parquet::arrow::SchemaManifest& parquet_schema, + const ColumnConstants& constants = {}); /// \brief Get the selected column indices by walking through the projection result. /// diff --git a/src/iceberg/schema_util.cc b/src/iceberg/schema_util.cc index b9f32346a9..7d22d002cb 100644 --- a/src/iceberg/schema_util.cc +++ b/src/iceberg/schema_util.cc @@ -194,6 +194,52 @@ Result ProjectNested(const Type& expected_type, const Type& sou } // namespace +Result ProjectConstant(const SchemaField& field, + const ColumnConstant& constant) { + FieldProjection projection{.kind = FieldProjection::Kind::kConstant}; + if (const auto* literal = std::get_if(&constant)) { + if (!field.type()->is_primitive() || literal->type() == nullptr || + literal->IsAboveMax() || literal->IsBelowMin()) { + return InvalidSchema("Invalid constant for field {}", field.field_id()); + } + ICEBERG_RETURN_UNEXPECTED(ValidateSchemaEvolution(*field.type(), *literal->type())); + if (literal->IsNull() && !field.optional()) { + return InvalidSchema("Null constant for required field {}", field.field_id()); + } + if (literal->IsNull()) { + projection.kind = FieldProjection::Kind::kNull; + return projection; + } + projection.from = *literal; + return projection; + } + + if (field.type()->type_id() != TypeId::kStruct) { + return InvalidSchema("Struct constant requires a struct field: {}", field.field_id()); + } + const auto& fields = internal::checked_cast(*field.type()).fields(); + const auto& values = std::get(constant); + if (fields.empty() && field.optional()) { + projection.kind = FieldProjection::Kind::kNull; + return projection; + } + for (const auto& child : fields) { + if (!child.type()->is_primitive()) { + return InvalidSchema("Struct constants require primitive children: {}", + child.field_id()); + } + if (const auto it = values.find(child.field_id()); it != values.end()) { + ICEBERG_ASSIGN_OR_RAISE(auto child_projection, ProjectConstant(child, it->second)); + projection.children.push_back(std::move(child_projection)); + } else if (child.optional()) { + projection.children.push_back({.kind = FieldProjection::Kind::kNull}); + } else { + return InvalidSchema("Missing required constant field {}", child.field_id()); + } + } + return projection; +} + Result Project(const Schema& expected_schema, const Schema& source_schema, bool prune_source) { ICEBERG_ASSIGN_OR_RAISE(auto field_projection, diff --git a/src/iceberg/schema_util.h b/src/iceberg/schema_util.h index 421d00e0ad..0c96795979 100644 --- a/src/iceberg/schema_util.h +++ b/src/iceberg/schema_util.h @@ -28,6 +28,7 @@ #include #include "iceberg/expression/literal.h" +#include "iceberg/file_reader.h" #include "iceberg/iceberg_export.h" #include "iceberg/result.h" #include "iceberg/type_fwd.h" @@ -95,6 +96,10 @@ ICEBERG_EXPORT Result Project(const Schema& expected_schema, const Schema& source_schema, bool prune_source); +/// Build a typed constant projection, matching struct children by field ID. +ICEBERG_EXPORT Result ProjectConstant(const SchemaField& field, + const ColumnConstant& constant); + ICEBERG_EXPORT std::string_view ToString(FieldProjection::Kind kind); ICEBERG_EXPORT std::string ToString(const FieldProjection& projection); ICEBERG_EXPORT std::string ToString(const SchemaProjection& projection); diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index d04e3385b6..93711fc5f2 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -174,25 +174,28 @@ ScanTask::~ScanTask() = default; FileScanTask::FileScanTask(std::shared_ptr data_file, std::vector> delete_files, - std::shared_ptr residual_filter) + std::shared_ptr residual_filter, + std::shared_ptr spec) : data_file_(std::move(data_file)), delete_files_(std::move(delete_files)), - residual_filter_(std::move(residual_filter)) { + residual_filter_(std::move(residual_filter)), + spec_(std::move(spec)) { ICEBERG_DCHECK(data_file_ != nullptr, "Data file cannot be null for FileScanTask"); } FileScanTask::FileScanTask(std::shared_ptr data_file, std::vector> delete_files, - std::shared_ptr residual_filter, Range range) + std::shared_ptr residual_filter, Range range, + std::shared_ptr spec) : FileScanTask(std::move(data_file), std::move(delete_files), - std::move(residual_filter)) { + std::move(residual_filter), std::move(spec)) { range_ = range; } Result> FileScanTask::MakeSplit( std::shared_ptr data_file, int64_t start, int64_t length, std::vector> delete_files, - std::shared_ptr filter) { + std::shared_ptr filter, std::shared_ptr spec) { if (data_file == nullptr) { return InvalidArgument("Cannot create a file scan task without a data file"); } @@ -215,7 +218,7 @@ Result> FileScanTask::MakeSplit( // only valid empty range: [0, 0) of an empty file. if (start == 0 && length == file_size) { return std::make_shared(std::move(data_file), std::move(delete_files), - std::move(filter)); + std::move(filter), std::move(spec)); } if (length == 0) { return InvalidArgument("A partial split of file {} must have positive length", @@ -224,7 +227,7 @@ Result> FileScanTask::MakeSplit( return std::shared_ptr( new FileScanTask(std::move(data_file), std::move(delete_files), std::move(filter), - Range{.start = start, .length = length})); + Range{.start = start, .length = length}, std::move(spec))); } int64_t FileScanTask::length() const { diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h index 1307089e39..369d9a2286 100644 --- a/src/iceberg/table_scan.h +++ b/src/iceberg/table_scan.h @@ -72,11 +72,13 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask { /// \param data_file The data file to read. /// \param delete_files Delete files that apply to this data file. /// \param filter Optional residual filter to apply after reading. + /// \param spec Partition spec corresponding to the data file partition tuple. /// \note The shared data file, delete files, and residual filter must remain /// unchanged for this task's lifetime. Metadata changes require rebuilding tasks. explicit FileScanTask(std::shared_ptr data_file, std::vector> delete_files = {}, - std::shared_ptr filter = nullptr); + std::shared_ptr filter = nullptr, + std::shared_ptr spec = nullptr); /// \brief Make a task for a validated half-open byte range [start, start + length). /// @@ -87,7 +89,8 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask { static Result> MakeSplit( std::shared_ptr data_file, int64_t start, int64_t length, std::vector> delete_files = {}, - std::shared_ptr filter = nullptr); + std::shared_ptr filter = nullptr, + std::shared_ptr spec = nullptr); /// \brief The inclusive byte offset where this task starts, or zero for a whole file. int64_t start() const { return range_ ? range_->start : 0; } @@ -109,6 +112,9 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask { /// \brief Residual filter to apply after reading. const std::shared_ptr& residual_filter() const { return residual_filter_; } + /// The partition spec used to write this file, including historical field IDs. + const std::shared_ptr& spec() const { return spec_; } + Kind kind() const override { return Kind::kFileScanTask; } int64_t size_bytes() const override; int32_t files_count() const override; @@ -127,12 +133,14 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask { FileScanTask(std::shared_ptr data_file, std::vector> delete_files, - std::shared_ptr filter, Range range); + std::shared_ptr filter, Range range, + std::shared_ptr spec); std::shared_ptr data_file_; std::vector> delete_files_; std::shared_ptr residual_filter_; std::optional range_; + std::shared_ptr spec_; }; /// \brief Stream of file scan tasks. diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc index a161164102..8a7ca276bb 100644 --- a/src/iceberg/test/avro_test.cc +++ b/src/iceberg/test/avro_test.cc @@ -481,6 +481,31 @@ class AvroReaderParameterizedTest : public AvroReaderTest, } }; +TEST_P(AvroReaderParameterizedTest, ReadsPartitionConstants) { + CreateSimpleAvroFile(); + auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(3, "region", string()), + SchemaField::MakeOptional(4, "missing", string()), + SchemaField::MakeOptional( + MetadataColumns::kPartitionColumnId, "_partition", + struct_({SchemaField::MakeOptional(1001, "bucket", int64()), + SchemaField::MakeOptional(1002, "added", string())}))}); + ReaderOptions options{.path = temp_avro_file_, .io = file_io_, .projection = schema}; + options.properties.Set(ReaderProperties::kAvroSkipDatum, skip_datum_); + options.constants = { + {1, Literal::Int(999)}, + {3, Literal::String("CN")}, + {4, Literal::Null(string())}, + {MetadataColumns::kPartitionColumnId, StructConstant{{1001, Literal::Int(7)}}}}; + ICEBERG_UNWRAP_OR_FAIL(auto reader, + ReaderFactoryRegistry::Open(FileFormatType::kAvro, options)); + VerifyNextBatch(*reader, R"([[1,"CN",null,[7,null]], + [2,"CN",null,[7,null]], + [3,"CN",null,[7,null]]])"); + VerifyExhausted(*reader); +} + TEST_P(AvroReaderParameterizedTest, AvroWriterBasicType) { auto schema = std::make_shared(std::vector{ SchemaField::MakeRequired(1, "name", std::make_shared())}); diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index 05416bd7d7..f5ea2c5b13 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -33,6 +33,7 @@ #include #include #include +#include #include #include #include @@ -51,6 +52,7 @@ #include "iceberg/file_format.h" #include "iceberg/file_io.h" #include "iceberg/file_reader.h" +#include "iceberg/json_serde_internal.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/metadata_columns.h" #include "iceberg/parquet/parquet_register.h" @@ -63,6 +65,7 @@ #include "iceberg/test/matchers.h" #include "iceberg/test/std_io.h" #include "iceberg/test/temp_file_test_base.h" +#include "iceberg/transform.h" #include "iceberg/type.h" #include "iceberg/util/macros.h" @@ -418,6 +421,126 @@ class FileScanTaskReaderTest : public TempFileTestBase { std::shared_ptr projected_schema_; }; +TEST_F(FileScanTaskReaderTest, IdentityPartitionConstantsSurviveJsonAndSplit) { + auto physical = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32())}); + ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(physical, "[[1], [2], [3]]")); + ICEBERG_UNWRAP_OR_FAIL( + auto spec, PartitionSpec::Make(7, {{2, 1000, "name", Transform::Identity()}})); + file->partition_spec_id = 7; + file->partition = PartitionValues({Literal::String("CN")}); + FileScanTask task(file, {}, nullptr, std::move(spec)); + ICEBERG_UNWRAP_OR_FAIL( + auto json, ToJson(task, {{7, std::const_pointer_cast(task.spec())}}, + *table_schema_)); + ICEBERG_UNWRAP_OR_FAIL(auto parsed, FileScanTaskFromJson(json)); + ASSERT_NE(parsed->spec(), nullptr); + EXPECT_EQ(parsed->spec()->spec_id(), 7); + ICEBERG_UNWRAP_OR_FAIL( + auto split, FileScanTask::MakeSplit(parsed->data_file(), 0, parsed->length(), {}, + nullptr, parsed->spec())); + EXPECT_EQ(split->spec(), parsed->spec()); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = projected_schema_})); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*split)); + VerifyStream(&stream, R"([[1, "CN"], [2, "CN"], [3, "CN"]])"); +} + +TEST_F(FileScanTaskReaderTest, IdentityConstantsApplyBeforeEqualityDeletes) { + auto physical = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32())}); + ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(physical, "[[1], [2], [3]]")); + auto delete_schema = std::make_shared( + std::vector{SchemaField::MakeOptional(2, "name", string())}); + ICEBERG_UNWRAP_OR_FAIL( + auto deletes, MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"), + delete_schema, R"([["CN"]])", {2})); + ICEBERG_UNWRAP_OR_FAIL( + auto spec, PartitionSpec::Make(7, {{2, 1000, "name", Transform::Identity()}})); + file->partition_spec_id = 7; + file->partition = PartitionValues({Literal::String("CN")}); + FileScanTask task(file, {deletes}, nullptr, std::move(spec)); + ICEBERG_UNWRAP_OR_FAIL(auto reader, + FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = physical})); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie(); + auto table = batches->ToTable().ValueOrDie(); + EXPECT_EQ(table->num_rows(), 0); +} + +TEST_F(FileScanTaskReaderTest, PartitionMetadataMatchesHistoricalFieldIds) { + ICEBERG_UNWRAP_OR_FAIL( + auto file, MakeDataFile(table_schema_, + R"([[1, "physical", "blue"], [2, "physical", "red"]])", 2)); + ICEBERG_UNWRAP_OR_FAIL( + auto spec, PartitionSpec::Make(4, {{1, 1000, "bucket", Transform::Bucket(16)}, + {2, 1001, "name", Transform::Identity()}})); + file->partition_spec_id = 4; + file->partition = PartitionValues({Literal::Int(5), Literal::String("constant")}); + auto projection = std::make_shared(std::vector{ + SchemaField::MakeOptional(2, "renamed", string()), MetadataColumns::kSpecId, + SchemaField::MakeOptional( + MetadataColumns::kPartitionColumnId, "_partition", + struct_({SchemaField::MakeOptional(1001, "region", string()), + SchemaField::MakeOptional(1002, "added", int32()), + SchemaField::MakeOptional(1000, "bucket", int64())}))}); + FileScanTask task(file, {}, nullptr, std::move(spec)); + ICEBERG_UNWRAP_OR_FAIL(auto reader, + FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = projection})); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + VerifyStream(&stream, R"([["physical", 4, ["constant", null, 5]], + ["physical", 4, ["constant", null, 5]]])"); +} + +TEST_F(FileScanTaskReaderTest, NullIdentityOverridesInitialDefault) { + auto physical = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32())}); + ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(physical, "[[1], [2], [3]]")); + ICEBERG_UNWRAP_OR_FAIL( + auto spec, PartitionSpec::Make(0, {{2, 1000, "name", Transform::Identity()}})); + file->partition = PartitionValues({Literal::Null(string())}); + auto projection = std::make_shared(std::vector{ + SchemaField(2, "name", string(), true, {}, + std::make_shared(Literal::String("default")))}); + FileScanTask task(file, {}, nullptr, std::move(spec)); + ICEBERG_UNWRAP_OR_FAIL(auto reader, + FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = projection})); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + VerifyStream(&stream, "[[null], [null], [null]]"); +} + +TEST_F(FileScanTaskReaderTest, RejectsMissingOrInconsistentPartitionContext) { + ICEBERG_UNWRAP_OR_FAIL(auto file, + MakeDataFile(table_schema_, R"([[1, "Foo", "blue"]])", 1)); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = projected_schema_})); + file->partition = PartitionValues({Literal::String("CN")}); + EXPECT_FALSE(reader->Open(FileScanTask(file)).has_value()); + ICEBERG_UNWRAP_OR_FAIL( + auto spec, PartitionSpec::Make(3, {{2, 1000, "name", Transform::Identity()}})); + auto shared_spec = std::shared_ptr(std::move(spec)); + file->partition_spec_id = 9; + EXPECT_FALSE(reader->Open(FileScanTask(file, {}, nullptr, shared_spec)).has_value()); + file->partition_spec_id = 3; + file->partition = PartitionValues{}; + EXPECT_FALSE(reader->Open(FileScanTask(file, {}, nullptr, shared_spec)).has_value()); +} + TEST_F(FileScanTaskReaderTest, OpenWithoutDeletesReadsProjectedSchema) { ICEBERG_UNWRAP_OR_FAIL( auto data_file, diff --git a/src/iceberg/test/manifest_group_test.cc b/src/iceberg/test/manifest_group_test.cc index 063ce5804b..84bd971a96 100644 --- a/src/iceberg/test/manifest_group_test.cc +++ b/src/iceberg/test/manifest_group_test.cc @@ -290,6 +290,10 @@ TEST_P(ManifestGroupTest, CreateAndGetEntries) { ASSERT_EQ(tasks.size(), 2); EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet", "/path/to/data2.parquet")); + for (const auto& task : tasks) { + ASSERT_NE(task->spec(), nullptr); + EXPECT_EQ(task->spec()->spec_id(), task->data_file()->partition_spec_id); + } EXPECT_EQ(tasks[0]->delete_files().size(), 1); EXPECT_EQ(tasks[1]->delete_files()[0]->file_path, "/path/to/delete.parquet"); EXPECT_EQ(tasks[1]->delete_files().size(), 1); diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc index c35ba44717..a89c3c98b1 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -552,6 +552,25 @@ class ParquetReaderTest : public TempFileTestBase { std::string temp_parquet_file_; }; +TEST_F(ParquetReaderTest, ConstantsValidateTypesAndRequiredNulls) { + CreateSimpleParquetFile(); + auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(3, "constant", int64())}); + for (const auto& value : {Literal::String("7"), Literal::Null(int64())}) { + ReaderOptions options{ + .path = temp_parquet_file_, .io = file_io_, .projection = schema}; + options.constants.emplace(3, value); + EXPECT_FALSE( + ReaderFactoryRegistry::Open(FileFormatType::kParquet, options).has_value()); + } + ReaderOptions options{.path = temp_parquet_file_, .io = file_io_, .projection = schema}; + options.constants.emplace(3, Literal::Int(7)); + ICEBERG_UNWRAP_OR_FAIL(auto reader, + ReaderFactoryRegistry::Open(FileFormatType::kParquet, options)); + VerifyNextBatch(*reader, "[[7], [7], [7]]"); + VerifyExhausted(*reader); +} + TEST_F(ParquetReaderTest, ReadTwoFields) { CreateSimpleParquetFile(); diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc index ae5ba1a74e..b3ba428e52 100644 --- a/src/iceberg/test/rest_json_serde_test.cc +++ b/src/iceberg/test/rest_json_serde_test.cc @@ -2652,6 +2652,8 @@ TEST(FetchScanTasksResponseRoundtripTest, PreservesResidualFilter) { ASSERT_TRUE(result->file_scan_tasks.has_value()); ASSERT_EQ(result->file_scan_tasks->size(), 1); ASSERT_NE(result->file_scan_tasks->at(0)->residual_filter(), nullptr); + ASSERT_NE(result->file_scan_tasks->at(0)->spec(), nullptr); + EXPECT_EQ(result->file_scan_tasks->at(0)->spec()->spec_id(), 0); ICEBERG_UNWRAP_OR_FAIL(auto roundtrip_json, ToJson(*result, UnpartitionedSpecs(), EmptySchema())); From 5787b422468b416533407700027dd0d03d540aef Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Wed, 28 Oct 2026 19:07:14 +0800 Subject: [PATCH 2/9] fix(read): prefer partition constants as in Iceberg Java --- src/iceberg/avro/avro_schema_util.cc | 8 ++++---- src/iceberg/file_reader.h | 4 ++-- src/iceberg/parquet/parquet_schema_util.cc | 11 +++++------ src/iceberg/test/avro_test.cc | 6 +++--- src/iceberg/test/file_scan_task_reader_test.cc | 4 ++-- 5 files changed, 16 insertions(+), 17 deletions(-) diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc index 819b4e2641..e1003117cd 100644 --- a/src/iceberg/avro/avro_schema_util.cc +++ b/src/iceberg/avro/avro_schema_util.cc @@ -752,14 +752,14 @@ Result ProjectStruct(const StructType& struct_type, int32_t field_id = expected_field.field_id(); FieldProjection child_projection; - if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) { + if (auto constant = constants.find(field_id); constant != constants.end()) { + ICEBERG_ASSIGN_OR_RAISE(child_projection, + ProjectConstant(expected_field, constant->second)); + } else if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) { ICEBERG_ASSIGN_OR_RAISE( child_projection, ProjectField(expected_field, iter->second.field_node, iter->second.local_index, prune_source, constants)); - } else if (auto constant = constants.find(field_id); constant != constants.end()) { - ICEBERG_ASSIGN_OR_RAISE(child_projection, - ProjectConstant(expected_field, constant->second)); } else if (MetadataColumns::IsMetadataColumn(field_id)) { child_projection.kind = FieldProjection::Kind::kMetadata; } else if (expected_field.initial_default() != nullptr) { diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index 6c536b7b4b..69a8cdf8a8 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -128,8 +128,8 @@ struct ICEBERG_EXPORT ReaderOptions { /// \brief Format-specific or implementation-specific properties. ReaderProperties properties; - /// File-level values keyed by field ID, used when physical columns are absent. - /// Constants precede initial defaults and optional-field nulls. + /// File-level values keyed by field ID, overriding physical columns, + /// initial defaults and optional-field nulls, matching Iceberg Java. /// Example: `options.constants.emplace(2, Literal::String("CN"));` ColumnConstants constants; }; diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 720dfb862f..ebf6c4cd14 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -593,15 +593,14 @@ class ProjectionBuilder { int32_t field_id = field.field_id(); FieldProjection child_projection; - if (auto iter = field_context_map.find(field_id); - iter != field_context_map.cend()) { + if (auto constant = constants_.find(field_id); constant != constants_.end()) { + ICEBERG_ASSIGN_OR_RAISE(child_projection, + ProjectConstant(field, constant->second)); + } else if (auto iter = field_context_map.find(field_id); + iter != field_context_map.cend()) { const auto& parquet_field = iter->second.parquet_field; ICEBERG_ASSIGN_OR_RAISE(child_projection, ProjectField(field, parquet_field, iter->second.local_index)); - } else if (auto constant = constants_.find(field_id); - constant != constants_.end()) { - ICEBERG_ASSIGN_OR_RAISE(child_projection, - ProjectConstant(field, constant->second)); } else if (MetadataColumns::IsMetadataColumn(field_id)) { child_projection.kind = FieldProjection::Kind::kMetadata; } else if (field.initial_default() != nullptr) { diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc index 8a7ca276bb..31c35aafb6 100644 --- a/src/iceberg/test/avro_test.cc +++ b/src/iceberg/test/avro_test.cc @@ -500,9 +500,9 @@ TEST_P(AvroReaderParameterizedTest, ReadsPartitionConstants) { {MetadataColumns::kPartitionColumnId, StructConstant{{1001, Literal::Int(7)}}}}; ICEBERG_UNWRAP_OR_FAIL(auto reader, ReaderFactoryRegistry::Open(FileFormatType::kAvro, options)); - VerifyNextBatch(*reader, R"([[1,"CN",null,[7,null]], - [2,"CN",null,[7,null]], - [3,"CN",null,[7,null]]])"); + VerifyNextBatch(*reader, R"([[999,"CN",null,[7,null]], + [999,"CN",null,[7,null]], + [999,"CN",null,[7,null]]])"); VerifyExhausted(*reader); } diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index f5ea2c5b13..f51cb4926e 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -497,8 +497,8 @@ TEST_F(FileScanTaskReaderTest, PartitionMetadataMatchesHistoricalFieldIds) { .schemas = {table_schema_}, .projected_schema = projection})); ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); - VerifyStream(&stream, R"([["physical", 4, ["constant", null, 5]], - ["physical", 4, ["constant", null, 5]]])"); + VerifyStream(&stream, R"([["constant", 4, ["constant", null, 5]], + ["constant", 4, ["constant", null, 5]]])"); } TEST_F(FileScanTaskReaderTest, NullIdentityOverridesInitialDefault) { From a5d09a94788833c25f3926db3f320537c558134c Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Wed, 28 Oct 2026 19:11:43 +0800 Subject: [PATCH 3/9] refactor(read): limit partition constants to identity source columns --- src/iceberg/avro/avro_data_util.cc | 18 ------ src/iceberg/data/file_scan_task_reader.cc | 64 +++++++------------ src/iceberg/file_reader.h | 10 +-- src/iceberg/parquet/parquet_data_util.cc | 12 ---- src/iceberg/schema_util.cc | 50 +++------------ src/iceberg/schema_util.h | 4 +- src/iceberg/test/avro_test.cc | 23 +++---- .../test/file_scan_task_reader_test.cc | 27 -------- 8 files changed, 44 insertions(+), 164 deletions(-) diff --git a/src/iceberg/avro/avro_data_util.cc b/src/iceberg/avro/avro_data_util.cc index 5689187eae..8e460b68d0 100644 --- a/src/iceberg/avro/avro_data_util.cc +++ b/src/iceberg/avro/avro_data_util.cc @@ -597,11 +597,6 @@ Status PrepareStructDefaultScalars(std::span projections, if (field_projection.kind == FieldProjection::Kind::kDefault || field_projection.kind == FieldProjection::Kind::kConstant) { - if (field_builder->type()->id() == ::arrow::Type::STRUCT) { - ICEBERG_RETURN_UNEXPECTED( - PrepareStructDefaultScalars(field_projection.children, field_builder)); - continue; - } // Get-or-create the single Avro attributes container: another Avro attribute may // have created it already, so guard on `default_scalar` being unset rather than on // the container's presence (otherwise a pre-existing container would skip @@ -648,19 +643,6 @@ Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* bui Status AppendDefaultToBuilder(const FieldProjection& projection, ::arrow::ArrayBuilder* builder) { - if (projection.kind == FieldProjection::Kind::kNull) { - ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendNull()); - return {}; - } - if (builder->type()->id() == ::arrow::Type::STRUCT) { - auto* struct_builder = internal::checked_cast<::arrow::StructBuilder*>(builder); - ICEBERG_ARROW_RETURN_NOT_OK(struct_builder->Append()); - for (size_t i = 0; i < projection.children.size(); ++i) { - ICEBERG_RETURN_UNEXPECTED(AppendDefaultToBuilder( - projection.children[i], struct_builder->field_builder(static_cast(i)))); - } - return {}; - } // Avro projections carry a single attributes type, so once one is attached it is an // AvroExtraAttributes; use checked_cast instead of a per-row dynamic_cast. if (projection.attributes != nullptr) { diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 9de45a39d8..d64857fa14 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -30,7 +30,6 @@ #include "iceberg/data/delete_filter.h" #include "iceberg/file_reader.h" #include "iceberg/manifest/manifest_entry.h" -#include "iceberg/metadata_columns.h" #include "iceberg/partition_spec.h" #include "iceberg/schema.h" #include "iceberg/schema_internal.h" @@ -42,52 +41,33 @@ namespace iceberg { namespace { -Result ConstantsOf(const FileScanTask& task, const Schema& projection) { +Result ConstantsOf(const FileScanTask& task) { const auto& file = *task.data_file(); const auto& spec = task.spec(); - ICEBERG_ASSIGN_OR_RAISE(auto partition_field, - projection.FindFieldById(MetadataColumns::kPartitionColumnId)); - ICEBERG_ASSIGN_OR_RAISE(auto spec_field, - projection.FindFieldById(MetadataColumns::kSpecIdColumnId)); ColumnConstants constants; - auto spec_id = file.partition_spec_id; - if (spec) { - if (spec_id.has_value() && *spec_id != spec->spec_id()) { - return InvalidArgument("Data file spec ID {} does not match task spec ID {}", - *spec_id, spec->spec_id()); + if (!spec) { + if (file.partition.num_fields() != 0) { + return InvalidArgument("A partition spec is required to read partition values"); } - spec_id = spec->spec_id(); - if (file.partition.num_fields() != spec->fields().size()) { - return InvalidArgument( - "Partition value count {} does not match spec field count {}", - file.partition.num_fields(), spec->fields().size()); - } - StructConstant partition; - for (size_t i = 0; i < spec->fields().size(); ++i) { - const auto& field = spec->fields()[i]; - const auto& value = file.partition.values()[i]; - if (field.transform() == nullptr) { - return InvalidArgument("Missing transform for partition field {}", - field.field_id()); - } - if (field.transform()->transform_type() == TransformType::kIdentity) { - constants.emplace(field.source_id(), value); - } - if (partition_field.has_value()) { - partition.emplace(field.field_id(), value); - } - } - if (partition_field.has_value()) { - constants.emplace(MetadataColumns::kPartitionColumnId, std::move(partition)); - } - } else if (file.partition.num_fields() != 0 || partition_field.has_value()) { - return InvalidArgument("A partition spec is required to read partition values"); + return constants; + } + if (file.partition_spec_id.has_value() && *file.partition_spec_id != spec->spec_id()) { + return InvalidArgument("Data file spec ID {} does not match task spec ID {}", + *file.partition_spec_id, spec->spec_id()); } - if (spec_field.has_value()) { - if (!spec_id.has_value()) { - return InvalidArgument("A partition spec ID is required to read _spec_id"); + if (file.partition.num_fields() != spec->fields().size()) { + return InvalidArgument("Partition value count {} does not match spec field count {}", + file.partition.num_fields(), spec->fields().size()); + } + for (size_t i = 0; i < spec->fields().size(); ++i) { + const auto& field = spec->fields()[i]; + if (field.transform() == nullptr) { + return InvalidArgument("Missing transform for partition field {}", + field.field_id()); + } + if (field.transform()->transform_type() == TransformType::kIdentity) { + constants.emplace(field.source_id(), file.partition.values()[i]); } - constants.emplace(MetadataColumns::kSpecIdColumnId, Literal::Int(*spec_id)); } return constants; } @@ -98,7 +78,7 @@ Result MakeReaderOptions( std::shared_ptr name_mapping, ReaderProperties properties, std::optional first_row_id, std::optional data_sequence_number) { const auto& data_file = *task.data_file(); - ICEBERG_ASSIGN_OR_RAISE(auto constants, ConstantsOf(task, *projection)); + ICEBERG_ASSIGN_OR_RAISE(auto constants, ConstantsOf(task)); return ReaderOptions{ .path = data_file.file_path, .length = static_cast(data_file.file_size_in_bytes), diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index 69a8cdf8a8..6b155cf139 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -27,7 +27,6 @@ #include #include #include -#include #include "iceberg/arrow_c_data.h" #include "iceberg/expression/literal.h" @@ -95,13 +94,8 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase { const std::unordered_map& properties); }; -/// A non-null struct constant, keyed by child field ID. Children are primitive -/// literals, including typed nulls. Projected children absent from the map are null. -using StructConstant = std::unordered_map; - -/// A file-level primitive or struct constant, independent of the file format. -using ColumnConstant = std::variant; -using ColumnConstants = std::unordered_map; +/// File-level primitive values keyed by source field ID, including typed nulls. +using ColumnConstants = std::unordered_map; /// \brief Options for creating a reader. struct ICEBERG_EXPORT ReaderOptions { diff --git a/src/iceberg/parquet/parquet_data_util.cc b/src/iceberg/parquet/parquet_data_util.cc index dc43c5358e..fbb8e90088 100644 --- a/src/iceberg/parquet/parquet_data_util.cc +++ b/src/iceberg/parquet/parquet_data_util.cc @@ -251,18 +251,6 @@ Result> ProjectStructArray( ICEBERG_ASSIGN_OR_RAISE( projected_array, MakeNullArray(output_arrow_type, struct_array->length(), pool)); - } else if (field_projection.kind == FieldProjection::Kind::kConstant && - projected_field.type()->type_id() == TypeId::kStruct) { - // An empty source struct supplies the batch length; children come entirely - // from the constant projection and retain the requested field IDs/order. - auto source = std::make_shared<::arrow::StructArray>( - ::arrow::struct_({}), struct_array->length(), ::arrow::ArrayVector{}); - ICEBERG_ASSIGN_OR_RAISE( - projected_array, - ProjectNestedArray( - source, output_arrow_type, - internal::checked_cast(*projected_field.type()), - field_projection.children, metadata_context, row_positions, pool)); } else if (field_projection.kind == FieldProjection::Kind::kDefault || field_projection.kind == FieldProjection::Kind::kConstant) { ICEBERG_ASSIGN_OR_RAISE( diff --git a/src/iceberg/schema_util.cc b/src/iceberg/schema_util.cc index 7d22d002cb..fb97e9bf1e 100644 --- a/src/iceberg/schema_util.cc +++ b/src/iceberg/schema_util.cc @@ -195,49 +195,19 @@ Result ProjectNested(const Type& expected_type, const Type& sou } // namespace Result ProjectConstant(const SchemaField& field, - const ColumnConstant& constant) { - FieldProjection projection{.kind = FieldProjection::Kind::kConstant}; - if (const auto* literal = std::get_if(&constant)) { - if (!field.type()->is_primitive() || literal->type() == nullptr || - literal->IsAboveMax() || literal->IsBelowMin()) { - return InvalidSchema("Invalid constant for field {}", field.field_id()); - } - ICEBERG_RETURN_UNEXPECTED(ValidateSchemaEvolution(*field.type(), *literal->type())); - if (literal->IsNull() && !field.optional()) { - return InvalidSchema("Null constant for required field {}", field.field_id()); - } - if (literal->IsNull()) { - projection.kind = FieldProjection::Kind::kNull; - return projection; - } - projection.from = *literal; - return projection; - } - - if (field.type()->type_id() != TypeId::kStruct) { - return InvalidSchema("Struct constant requires a struct field: {}", field.field_id()); + const Literal& literal) { + if (!field.type()->is_primitive() || literal.type() == nullptr || + literal.IsAboveMax() || literal.IsBelowMin()) { + return InvalidSchema("Invalid constant for field {}", field.field_id()); } - const auto& fields = internal::checked_cast(*field.type()).fields(); - const auto& values = std::get(constant); - if (fields.empty() && field.optional()) { - projection.kind = FieldProjection::Kind::kNull; - return projection; + ICEBERG_RETURN_UNEXPECTED(ValidateSchemaEvolution(*field.type(), *literal.type())); + if (literal.IsNull() && !field.optional()) { + return InvalidSchema("Null constant for required field {}", field.field_id()); } - for (const auto& child : fields) { - if (!child.type()->is_primitive()) { - return InvalidSchema("Struct constants require primitive children: {}", - child.field_id()); - } - if (const auto it = values.find(child.field_id()); it != values.end()) { - ICEBERG_ASSIGN_OR_RAISE(auto child_projection, ProjectConstant(child, it->second)); - projection.children.push_back(std::move(child_projection)); - } else if (child.optional()) { - projection.children.push_back({.kind = FieldProjection::Kind::kNull}); - } else { - return InvalidSchema("Missing required constant field {}", child.field_id()); - } + if (literal.IsNull()) { + return FieldProjection{.kind = FieldProjection::Kind::kNull}; } - return projection; + return FieldProjection{.kind = FieldProjection::Kind::kConstant, .from = literal}; } Result Project(const Schema& expected_schema, diff --git a/src/iceberg/schema_util.h b/src/iceberg/schema_util.h index 0c96795979..55c726c23a 100644 --- a/src/iceberg/schema_util.h +++ b/src/iceberg/schema_util.h @@ -96,9 +96,9 @@ ICEBERG_EXPORT Result Project(const Schema& expected_schema, const Schema& source_schema, bool prune_source); -/// Build a typed constant projection, matching struct children by field ID. +/// Build a primitive constant projection, validating type promotion and nullability. ICEBERG_EXPORT Result ProjectConstant(const SchemaField& field, - const ColumnConstant& constant); + const Literal& literal); ICEBERG_EXPORT std::string_view ToString(FieldProjection::Kind kind); ICEBERG_EXPORT std::string ToString(const FieldProjection& projection); diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc index 31c35aafb6..1796c53d6e 100644 --- a/src/iceberg/test/avro_test.cc +++ b/src/iceberg/test/avro_test.cc @@ -483,26 +483,19 @@ class AvroReaderParameterizedTest : public AvroReaderTest, TEST_P(AvroReaderParameterizedTest, ReadsPartitionConstants) { CreateSimpleAvroFile(); - auto schema = std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "id", int32()), - SchemaField::MakeRequired(3, "region", string()), - SchemaField::MakeOptional(4, "missing", string()), - SchemaField::MakeOptional( - MetadataColumns::kPartitionColumnId, "_partition", - struct_({SchemaField::MakeOptional(1001, "bucket", int64()), - SchemaField::MakeOptional(1002, "added", string())}))}); + auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(3, "region", string()), + SchemaField::MakeOptional(4, "missing", string())}); ReaderOptions options{.path = temp_avro_file_, .io = file_io_, .projection = schema}; options.properties.Set(ReaderProperties::kAvroSkipDatum, skip_datum_); options.constants = { - {1, Literal::Int(999)}, - {3, Literal::String("CN")}, - {4, Literal::Null(string())}, - {MetadataColumns::kPartitionColumnId, StructConstant{{1001, Literal::Int(7)}}}}; + {1, Literal::Int(999)}, {3, Literal::String("CN")}, {4, Literal::Null(string())}}; ICEBERG_UNWRAP_OR_FAIL(auto reader, ReaderFactoryRegistry::Open(FileFormatType::kAvro, options)); - VerifyNextBatch(*reader, R"([[999,"CN",null,[7,null]], - [999,"CN",null,[7,null]], - [999,"CN",null,[7,null]]])"); + VerifyNextBatch(*reader, R"([[999,"CN",null], + [999,"CN",null], + [999,"CN",null]])"); VerifyExhausted(*reader); } diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index f51cb4926e..717ff0e31c 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -474,33 +474,6 @@ TEST_F(FileScanTaskReaderTest, IdentityConstantsApplyBeforeEqualityDeletes) { EXPECT_EQ(table->num_rows(), 0); } -TEST_F(FileScanTaskReaderTest, PartitionMetadataMatchesHistoricalFieldIds) { - ICEBERG_UNWRAP_OR_FAIL( - auto file, MakeDataFile(table_schema_, - R"([[1, "physical", "blue"], [2, "physical", "red"]])", 2)); - ICEBERG_UNWRAP_OR_FAIL( - auto spec, PartitionSpec::Make(4, {{1, 1000, "bucket", Transform::Bucket(16)}, - {2, 1001, "name", Transform::Identity()}})); - file->partition_spec_id = 4; - file->partition = PartitionValues({Literal::Int(5), Literal::String("constant")}); - auto projection = std::make_shared(std::vector{ - SchemaField::MakeOptional(2, "renamed", string()), MetadataColumns::kSpecId, - SchemaField::MakeOptional( - MetadataColumns::kPartitionColumnId, "_partition", - struct_({SchemaField::MakeOptional(1001, "region", string()), - SchemaField::MakeOptional(1002, "added", int32()), - SchemaField::MakeOptional(1000, "bucket", int64())}))}); - FileScanTask task(file, {}, nullptr, std::move(spec)); - ICEBERG_UNWRAP_OR_FAIL(auto reader, - FileScanTaskReader::Make({.io = file_io_, - .table_schema = table_schema_, - .schemas = {table_schema_}, - .projected_schema = projection})); - ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); - VerifyStream(&stream, R"([["constant", 4, ["constant", null, 5]], - ["constant", 4, ["constant", null, 5]]])"); -} - TEST_F(FileScanTaskReaderTest, NullIdentityOverridesInitialDefault) { auto physical = std::make_shared( std::vector{SchemaField::MakeRequired(1, "id", int32())}); From d2837d8cb28ddcbeb1769f5f37b900da6bec7930 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Thu, 29 Oct 2026 11:03:25 +0800 Subject: [PATCH 4/9] fix(read): validate constant types and cover partition evolution --- src/iceberg/avro/avro_schema_util_internal.h | 2 +- .../parquet/parquet_schema_util_internal.h | 2 +- src/iceberg/schema_util.cc | 4 +- .../test/file_scan_task_reader_test.cc | 69 +++++++++++++++++++ src/iceberg/test/schema_util_test.cc | 6 ++ 5 files changed, 79 insertions(+), 4 deletions(-) diff --git a/src/iceberg/avro/avro_schema_util_internal.h b/src/iceberg/avro/avro_schema_util_internal.h index 7cc0012b5a..02ced14441 100644 --- a/src/iceberg/avro/avro_schema_util_internal.h +++ b/src/iceberg/avro/avro_schema_util_internal.h @@ -142,7 +142,7 @@ class HasIdVisitor { /// \param expected_schema The Iceberg Schema that defines the expected structure. /// \param avro_node The Avro node to read data from. /// \param prune_source Whether the source schema can be pruned. -/// \param constants File-level values used for fields absent from the file. +/// \param constants File-level values that override physical fields and defaults. /// \return The schema projection result. Result Project(const Schema& expected_schema, const ::avro::NodePtr& avro_node, bool prune_source, diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h index bc350e11fd..4ced76a91c 100644 --- a/src/iceberg/parquet/parquet_schema_util_internal.h +++ b/src/iceberg/parquet/parquet_schema_util_internal.h @@ -47,7 +47,7 @@ struct ParquetExtraAttributes : public FieldProjection::ExtraAttributes { /// /// \param expected_schema The Iceberg Schema that defines the expected structure. /// \param parquet_schema The Parquet schema to read data from. -/// \param constants File-level values used for fields absent from the file. +/// \param constants File-level values that override physical fields and defaults. /// \return The schema projection result with column indices of projected Parquet columns /// specified via ParquetExtraAttributes. Result Project(const Schema& expected_schema, diff --git a/src/iceberg/schema_util.cc b/src/iceberg/schema_util.cc index fb97e9bf1e..4eb105fd14 100644 --- a/src/iceberg/schema_util.cc +++ b/src/iceberg/schema_util.cc @@ -196,8 +196,8 @@ Result ProjectNested(const Type& expected_type, const Type& sou Result ProjectConstant(const SchemaField& field, const Literal& literal) { - if (!field.type()->is_primitive() || literal.type() == nullptr || - literal.IsAboveMax() || literal.IsBelowMin()) { + if (field.type() == nullptr || !field.type()->is_primitive() || + literal.type() == nullptr || literal.IsAboveMax() || literal.IsBelowMin()) { return InvalidSchema("Invalid constant for field {}", field.field_id()); } ICEBERG_RETURN_UNEXPECTED(ValidateSchemaEvolution(*field.type(), *literal.type())); diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index 717ff0e31c..6bf24f3726 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -494,6 +494,75 @@ TEST_F(FileScanTaskReaderTest, NullIdentityOverridesInitialDefault) { VerifyStream(&stream, "[[null], [null], [null]]"); } +TEST_F(FileScanTaskReaderTest, ReadsEvolvedSpecsWithoutLeakingConstantsBetweenFiles) { + ICEBERG_UNWRAP_OR_FAIL( + auto identity, PartitionSpec::Make(7, {{2, 1000, "name", Transform::Identity()}})); + ICEBERG_UNWRAP_OR_FAIL(auto unpartitioned, PartitionSpec::Make(8, {})); + ICEBERG_UNWRAP_OR_FAIL( + auto bucket, + PartitionSpec::Make(9, {{2, 1001, "name_bucket", Transform::Bucket(16)}})); + ICEBERG_UNWRAP_OR_FAIL( + auto reordered, + PartitionSpec::Make(10, {{3, 1002, "category", Transform::Identity()}, + {2, 1000, "renamed_name", Transform::Identity()}})); + struct Case { + std::shared_ptr spec; + PartitionValues partition; + std::string expected; + }; + auto identity_spec = std::shared_ptr(std::move(identity)); + const std::vector cases{ + {identity_spec, PartitionValues(Literal::String("CN")), + R"([[1, "CN"], [2, "CN"], [3, "CN"]])"}, + {std::move(unpartitioned), PartitionValues(), + R"([[1, "file"], [2, "file"], [3, "file"]])"}, + {std::move(bucket), PartitionValues(Literal::Int(3)), + R"([[1, "file"], [2, "file"], [3, "file"]])"}, + {std::move(reordered), + PartitionValues({Literal::String("retired"), Literal::String("US")}), + R"([[1, "US"], [2, "US"], [3, "US"]])"}, + {identity_spec, PartitionValues(Literal::Null(string())), + "[[1, null], [2, null], [3, null]]"}}; + // The current schema omits category, but historical task schemas retain its type. + ICEBERG_UNWRAP_OR_FAIL( + auto reader, FileScanTaskReader::Make({.io = file_io_, + .table_schema = projected_schema_, + .schemas = {table_schema_}, + .projected_schema = projected_schema_})); + std::vector> readers; + for (const auto& test : cases) { + SCOPED_TRACE(test.spec->spec_id()); + ICEBERG_UNWRAP_OR_FAIL( + auto file, + MakeDataFile(projected_schema_, R"([[1, "file"], [2, "file"], [3, "file"]])")); + file->partition_spec_id = test.spec->spec_id(); + file->partition = test.partition; + ICEBERG_UNWRAP_OR_FAIL(auto json, + ToJson(FileScanTask(file, {}, nullptr, test.spec), + {{test.spec->spec_id(), test.spec}}, *table_schema_)); + ICEBERG_UNWRAP_OR_FAIL(auto task, FileScanTaskFromJson(json)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*task)); + auto imported = ::arrow::ImportRecordBatchReader(&stream); + ASSERT_TRUE(imported.ok()) << imported.status().ToString(); + readers.push_back(imported.ValueOrDie()); + } + // Keep all streams alive to detect accidental sharing of per-file constants. + for (size_t i = 0; i < cases.size(); ++i) { + SCOPED_TRACE(i); + auto next = readers[i]->Next(); + ASSERT_TRUE(next.ok()) << next.status().ToString(); + auto batch = next.ValueOrDie(); + ASSERT_NE(batch, nullptr); + ASSERT_TRUE(batch->ValidateFull().ok()); + ICEBERG_UNWRAP_OR_FAIL(auto expected, + MakeRecordBatch(batch->schema(), cases[i].expected)); + EXPECT_TRUE(batch->Equals(*expected)); + next = readers[i]->Next(); + ASSERT_TRUE(next.ok()) << next.status().ToString(); + EXPECT_EQ(next.ValueOrDie(), nullptr); + } +} + TEST_F(FileScanTaskReaderTest, RejectsMissingOrInconsistentPartitionContext) { ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(table_schema_, R"([[1, "Foo", "blue"]])", 1)); diff --git a/src/iceberg/test/schema_util_test.cc b/src/iceberg/test/schema_util_test.cc index 9a3eff8870..9deb935432 100644 --- a/src/iceberg/test/schema_util_test.cc +++ b/src/iceberg/test/schema_util_test.cc @@ -101,6 +101,12 @@ std::shared_ptr CreateMapOfList() { } // namespace +TEST(SchemaUtilTest, ProjectConstantRejectsMissingFieldType) { + auto field = SchemaField::MakeOptional(1, "invalid", nullptr); + ASSERT_THAT(ProjectConstant(field, Literal::Long(7)), + IsError(ErrorKind::kInvalidSchema)); +} + TEST(SchemaUtilTest, ProjectIdenticalSchemas) { Schema schema = CreateFlatSchema(); From 2fa71878303f8e696e4498f4c46dbf26896faf08 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Thu, 29 Oct 2026 12:00:33 +0800 Subject: [PATCH 5/9] fix(test): use designated initializers for partition evolution cases --- .../test/file_scan_task_reader_test.cc | 26 +++++++++++-------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index 6bf24f3726..03ddcdf0c3 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -512,17 +512,21 @@ TEST_F(FileScanTaskReaderTest, ReadsEvolvedSpecsWithoutLeakingConstantsBetweenFi }; auto identity_spec = std::shared_ptr(std::move(identity)); const std::vector cases{ - {identity_spec, PartitionValues(Literal::String("CN")), - R"([[1, "CN"], [2, "CN"], [3, "CN"]])"}, - {std::move(unpartitioned), PartitionValues(), - R"([[1, "file"], [2, "file"], [3, "file"]])"}, - {std::move(bucket), PartitionValues(Literal::Int(3)), - R"([[1, "file"], [2, "file"], [3, "file"]])"}, - {std::move(reordered), - PartitionValues({Literal::String("retired"), Literal::String("US")}), - R"([[1, "US"], [2, "US"], [3, "US"]])"}, - {identity_spec, PartitionValues(Literal::Null(string())), - "[[1, null], [2, null], [3, null]]"}}; + {.spec = identity_spec, + .partition = PartitionValues(Literal::String("CN")), + .expected = R"([[1, "CN"], [2, "CN"], [3, "CN"]])"}, + {.spec = std::move(unpartitioned), + .partition = PartitionValues(), + .expected = R"([[1, "file"], [2, "file"], [3, "file"]])"}, + {.spec = std::move(bucket), + .partition = PartitionValues(Literal::Int(3)), + .expected = R"([[1, "file"], [2, "file"], [3, "file"]])"}, + {.spec = std::move(reordered), + .partition = PartitionValues({Literal::String("retired"), Literal::String("US")}), + .expected = R"([[1, "US"], [2, "US"], [3, "US"]])"}, + {.spec = identity_spec, + .partition = PartitionValues(Literal::Null(string())), + .expected = "[[1, null], [2, null], [3, null]]"}}; // The current schema omits category, but historical task schemas retain its type. ICEBERG_UNWRAP_OR_FAIL( auto reader, FileScanTaskReader::Make({.io = file_io_, From 284a2980aacef2dda7429b2b9a33d40b88241c21 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Thu, 29 Oct 2026 16:44:31 +0800 Subject: [PATCH 6/9] refactor(read): trust constants during schema projection --- src/iceberg/avro/avro_schema_util.cc | 2 -- src/iceberg/parquet/parquet_schema_util.cc | 1 - src/iceberg/schema_util.cc | 14 -------------- src/iceberg/schema_util_internal.h | 7 ------- src/iceberg/test/parquet_test.cc | 10 ++-------- src/iceberg/test/schema_util_test.cc | 16 +--------------- 6 files changed, 3 insertions(+), 47 deletions(-) diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc index 4e2e84d8ce..e58d1f47a6 100644 --- a/src/iceberg/avro/avro_schema_util.cc +++ b/src/iceberg/avro/avro_schema_util.cc @@ -753,8 +753,6 @@ Result ProjectStruct( FieldProjection child_projection; if (auto constant = id_to_constant.find(field_id); constant != id_to_constant.end()) { - ICEBERG_RETURN_UNEXPECTED( - internal::ValidateConstant(expected_field, constant->second)); child_projection.kind = constant->second.IsNull() ? FieldProjection::Kind::kNull : FieldProjection::Kind::kConstant; diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 4a3f622a74..34ae158cd6 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -591,7 +591,6 @@ class ProjectionBuilder { if (auto constant = id_to_constant_.find(field_id); constant != id_to_constant_.end()) { - ICEBERG_RETURN_UNEXPECTED(internal::ValidateConstant(field, constant->second)); child_projection.kind = constant->second.IsNull() ? FieldProjection::Kind::kNull : FieldProjection::Kind::kConstant; diff --git a/src/iceberg/schema_util.cc b/src/iceberg/schema_util.cc index 54dba648ee..e9efbc50ea 100644 --- a/src/iceberg/schema_util.cc +++ b/src/iceberg/schema_util.cc @@ -170,8 +170,6 @@ Result ProjectNested( FieldProjection child_projection; if (auto constant = id_to_constant.find(field_id); constant != id_to_constant.end()) { - ICEBERG_RETURN_UNEXPECTED( - internal::ValidateConstant(expected_field, constant->second)); child_projection.kind = constant->second.IsNull() ? FieldProjection::Kind::kNull : FieldProjection::Kind::kConstant; @@ -205,18 +203,6 @@ Result ProjectNested( } // namespace -Status internal::ValidateConstant(const SchemaField& field, const Literal& literal) { - if (field.type() == nullptr || !field.type()->is_primitive() || - literal.type() == nullptr || literal.IsAboveMax() || literal.IsBelowMin()) { - return InvalidSchema("Invalid constant for field {}", field.field_id()); - } - ICEBERG_RETURN_UNEXPECTED(ValidateSchemaEvolution(*field.type(), *literal.type())); - if (literal.IsNull() && !field.optional()) { - return InvalidSchema("Null constant for required field {}", field.field_id()); - } - return {}; -} - Result Project( const Schema& expected_schema, const Schema& source_schema, bool prune_source, const std::unordered_map& id_to_constant) { diff --git a/src/iceberg/schema_util_internal.h b/src/iceberg/schema_util_internal.h index eb527777c3..e0faf4f6ea 100644 --- a/src/iceberg/schema_util_internal.h +++ b/src/iceberg/schema_util_internal.h @@ -25,13 +25,6 @@ namespace iceberg { -namespace internal { - -// Validate a primitive constant before substituting it for a projected field. -Status ValidateConstant(const SchemaField& field, const Literal& literal); - -} // namespace internal - // Fix `from` field of `FieldProjection` to use pruned field index. inline void PruneFieldProjection(FieldProjection& field_projection) { std::map local_index_to_pruned_index; diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc index 8665642c25..7059ff61c2 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -553,17 +553,11 @@ class ParquetReaderTest : public TempFileTestBase { std::string temp_parquet_file_; }; -TEST_F(ParquetReaderTest, ConstantsValidateTypesAndRequiredNulls) { +TEST_F(ParquetReaderTest, PromotesPartitionConstant) { CreateSimpleParquetFile(); auto schema = std::make_shared( std::vector{SchemaField::MakeRequired(3, "constant", int64())}); - for (const auto& value : {Literal::String("7"), Literal::Null(int64())}) { - ReaderOptions options{ - .path = temp_parquet_file_, .io = file_io_, .projection = schema}; - options.id_to_constant.emplace(3, value); - EXPECT_FALSE( - ReaderFactoryRegistry::Open(FileFormatType::kParquet, options).has_value()); - } + ReaderOptions options{.path = temp_parquet_file_, .io = file_io_, .projection = schema}; options.id_to_constant.emplace(3, Literal::Int(7)); ICEBERG_UNWRAP_OR_FAIL(auto reader, diff --git a/src/iceberg/test/schema_util_test.cc b/src/iceberg/test/schema_util_test.cc index 99e04d2809..a771d21bd4 100644 --- a/src/iceberg/test/schema_util_test.cc +++ b/src/iceberg/test/schema_util_test.cc @@ -101,14 +101,6 @@ std::shared_ptr CreateMapOfList() { } // namespace -TEST(SchemaUtilTest, ProjectConstantRejectsMissingFieldType) { - auto field = SchemaField::MakeOptional(1, "invalid", nullptr); - auto projection = Project(Schema({field}), Schema({}), /*prune_source=*/true, - {{1, Literal::Long(7)}}); - ASSERT_FALSE(projection.has_value()); - EXPECT_EQ(projection.error().kind, ErrorKind::kInvalidSchema); -} - TEST(SchemaUtilTest, ProjectConstantsOverridePhysicalFieldsAndPruneIndices) { Schema schema({SchemaField::MakeRequired(1, "region", string()), SchemaField::MakeRequired(2, "id", int64())}); @@ -122,7 +114,7 @@ TEST(SchemaUtilTest, ProjectConstantsOverridePhysicalFieldsAndPruneIndices) { } } -TEST(SchemaUtilTest, ProjectConstantsFillMissingFieldsAndValidateNulls) { +TEST(SchemaUtilTest, ProjectConstantsFillMissingFields) { Schema expected({SchemaField::MakeRequired(1, "id", int64()), SchemaField::MakeOptional(2, "region", string())}); auto projection = Project(expected, Schema({}), /*prune_source=*/true, @@ -131,12 +123,6 @@ TEST(SchemaUtilTest, ProjectConstantsFillMissingFieldsAndValidateNulls) { ASSERT_EQ(projection->fields.size(), 2); EXPECT_EQ(projection->fields[0].kind, FieldProjection::Kind::kConstant); EXPECT_EQ(projection->fields[1].kind, FieldProjection::Kind::kNull); - EXPECT_FALSE( - Project(expected, Schema({}), /*prune_source=*/true, {{1, Literal::Null(int64())}}) - .has_value()); - EXPECT_FALSE(Project(expected, Schema({}), /*prune_source=*/true, - {{1, Literal::String("invalid")}}) - .has_value()); } TEST(SchemaUtilTest, ProjectConstantsIntoExistingStruct) { From ea8e8df7aa7c945aa7e23efbc8cd745dc273a164 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Thu, 29 Oct 2026 16:52:16 +0800 Subject: [PATCH 7/9] refactor(read): trust task partition context --- src/iceberg/data/file_scan_task_reader.cc | 8 -------- src/iceberg/test/file_scan_task_reader_test.cc | 10 +--------- 2 files changed, 1 insertion(+), 17 deletions(-) diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 64137688c1..54307950e2 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -50,14 +50,6 @@ Result> ConstantsOf(const FileScanTask& tas } return id_to_constant; } - if (file.partition_spec_id.has_value() && *file.partition_spec_id != spec->spec_id()) { - return InvalidArgument("Data file spec ID {} does not match task spec ID {}", - *file.partition_spec_id, spec->spec_id()); - } - if (file.partition.num_fields() != spec->fields().size()) { - return InvalidArgument("Partition value count {} does not match spec field count {}", - file.partition.num_fields(), spec->fields().size()); - } for (size_t i = 0; i < spec->fields().size(); ++i) { const auto& field = spec->fields()[i]; if (field.transform() == nullptr) { diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index 107477d455..d959aa750d 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -570,7 +570,7 @@ TEST_F(FileScanTaskReaderTest, ReadsEvolvedSpecsWithoutLeakingConstantsBetweenFi } } -TEST_F(FileScanTaskReaderTest, RejectsMissingOrInconsistentPartitionContext) { +TEST_F(FileScanTaskReaderTest, RejectsMissingPartitionSpec) { ICEBERG_UNWRAP_OR_FAIL(auto file, MakeDataFile(table_schema_, R"([[1, "Foo", "blue"]])", 1)); ICEBERG_UNWRAP_OR_FAIL( @@ -580,14 +580,6 @@ TEST_F(FileScanTaskReaderTest, RejectsMissingOrInconsistentPartitionContext) { .projected_schema = projected_schema_})); file->partition = PartitionValues({Literal::String("CN")}); EXPECT_FALSE(reader->Open(FileScanTask(file)).has_value()); - ICEBERG_UNWRAP_OR_FAIL( - auto spec, PartitionSpec::Make(3, {{2, 1000, "name", Transform::Identity()}})); - auto shared_spec = std::shared_ptr(std::move(spec)); - file->partition_spec_id = 9; - EXPECT_FALSE(reader->Open(FileScanTask(file, {}, nullptr, shared_spec)).has_value()); - file->partition_spec_id = 3; - file->partition = PartitionValues{}; - EXPECT_FALSE(reader->Open(FileScanTask(file, {}, nullptr, shared_spec)).has_value()); } TEST_F(FileScanTaskReaderTest, OpenWithoutDeletesReadsProjectedSchema) { From e7ab7f49113a48c4270b2fa185f077e2a80e5c22 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Thu, 29 Oct 2026 17:11:16 +0800 Subject: [PATCH 8/9] refactor(read): preserve row group filter interface --- .../parquet/parquet_metrics_row_group_filter.cc | 6 ++---- .../parquet_metrics_row_group_filter_internal.h | 4 +--- src/iceberg/parquet/parquet_reader.cc | 5 ++--- src/iceberg/test/parquet_test.cc | 15 --------------- 4 files changed, 5 insertions(+), 25 deletions(-) diff --git a/src/iceberg/parquet/parquet_metrics_row_group_filter.cc b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc index 3167bdc802..eb543e8723 100644 --- a/src/iceberg/parquet/parquet_metrics_row_group_filter.cc +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc @@ -291,14 +291,12 @@ class MetricsVisitor : public BoundVisitor { Result> ParquetMetricsRowGroupFilter::Make( const std::shared_ptr& filter, - const ::parquet::SchemaDescriptor& file_schema, - const std::unordered_map& id_to_constant) { + const ::parquet::SchemaDescriptor& file_schema) { auto result = std::unique_ptr(new ParquetMetricsRowGroupFilter()); for (int i = 0; i < file_schema.num_columns(); ++i) { auto id = file_schema.Column(i)->schema_node()->field_id(); - // Footer statistics describe physical values, not substituted constants. - if (id >= 0 && !id_to_constant.contains(id)) { + if (id >= 0) { result->field_id_to_column_idx_.emplace(id, i); } } diff --git a/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h index b28f7bf4ac..5d80fe3bfe 100644 --- a/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h @@ -25,7 +25,6 @@ #include -#include "iceberg/expression/literal.h" #include "iceberg/iceberg_bundle_export.h" #include "iceberg/result.h" #include "iceberg/type_fwd.h" @@ -39,8 +38,7 @@ class ICEBERG_BUNDLE_EXPORT ParquetMetricsRowGroupFilter { public: static Result> Make( const std::shared_ptr& filter, - const ::parquet::SchemaDescriptor& file_schema, - const std::unordered_map& id_to_constant = {}); + const ::parquet::SchemaDescriptor& file_schema); Result ShouldRead(const ::parquet::arrow::SchemaManifest& manifest, const ::parquet::RowGroupMetaData& row_group) const; diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index e9e3afb093..85f0c9b38d 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -318,9 +318,8 @@ class ParquetReader::Impl { ICEBERG_ASSIGN_OR_RAISE( filter, Binder::Bind(*options.projection, filter, options.case_sensitive)); } - ICEBERG_ASSIGN_OR_RAISE( - stats_filter_, ParquetMetricsRowGroupFilter::Make( - filter, *reader_->manifest().descr, options.id_to_constant)); + ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make( + filter, *reader_->manifest().descr)); } ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options)); diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc index 7059ff61c2..35b68d113f 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -566,21 +566,6 @@ TEST_F(ParquetReaderTest, PromotesPartitionConstant) { VerifyExhausted(*reader); } -TEST_F(ParquetReaderTest, ConstantsDoNotUsePhysicalColumnStatistics) { - CreateSimpleParquetFile(); - auto schema = std::make_shared( - std::vector{SchemaField::MakeRequired(1, "id", int32())}); - ReaderOptions options{.path = temp_parquet_file_, - .io = file_io_, - .projection = schema, - .filter = Expressions::Equal("id", Literal::Int(7))}; - options.id_to_constant.emplace(1, Literal::Int(7)); - ICEBERG_UNWRAP_OR_FAIL(auto reader, - ReaderFactoryRegistry::Open(FileFormatType::kParquet, options)); - VerifyNextBatch(*reader, "[[7], [7], [7]]"); - VerifyExhausted(*reader); -} - TEST_F(ParquetReaderTest, ReadTwoFields) { CreateSimpleParquetFile(); From 43bfc5ff3f632211e8d744978a574ceb54b42440 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Thu, 29 Oct 2026 17:21:15 +0800 Subject: [PATCH 9/9] refactor(read): project null partition values as constants --- src/iceberg/avro/avro_schema_util.cc | 4 +--- src/iceberg/parquet/parquet_schema_util.cc | 4 +--- src/iceberg/schema_util.cc | 4 +--- src/iceberg/test/file_scan_task_reader_test.cc | 3 ++- src/iceberg/test/schema_util_test.cc | 2 +- 5 files changed, 6 insertions(+), 11 deletions(-) diff --git a/src/iceberg/avro/avro_schema_util.cc b/src/iceberg/avro/avro_schema_util.cc index e58d1f47a6..819dbf9757 100644 --- a/src/iceberg/avro/avro_schema_util.cc +++ b/src/iceberg/avro/avro_schema_util.cc @@ -753,9 +753,7 @@ Result ProjectStruct( FieldProjection child_projection; if (auto constant = id_to_constant.find(field_id); constant != id_to_constant.end()) { - child_projection.kind = constant->second.IsNull() - ? FieldProjection::Kind::kNull - : FieldProjection::Kind::kConstant; + child_projection.kind = FieldProjection::Kind::kConstant; child_projection.from = constant->second; } else if (auto iter = node_info_map.find(field_id); iter != node_info_map.cend()) { ICEBERG_ASSIGN_OR_RAISE( diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 34ae158cd6..21e1f4b5c1 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -591,9 +591,7 @@ class ProjectionBuilder { if (auto constant = id_to_constant_.find(field_id); constant != id_to_constant_.end()) { - child_projection.kind = constant->second.IsNull() - ? FieldProjection::Kind::kNull - : FieldProjection::Kind::kConstant; + child_projection.kind = FieldProjection::Kind::kConstant; child_projection.from = constant->second; } else if (auto iter = field_context_map.find(field_id); iter != field_context_map.cend()) { diff --git a/src/iceberg/schema_util.cc b/src/iceberg/schema_util.cc index e9efbc50ea..d6f77e8f90 100644 --- a/src/iceberg/schema_util.cc +++ b/src/iceberg/schema_util.cc @@ -170,9 +170,7 @@ Result ProjectNested( FieldProjection child_projection; if (auto constant = id_to_constant.find(field_id); constant != id_to_constant.end()) { - child_projection.kind = constant->second.IsNull() - ? FieldProjection::Kind::kNull - : FieldProjection::Kind::kConstant; + child_projection.kind = FieldProjection::Kind::kConstant; child_projection.from = constant->second; } else if (auto iter = source_field_map.find(field_id); iter != source_field_map.cend()) { diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index d959aa750d..d32b8a9968 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -497,7 +497,8 @@ TEST_F(FileScanTaskReaderTest, NullIdentityOverridesInitialDefault) { VerifyStream(&stream, "[[null], [null], [null]]"); } -TEST_F(FileScanTaskReaderTest, ReadsEvolvedSpecsWithoutLeakingConstantsBetweenFiles) { +TEST_F(FileScanTaskReaderTest, + ReadsTaskSpecificSpecsWithoutLeakingConstantsBetweenFiles) { ICEBERG_UNWRAP_OR_FAIL( auto identity, PartitionSpec::Make(7, {{2, 1000, "name", Transform::Identity()}})); ICEBERG_UNWRAP_OR_FAIL(auto unpartitioned, PartitionSpec::Make(8, {})); diff --git a/src/iceberg/test/schema_util_test.cc b/src/iceberg/test/schema_util_test.cc index a771d21bd4..aed14b22b2 100644 --- a/src/iceberg/test/schema_util_test.cc +++ b/src/iceberg/test/schema_util_test.cc @@ -122,7 +122,7 @@ TEST(SchemaUtilTest, ProjectConstantsFillMissingFields) { ASSERT_THAT(projection, IsOk()); ASSERT_EQ(projection->fields.size(), 2); EXPECT_EQ(projection->fields[0].kind, FieldProjection::Kind::kConstant); - EXPECT_EQ(projection->fields[1].kind, FieldProjection::Kind::kNull); + EXPECT_EQ(projection->fields[1].kind, FieldProjection::Kind::kConstant); } TEST(SchemaUtilTest, ProjectConstantsIntoExistingStruct) {