From 244a861d85eb73edb8a390edb0e57c5a064a0601 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Wed, 2 Sep 2026 22:04:00 +0200 Subject: [PATCH 1/3] Port of reading first_row_id and last_seq_num from iceberg v3 manifest files --- .../Common/AvroForIcebergDeserializer.cpp | 10 + .../DataLakes/Iceberg/Constant.h | 6 + .../Iceberg/IcebergDataObjectInfo.cpp | 32 +- .../DataLakes/Iceberg/IcebergDataObjectInfo.h | 4 + .../DataLakes/Iceberg/IcebergIterator.cpp | 1 + .../DataLakes/Iceberg/IcebergMetadata.cpp | 1 + .../DataLakes/Iceberg/IcebergMetadata.h | 1 + .../Iceberg/IcebergMetadataFilesCache.h | 2 + .../DataLakes/Iceberg/ManifestFile.h | 4 + .../Iceberg/ManifestFileIterator.cpp | 102 +++++ .../DataLakes/Iceberg/ManifestFileIterator.h | 3 + .../Iceberg/ManifestFilesPruning.cpp | 70 +++- .../DataLakes/Iceberg/ManifestFilesPruning.h | 9 +- .../Iceberg/StatelessMetadataFileGetter.cpp | 12 +- .../StorageObjectStorageSource.cpp | 88 ++++- .../System/StorageSystemIcebergFiles.cpp | 13 +- src/Storages/VirtualColumnUtils.cpp | 79 +++- src/Storages/VirtualColumnUtils.h | 5 + .../test_row_lineage.py | 278 ++++++++++++++ .../test_row_lineage_pruning.py | 352 ++++++++++++++++++ .../04203_system_iceberg_files.reference | 1 + 21 files changed, 1052 insertions(+), 21 deletions(-) create mode 100644 tests/integration/test_storage_iceberg_with_spark/test_row_lineage.py create mode 100644 tests/integration/test_storage_iceberg_with_spark/test_row_lineage_pruning.py diff --git a/src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp b/src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp index 277d04c3a77a..32f566d652ac 100644 --- a/src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp @@ -169,6 +169,13 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE } } + std::optional first_row_id; + if (format_version > 2 && hasPath(c_data_file_first_row_id)) + { + const auto first_row_id_value = getValueFromRowByName(row_index, c_data_file_first_row_id); + if (!first_row_id_value.isNull()) + first_row_id = first_row_id_value.safeGet(); + } const auto file_path_from_metadata = IcebergPathFromMetadata::deserialize( getValueFromRowByName(row_index, c_data_file_file_path, TypeIndex::String).safeGet()); @@ -280,6 +287,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE status, sequence_number, snapshot_id, + first_row_id, partition_key_value, columns_infos, value_for_bounds, @@ -344,6 +352,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE status, sequence_number, snapshot_id, + first_row_id, partition_key_value, columns_infos, value_for_bounds, @@ -377,6 +386,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE status, sequence_number, snapshot_id, + first_row_id, partition_key_value, columns_infos, value_for_bounds, diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h index cd7c9f29d7e3..458f6662a17b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h @@ -93,6 +93,7 @@ DEFINE_ICEBERG_FIELD(replace); DEFINE_ICEBERG_FIELD_ALIAS(format_version, format-version); DEFINE_ICEBERG_FIELD_ALIAS(current_snapshot_id, current-snapshot-id); DEFINE_ICEBERG_FIELD_ALIAS(first_row_id, first-row-id); +DEFINE_ICEBERG_FIELD_ALIAS(manifest_first_row_id, first_row_id); DEFINE_ICEBERG_FIELD_ALIAS(added_rows, added-rows); DEFINE_ICEBERG_FIELD_ALIAS(next_row_id, next-row-id); DEFINE_ICEBERG_FIELD_ALIAS(metadata_snapshot_id, snapshot-id); @@ -182,6 +183,11 @@ DEFINE_ICEBERG_FIELD_COMPOUND(data_file, content_size_in_bytes); DEFINE_ICEBERG_FIELD_COMPOUND(data_file, sort_order_id); DEFINE_ICEBERG_FIELD_COMPOUND(data_file, record_count); DEFINE_ICEBERG_FIELD_COMPOUND(data_file, file_size_in_bytes); +DEFINE_ICEBERG_FIELD_COMPOUND(data_file, key_metadata); +DEFINE_ICEBERG_FIELD_COMPOUND(data_file, first_row_id); + +constexpr Int32 row_id_field_id = 2147483540; +constexpr Int32 last_updated_sequence_number_field_id = 2147483539; /// Fallback defaults for snapshot retention policy when table properties are absent. /// These values follow the Java reference implementation; the Iceberg spec does not diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.cpp index 0fab9d175f99..d0827128be8e 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.cpp @@ -88,7 +88,8 @@ IcebergDataObjectInfo::IcebergDataObjectInfo( /* position_deletes_objects */ {}, /* equality_deletes_objects */ {}, data_manifest_file_entry_->parsed_entry->record_count, - data_manifest_file_entry_->parsed_entry->file_size_in_bytes} + data_manifest_file_entry_->parsed_entry->file_size_in_bytes, + data_manifest_file_entry_->first_row_id} , resolved_storage(std::move(resolved_storage_)) { /// resolved_storage and resolved_key must be provided together or neither must be provided @@ -245,6 +246,19 @@ void IcebergObjectSerializableInfo::serializeForClusterFunctionProtocol(WriteBuf writeVarUInt(0, out); } } + + if (protocol_version >= DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_CDC_READING) + { + if (first_row_id.has_value()) + { + writeVarUInt(1, out); + writeVarUInt(*first_row_id, out); + } + else + { + writeVarUInt(0, out); + } + } } void IcebergObjectSerializableInfo::deserializeForClusterFunctionProtocol(ReadBuffer & in, size_t protocol_version) @@ -336,6 +350,22 @@ void IcebergObjectSerializableInfo::deserializeForClusterFunctionProtocol(ReadBu file_size_in_bytes = std::nullopt; } } + + if (protocol_version >= DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_CDC_READING) + { + size_t has_first_row_id = 0; + readVarUInt(has_first_row_id, in); + if (has_first_row_id) + { + UInt64 value = 0; + readVarUInt(value, in); + first_row_id = value; + } + else + { + first_row_id = std::nullopt; + } + } } void IcebergObjectSerializableInfo::checkVersion(size_t protocol_version) const diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.h index 4b79cdec18b7..dad10dec7b1c 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.h @@ -1,6 +1,8 @@ #pragma once #include "config.h" +#include + #include #include #include @@ -10,6 +12,7 @@ #include #include +#include #include @@ -41,6 +44,7 @@ struct IcebergObjectSerializableInfo std::vector equality_deletes_objects; std::optional record_count; std::optional file_size_in_bytes; + std::optional first_row_id; /// Set to true by the coordinator when the file is outside of the table location bool requires_external_storage = false; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp index b843d87e2868..b9661b7d7d36 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp @@ -198,6 +198,7 @@ std::optional SingleThreadIcebergKeysIterator::ne *persistent_components.schema_processor, manifest_list_entry.added_sequence_number, manifest_list_entry.added_snapshot_id, + manifest_list_entry.first_row_id, local_context, filter_dag, table_snapshot->schema_id); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 24930b88462a..2a3b8ba17c0a 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -1123,6 +1123,7 @@ IcebergFileRecord buildIcebergFileRecord( record.schema_id = processed->resolved_schema_id; record.sequence_number = processed->sequence_number; record.sort_order_id = parsed.sort_order_id; + record.first_row_id = processed->first_row_id; for (const auto & [column_id, info] : parsed.columns_infos) { diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h index b28d476438da..171bc65ecb09 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h @@ -52,6 +52,7 @@ struct IcebergFileRecord std::map column_sizes; std::map value_counts; std::vector equality_ids; + std::optional first_row_id; }; class IcebergMetadata : public IDataLakeMetadata diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadataFilesCache.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadataFilesCache.h index 11b5f7233088..9363e638f727 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadataFilesCache.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadataFilesCache.h @@ -1,5 +1,6 @@ #pragma once #include +#include #include #include #include "config.h" @@ -59,6 +60,7 @@ struct ManifestFileCacheKey Int64 added_sequence_number; Int64 added_snapshot_id; Iceberg::ManifestFileContentType content_type; + std::optional first_row_id; }; using ManifestFileCacheKeys = std::vector; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h index ff5fd50c0884..7ade760686e0 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h @@ -86,6 +86,7 @@ struct ParsedManifestFileEntry : boost::noncopyable ManifestEntryStatus status; std::optional parsed_sequence_number; std::optional parsed_snapshot_id; + std::optional parsed_first_row_id; DB::Row partition_key_value; std::unordered_map columns_infos; @@ -121,6 +122,7 @@ struct ParsedManifestFileEntry : boost::noncopyable ManifestEntryStatus status_, std::optional written_sequence_number_, std::optional written_snapshot_id_, + std::optional written_first_row_id_, DB::Row partition_key_value_, std::unordered_map columns_infos_, std::unordered_map> value_bounds_, @@ -139,6 +141,7 @@ struct ParsedManifestFileEntry : boost::noncopyable , status(status_) , parsed_sequence_number(written_sequence_number_) , parsed_snapshot_id(written_snapshot_id_) + , parsed_first_row_id(written_first_row_id_) , partition_key_value(std::move(partition_key_value_)) , columns_infos(std::move(columns_infos_)) , value_bounds(std::move(value_bounds_)) @@ -164,6 +167,7 @@ struct ProcessedManifestFileEntry Int64 sequence_number; Int32 resolved_schema_id; String manifest_file_path; + std::optional first_row_id; String dumpDeletesMatchingInfo() const; }; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp index 7856444feab0..7d367ac4cdae 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -49,6 +50,83 @@ extern const Event IcebergPartitionPrunedFiles; extern const Event IcebergMinMaxIndexPrunedFiles; }; +namespace +{ + +using namespace DB::Iceberg; + +std::optional getMaterializedRowLineageRange(const ParsedManifestFileEntry & parsed_entry, Int32 field_id) +{ + auto bounds = parsed_entry.value_bounds.find(field_id); + if (bounds == parsed_entry.value_bounds.end()) + return std::nullopt; + + auto column_info = parsed_entry.columns_infos.find(field_id); + if (column_info == parsed_entry.columns_infos.end() || !column_info->second.nulls_count.has_value() + || *column_info->second.nulls_count != 0) + return std::nullopt; + + String left_str; + String right_str; + if (!bounds->second.first.tryGet(left_str) || !bounds->second.second.tryGet(right_str)) + return std::nullopt; + + auto type = std::make_shared(); + auto left = deserializeFieldFromBinaryRepr(left_str, type, true); + auto right = deserializeFieldFromBinaryRepr(right_str, type, false); + if (!left || !right) + return std::nullopt; + + return DB::Range(*left, true, *right, true); +} + +bool isColumnPresenceKnown(const ParsedManifestFileEntry & parsed_entry) +{ + for (const auto & [field_id, column_info] : parsed_entry.columns_infos) + if (column_info.bytes_size.has_value()) + return true; + return false; +} + +void addRowLineageHyperrectangles(std::unordered_map & hyperrectangles, const ProcessedManifestFileEntry & entry) +{ + const auto & parsed_entry = *entry.parsed_entry; + if (!entry.first_row_id.has_value() || parsed_entry.record_count <= 0 || entry.sequence_number < 0) + return; + + const UInt64 inherited_sequence_number = static_cast(entry.sequence_number); + const UInt64 last_inherited_row_id = *entry.first_row_id + static_cast(parsed_entry.record_count) - 1; + const bool column_presence_is_known = isColumnPresenceKnown(parsed_entry); + const bool row_ids_are_readable = Poco::toUpper(parsed_entry.file_format) != "ORC"; + + for (const auto field_id : {row_id_field_id, last_updated_sequence_number_field_id}) + { + const bool is_row_id = field_id == row_id_field_id; + if (is_row_id && !row_ids_are_readable) + continue; + const UInt64 inherited_lower_bound = is_row_id ? *entry.first_row_id : inherited_sequence_number; + const UInt64 inherited_upper_bound = is_row_id ? last_inherited_row_id : inherited_sequence_number; + + if (!parsed_entry.columns_infos.contains(field_id)) + { + if (column_presence_is_known) + { + hyperrectangles.emplace(field_id, DB::Range(inherited_lower_bound, true, inherited_upper_bound, true)); + continue; + } + } + else if (auto range = getMaterializedRowLineageRange(parsed_entry, field_id)) + { + hyperrectangles.emplace(field_id, *range); + continue; + } + + hyperrectangles.emplace(field_id, DB::Range(UInt64(0), true, inherited_upper_bound, true)); + } +} + +} + namespace DB::Iceberg { @@ -118,6 +196,7 @@ std::shared_ptr ManifestFileIterator::create( IcebergSchemaProcessor & schema_processor, Int64 inherited_sequence_number_, Int64 inherited_snapshot_id_, + std::optional inherited_first_row_id_, DB::ContextPtr context_, std::shared_ptr filter_dag_, Int32 table_snapshot_schema_id_) @@ -217,6 +296,7 @@ std::shared_ptr ManifestFileIterator::create( schema_processor, inherited_sequence_number_, inherited_snapshot_id_, + inherited_first_row_id_, context_, manifest_schema_id, std::make_shared(std::move(partition_spec_vec)), @@ -234,6 +314,7 @@ ManifestFileIterator::ManifestFileIterator( IcebergSchemaProcessor & schema_processor, Int64 inherited_sequence_number_, Int64 inherited_snapshot_id_, + std::optional inherited_first_row_id_, DB::ContextPtr context_, Int32 manifest_schema_id_, std::shared_ptr common_partition_specification_, @@ -259,6 +340,21 @@ ManifestFileIterator::ManifestFileIterator( , filter_dag(std::move(filter_dag_)) , schema_processor_ptr(&schema_processor) { + if (!inherited_first_row_id_.has_value()) + return; + + entry_first_row_ids.resize(total_rows); + UInt64 next_row_id = *inherited_first_row_id_; + for (size_t row_index = 0; row_index < total_rows; ++row_index) + { + const auto parsed_entry = manifest_file_deserializer->getParsedManifestFileEntry(row_index); + if (parsed_entry->content_type != FileContentType::DATA || parsed_entry->status != ManifestEntryStatus::ADDED + || parsed_entry->parsed_first_row_id.has_value()) + continue; + + entry_first_row_ids[row_index] = next_row_id; + next_row_id += static_cast(parsed_entry->record_count); + } } ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index) @@ -346,6 +442,10 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index) auto entry = std::make_shared( parsed_entry, common_partition_specification, resolved_sequence_number, resolved_schema_id); + if (parsed_entry->parsed_first_row_id.has_value()) + entry->first_row_id = parsed_entry->parsed_first_row_id; + else if (!entry_first_row_ids.empty()) + entry->first_row_id = entry_first_row_ids[row_index]; PruningReturnStatus pruning_status = PruningReturnStatus::NOT_PRUNED; if (filter_dag) @@ -383,6 +483,8 @@ ProcessedManifestFileEntryPtr ManifestFileIterator::processRow(size_t row_index) hyperrectangles.emplace(column_id, DB::Range(*left, true, *right, true)); } + + addRowLineageHyperrectangles(hyperrectangles, *entry); } const ManifestFilesPruner * current_pruner = getOrCreatePruner(entry->resolved_schema_id); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h index e109cd107890..1577deb160af 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.h @@ -82,6 +82,7 @@ class ManifestFileIterator : public boost::noncopyable IcebergSchemaProcessor & schema_processor, Int64 inherited_sequence_number, Int64 inherited_snapshot_id, + std::optional inherited_first_row_id, DB::ContextPtr context, std::shared_ptr filter_dag_, Int32 table_snapshot_schema_id_); @@ -117,6 +118,7 @@ class ManifestFileIterator : public boost::noncopyable IcebergSchemaProcessor & schema_processor, Int64 inherited_sequence_number, Int64 inherited_snapshot_id, + std::optional inherited_first_row_id, DB::ContextPtr context, Int32 manifest_schema_id, std::shared_ptr common_partition_specification, @@ -135,6 +137,7 @@ class ManifestFileIterator : public boost::noncopyable // always zero in case of format version 1 const Int64 inherited_sequence_number; const Int64 inherited_snapshot_id; + std::vector> entry_first_row_ids; const DB::ContextPtr context; const Int32 manifest_schema_id; const std::shared_ptr common_partition_specification; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp index 0a02fd75c04f..c2cfd9c7378b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.cpp @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -58,7 +59,16 @@ DB::ASTPtr getASTFromTransform(const String & transform_name_src, const String & return makeASTFunction(transform_and_argument->transform_name, make_intrusive(column_name)); } -std::unique_ptr ManifestFilesPruner::transformFilterDagForManifest(const DB::ActionsDAG * source_dag, std::vector & used_columns_in_filter) const +namespace +{ + constexpr const char * row_id_column = "_row_id"; + constexpr const char * last_sequence_number_column = "_last_updated_sequence_number"; +} + +std::unique_ptr ManifestFilesPruner::transformFilterDagForManifest( + const DB::ActionsDAG * source_dag, + std::vector & used_columns_in_filter, + std::unordered_map & row_lineage_columns_in_filter) const { const auto & inputs = source_dag->getInputs(); @@ -67,6 +77,14 @@ std::unique_ptr ManifestFilesPruner::transformFilterDagForManife if (input->type == ActionsDAG::ActionType::INPUT) { std::string input_name = input->result_name; + if (input_name == row_id_column || input_name == last_sequence_number_column) + { + const Int32 field_id = input_name == row_id_column ? row_id_field_id : last_updated_sequence_number_field_id; + used_columns_in_filter.push_back(field_id); + row_lineage_columns_in_filter.emplace(field_id, DB::NameAndTypePair(input_name, input->result_type)); + continue; + } + std::optional input_id = schema_processor.tryGetColumnIDByName(current_schema_id, input_name); if (input_id) used_columns_in_filter.push_back(*input_id); @@ -76,6 +94,13 @@ std::unique_ptr ManifestFilesPruner::transformFilterDagForManife ActionsDAG dag_with_renames; for (const auto column_id : used_columns_in_filter) { + if (auto lineage_column = row_lineage_columns_in_filter.find(column_id); lineage_column != row_lineage_columns_in_filter.end()) + { + const auto * node = &dag_with_renames.addInput(lineage_column->second.name, lineage_column->second.type); + dag_with_renames.getOutputs().push_back(node); + continue; + } + auto column = schema_processor.tryGetFieldCharacteristics(current_schema_id, column_id); /// Columns which we dropped and don't exist in current schema @@ -117,7 +142,7 @@ ManifestFilesPruner::ManifestFilesPruner( std::unique_ptr transformed_dag; std::vector used_columns_in_filter; - transformed_dag = transformFilterDagForManifest(filter_dag, used_columns_in_filter); + transformed_dag = transformFilterDagForManifest(filter_dag, used_columns_in_filter, row_lineage_columns); chassert(transformed_dag != nullptr); if (manifest_file.hasPartitionKey()) @@ -130,11 +155,19 @@ ManifestFilesPruner::ManifestFilesPruner( for (Int32 used_column_id : used_columns_in_filter) { - auto name_and_type = schema_processor.tryGetFieldCharacteristics(initial_schema_id, used_column_id); - if (!name_and_type.has_value()) - continue; + std::optional name_and_type; + if (auto lineage_column = row_lineage_columns.find(used_column_id); lineage_column != row_lineage_columns.end()) + { + name_and_type = lineage_column->second; + } + else + { + name_and_type = schema_processor.tryGetFieldCharacteristics(initial_schema_id, used_column_id); + if (!name_and_type.has_value()) + continue; - name_and_type->name = DB::backQuote(DB::toString(used_column_id)); + name_and_type->name = DB::backQuote(DB::toString(used_column_id)); + } ExpressionActionsPtr expression = std::make_shared(ActionsDAG({name_and_type.value()}), ExpressionActionsSettings(context)); @@ -169,22 +202,31 @@ PruningReturnStatus ManifestFilesPruner::canBePruned( for (const auto & [column_id, key_condition] : min_max_key_conditions) { - std::optional name_and_type = schema_processor.tryGetFieldCharacteristics(initial_schema_id, column_id); + std::optional name_and_type; + bool has_no_nulls = true; - /// There is no such column in this manifest file - if (!name_and_type.has_value()) + if (auto lineage_column = row_lineage_columns.find(column_id); lineage_column != row_lineage_columns.end()) { - continue; + name_and_type = lineage_column->second; + } + else + { + name_and_type = schema_processor.tryGetFieldCharacteristics(initial_schema_id, column_id); + + if (!name_and_type.has_value()) + { + continue; + } + + auto info_it = entry->parsed_entry->columns_infos.find(column_id); + has_no_nulls = info_it != entry->parsed_entry->columns_infos.end() && info_it->second.nulls_count.has_value() + && *info_it->second.nulls_count == 0; } auto rect_it = entry_hyperrectangles.find(column_id); if (rect_it == entry_hyperrectangles.end()) continue; - auto info_it = entry->parsed_entry->columns_infos.find(column_id); - bool has_no_nulls = info_it != entry->parsed_entry->columns_infos.end() && info_it->second.nulls_count.has_value() - && *info_it->second.nulls_count == 0; - if (has_no_nulls && !key_condition.mayBeTrueInRange(1, &rect_it->second.left, &rect_it->second.right, {name_and_type->type})) { return PruningReturnStatus::MIN_MAX_INDEX_PRUNED; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h index d04ced3796ee..0cbfe3d1ee78 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFilesPruning.h @@ -1,7 +1,10 @@ #pragma once #include "config.h" +#include +#include +#include #include #include #include @@ -43,10 +46,14 @@ class ManifestFilesPruner std::optional partition_key_condition; std::unordered_map min_max_key_conditions; + std::unordered_map row_lineage_columns; /// NOTE: tricky part to support RENAME column. /// Takes ActionDAG representation of user's WHERE expression and /// rename columns to the their origina numeric ID's in iceberg - std::unique_ptr transformFilterDagForManifest(const DB::ActionsDAG * source_dag, std::vector & used_columns_in_filter) const; + std::unique_ptr transformFilterDagForManifest( + const DB::ActionsDAG * source_dag, + std::vector & used_columns_in_filter, + std::unordered_map & row_lineage_columns_in_filter) const; public: ManifestFilesPruner( diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.cpp index 1294ccc9b0c5..701bbb81c4b5 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.cpp @@ -132,6 +132,7 @@ Iceberg::ManifestFileIterator::ManifestFileEntriesHandle getManifestFileEntriesH *persistent_table_components.schema_processor, cache_key.added_sequence_number, cache_key.added_snapshot_id, + cache_key.first_row_id, local_context, nullptr, table_snapshot_schema_id); @@ -221,8 +222,17 @@ ManifestFileCacheKeys getManifestList( content_type = Iceberg::ManifestFileContentType( manifest_list_deserializer.getValueFromRowByName(i, f_content, TypeIndex::Int32).safeGet()); } + + std::optional first_row_id; + if (manifest_list_format_version > 2 && manifest_list_deserializer.hasPath(f_manifest_first_row_id)) + { + auto first_row_id_value = manifest_list_deserializer.getValueFromRowByName(i, f_manifest_first_row_id); + if (!first_row_id_value.isNull()) + first_row_id = first_row_id_value.safeGet(); + } + manifest_file_cache_keys.emplace_back( - manifest_file_name, static_cast(manifest_length), added_sequence_number, added_snapshot_id.safeGet(), content_type); + manifest_file_name, static_cast(manifest_length), added_sequence_number, added_snapshot_id.safeGet(), content_type, first_row_id); auto dump_row_metadata = [&]()->String { return manifest_list_deserializer.getContent(i); }; insertRowToLogTable( diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index c54be45de9a0..9764ae53584e 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -1,6 +1,7 @@ #include #include #include +#include "config.h" #include #include #include @@ -62,6 +63,8 @@ #include #include #include +#include +#include #include #include @@ -111,6 +114,33 @@ namespace ErrorCodes extern const int FILE_DOESNT_EXIST; } +namespace +{ +DataTypePtr rowLineageColumnType() +{ + return makeNullable(std::make_shared()); +} + +Names getMaterializedRowLineageColumns( + [[maybe_unused]] const ObjectInfo & object_info, + [[maybe_unused]] const ReadFromFormatInfo & read_from_format_info, + [[maybe_unused]] const String & format_name) +{ + Names result; +#if USE_AVRO + if (Poco::toLower(format_name) != "parquet" || !dynamic_cast(&object_info)) + return result; + + for (const auto * name : {"_row_id", "_last_updated_sequence_number"}) + { + if (read_from_format_info.requested_virtual_columns.contains(name)) + result.emplace_back(name); + } +#endif + return result; +} +} + static void logIcebergFileStats(const ObjectInfoPtr & object_info, const LoggerPtr & log) { #if USE_AVRO @@ -499,6 +529,32 @@ Chunk StorageObjectStorageSource::generate() const auto path = getUniqueStoragePathIdentifier(*configuration, *object_info, false); + ColumnPtr materialized_row_ids; + ColumnPtr materialized_last_updated_sequence_numbers; + /// Without an input format the chunk comes from the count-from-cache path, which reads no + /// file and therefore carries no row lineage columns. + if (const auto lineage_columns = reader.getInputFormat() + ? getMaterializedRowLineageColumns( + *object_info, read_from_format_info, object_info->getFileFormat().value_or(configuration->getFormat())) + : Names{}; + !lineage_columns.empty()) + { + auto columns = chunk.detachColumns(); + chassert(columns.size() >= lineage_columns.size()); + const size_t first_lineage_column = columns.size() - lineage_columns.size(); + + for (size_t i = 0; i < lineage_columns.size(); ++i) + { + if (lineage_columns[i] == "_row_id") + materialized_row_ids = columns[first_lineage_column + i]; + else + materialized_last_updated_sequence_numbers = columns[first_lineage_column + i]; + } + + columns.resize(first_lineage_column); + chunk.setColumns(std::move(columns), num_rows); + } + /// The order is important, hive partition columns must be added before virtual columns /// because they are part of the schema if (!read_from_format_info.hive_partition_columns_to_read_from_file_path.empty()) @@ -514,9 +570,16 @@ Chunk StorageObjectStorageSource::generate() std::string path_for_virtual_column = getMetadataPathFromObjectInfo(object_info).value_or(path); const String * iceberg_metadata_file_path = nullptr; + std::optional last_updated_sequence_number; + std::optional first_row_id; #if USE_AVRO if (const auto * iceberg_info = dynamic_cast(object_info.get())) + { iceberg_metadata_file_path = &iceberg_info->info.data_object_file_path_key.serialize(); + first_row_id = iceberg_info->info.first_row_id; + if (first_row_id.has_value()) + last_updated_sequence_number = iceberg_info->info.sequence_number; + } #endif VirtualColumnUtils::addRequestedFileLikeStorageVirtualsToChunk( @@ -532,6 +595,10 @@ Chunk StorageObjectStorageSource::generate() .tags = &(object_metadata->tags), .data_lake_snapshot_version = file_iterator->getSnapshotVersion(), .iceberg_metadata_file_path = iceberg_metadata_file_path, + .last_updated_sequence_number = last_updated_sequence_number, + .first_row_id = first_row_id, + .materialized_row_ids = materialized_row_ids, + .materialized_last_updated_sequence_numbers = materialized_last_updated_sequence_numbers, }, read_context, format_settings); @@ -881,6 +948,7 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade std::unordered_set constant_columns; NamesAndTypesList requested_columns_copy = read_from_format_info.requested_columns; + Names row_lineage_columns; std::unordered_map> requested_columns_list; { @@ -1053,6 +1121,7 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade { const auto format_name = object_info->getFileFormat().value_or(configuration->getFormat()); const bool input_format_does_not_read_file = Poco::toLower(format_name) == "one"; + row_lineage_columns = getMaterializedRowLineageColumns(*object_info, read_from_format_info, format_name); CompressionMethod compression_method = {}; if (input_format_does_not_read_file) @@ -1092,6 +1161,13 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade initial_header = sample_header; schema_changed = true; } + + for (const auto & column_name : row_lineage_columns) + { + if (!initial_header.has(column_name)) + initial_header.insert({rowLineageColumnType()->createColumn(), rowLineageColumnType(), column_name}); + } + /// Save stripped filters if we need to apply them as fallback FilterTransforms /// later in the pipeline when the file format doesn't support PREWHERE. FilterDAGInfoPtr stripped_row_level_filter; @@ -1286,6 +1362,12 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade if (schema_transform.has_value()) { + for (const auto & column_name : row_lineage_columns) + { + const auto & input = schema_transform->addInput(column_name, rowLineageColumnType()); + schema_transform->getOutputs().push_back(&input); + } + auto schema_modifying_actions = std::make_shared(std::move(schema_transform.value())); builder.addSimpleTransform([&](const SharedHeader & header) { @@ -1365,9 +1447,13 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade /// Add ExtractColumnsTransform to extract requested columns/subcolumns /// from chunk read by IInputFormat. + NamesAndTypesList columns_to_extract = requested_columns_copy; + for (const auto & column_name : row_lineage_columns) + columns_to_extract.emplace_back(column_name, rowLineageColumnType()); + builder.addSimpleTransform([&](const SharedHeader & header) { - return std::make_shared(header, requested_columns_copy); + return std::make_shared(header, columns_to_extract); }); auto pipeline = std::make_unique(QueryPipelineBuilder::getPipeline(std::move(builder))); diff --git a/src/Storages/System/StorageSystemIcebergFiles.cpp b/src/Storages/System/StorageSystemIcebergFiles.cpp index 23e95b03775c..4f40633294c0 100644 --- a/src/Storages/System/StorageSystemIcebergFiles.cpp +++ b/src/Storages/System/StorageSystemIcebergFiles.cpp @@ -123,12 +123,14 @@ class SystemIcebergFilesSource : public ISource MutableColumnPtr col_column_sizes = make_int32_int64_map(); MutableColumnPtr col_value_counts = make_int32_int64_map(); MutableColumnPtr col_equality_ids = ColumnArray::create(ColumnInt32::create()); + MutableColumnPtr col_first_row_id = ColumnNullable::create(ColumnUInt64::create(), ColumnUInt8::create()); std::vector col_ptrs{ col_database.get(), col_table.get(), col_snapshot_id.get(), col_content.get(), col_file_path.get(), col_file_format.get(), col_record_count.get(), col_file_size_in_bytes.get(), col_partition.get(), col_schema_id.get(), col_sequence_number.get(), col_sort_order_id.get(), - col_null_value_counts.get(), col_column_sizes.get(), col_value_counts.get(), col_equality_ids.get()}; + col_null_value_counts.get(), col_column_sizes.get(), col_value_counts.get(), col_equality_ids.get(), + col_first_row_id.get()}; size_t num_rows = 0; @@ -192,6 +194,11 @@ class SystemIcebergFilesSource : public ISource equality_ids_array.push_back(id); col_equality_ids->insert(equality_ids_array); + if (file.first_row_id.has_value()) + col_first_row_id->insert(*file.first_row_id); + else + col_first_row_id->insertDefault(); + ++num_rows; } } @@ -347,7 +354,8 @@ class SystemIcebergFilesSource : public ISource std::move(col_database), std::move(col_table), std::move(col_snapshot_id), std::move(col_content), std::move(col_file_path), std::move(col_file_format), std::move(col_record_count), std::move(col_file_size_in_bytes), std::move(col_partition), std::move(col_schema_id), std::move(col_sequence_number), std::move(col_sort_order_id), - std::move(col_null_value_counts), std::move(col_column_sizes), std::move(col_value_counts), std::move(col_equality_ids)}; + std::move(col_null_value_counts), std::move(col_column_sizes), std::move(col_value_counts), std::move(col_equality_ids), + std::move(col_first_row_id)}; return Chunk(std::move(columns), num_rows); } @@ -478,6 +486,7 @@ StorageSystemIcebergFiles::StorageSystemIcebergFiles(const StorageID & table_id_ {"column_sizes", std::make_shared(std::make_shared(), std::make_shared()), "Per-column on-disk size in bytes (column id -> bytes)."}, {"value_counts", std::make_shared(std::make_shared(), std::make_shared()), "Per-column total value count (column id -> count)."}, {"equality_ids", std::make_shared(std::make_shared()), "Equality field IDs for equality delete files (empty for non-equality-delete files)."}, + {"first_row_id", std::make_shared(std::make_shared()), "Row lineage `_row_id` assigned to the first row of the file, with inheritance from the manifest list resolved (NULL before format v3)."}, }); setInMemoryMetadata(storage_metadata); } diff --git a/src/Storages/VirtualColumnUtils.cpp b/src/Storages/VirtualColumnUtils.cpp index 7ab0d5a1b7ae..2b16d40877a6 100644 --- a/src/Storages/VirtualColumnUtils.cpp +++ b/src/Storages/VirtualColumnUtils.cpp @@ -1,9 +1,13 @@ #include +#include #include #include +#include #include +#include + #include #include @@ -15,6 +19,7 @@ #include +#include #include #include #include @@ -29,6 +34,7 @@ #include #include +#include #include #include @@ -198,6 +204,8 @@ static NamesAndTypesList getCommonVirtualsForFileLikeStorage() {"_tags", std::make_shared(std::make_shared(), std::make_shared())}, {"_data_lake_snapshot_version", makeNullable(std::make_shared())}, {"_row_number", makeNullable(std::make_shared())}, + {"_row_id", makeNullable(std::make_shared())}, + {"_last_updated_sequence_number", makeNullable(std::make_shared())}, {"_iceberg_metadata_file_path", std::make_shared(std::make_shared())}, {"_table", std::make_shared(std::make_shared())}, }; @@ -394,6 +402,31 @@ ColumnPtr getFilterByPathAndFileIndexes( return block.getByName("_idx").column; } +/// Builds a `Nullable(UInt64)` row lineage column out of the values materialized in the data file, +/// falling back to the value derived from the file metadata wherever the file has no value. +template +static ColumnPtr buildRowLineageColumn(size_t num_rows, const IColumn * materialized, FallbackFn && fallback) +{ + auto column = ColumnUInt64::create(); + auto null_map = ColumnUInt8::create(); + column->reserve(num_rows); + null_map->reserve(num_rows); + + for (size_t row = 0; row < num_rows; ++row) + { + std::optional value; + if (materialized && !materialized->isNullAt(row)) + value = materialized->getUInt(row); + else + value = fallback(row); + + column->insertValue(value.value_or(0)); + null_map->insertValue(value.has_value() ? 0 : 1); + } + + return ColumnNullable::create(std::move(column), std::move(null_map)); +} + void addRequestedFileLikeStorageVirtualsToChunk( Chunk & chunk, const NamesAndTypesList & requested_virtual_columns, @@ -500,6 +533,50 @@ void addRequestedFileLikeStorageVirtualsToChunk( chunk.addColumn(virtual_column.type->createColumnConstWithDefaultValue(chunk.getNumRows())->convertToFullColumnIfConst()); #endif } + else if (virtual_column.name == "_last_updated_sequence_number") + { + if (virtual_values.materialized_last_updated_sequence_numbers) + { + chunk.addColumn(buildRowLineageColumn( + chunk.getNumRows(), + virtual_values.materialized_last_updated_sequence_numbers.get(), + [&](size_t) { return virtual_values.last_updated_sequence_number; })); + } + else if (virtual_values.last_updated_sequence_number) + chunk.addColumn(virtual_column.type->createColumnConst(chunk.getNumRows(), *virtual_values.last_updated_sequence_number)->convertToFullColumnIfConst()); + else + chunk.addColumn(virtual_column.type->createColumnConstWithDefaultValue(chunk.getNumRows())->convertToFullColumnIfConst()); + } + else if (virtual_column.name == "_row_id") + { + std::vector row_positions; +#if USE_PARQUET + if (auto chunk_info = chunk.getChunkInfos().get(); chunk_info && virtual_values.first_row_id) + { + const auto & applied_filter = chunk_info->applied_filter; + size_t num_indices = applied_filter.has_value() ? applied_filter->size() : chunk.getNumRows(); + row_positions.reserve(chunk.getNumRows()); + for (size_t i = 0; i < num_indices; ++i) + if (!applied_filter.has_value() || applied_filter.value()[i]) + row_positions.push_back(chunk_info->row_num_offset + i); + } +#endif + if (row_positions.empty() && !virtual_values.materialized_row_ids) + { + chunk.addColumn(virtual_column.type->createColumnConstWithDefaultValue(chunk.getNumRows())->convertToFullColumnIfConst()); + continue; + } + + chunk.addColumn(buildRowLineageColumn( + chunk.getNumRows(), + virtual_values.materialized_row_ids.get(), + [&](size_t row) -> std::optional + { + if (row >= row_positions.size()) + return std::nullopt; + return *virtual_values.first_row_id + row_positions[row]; + })); + } else if (virtual_column.name == "_iceberg_metadata_file_path") { if (virtual_values.iceberg_metadata_file_path) @@ -529,7 +606,7 @@ bool hasRowDependentVirtualColumns(const NamesAndTypesList & requested_virtual_c return std::any_of( requested_virtual_columns.begin(), requested_virtual_columns.end(), - [](const auto & col) { return col.name == "_row_number"; }); + [](const auto & col) { return col.name == "_row_number" || col.name == "_row_id" || col.name == "_last_updated_sequence_number"; }); } static bool canEvaluateSubtree(const ActionsDAG::Node * node, const Block * allowed_inputs) diff --git a/src/Storages/VirtualColumnUtils.h b/src/Storages/VirtualColumnUtils.h index ae96dfb921c2..62f01d8c2be0 100644 --- a/src/Storages/VirtualColumnUtils.h +++ b/src/Storages/VirtualColumnUtils.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -152,6 +153,10 @@ struct VirtualsForFileLikeStorage /// Original file path as stored in Iceberg metadata (before resolution to storage path). /// Used by Iceberg position deletes to reference data files in the metadata path format. const String * iceberg_metadata_file_path { nullptr }; + std::optional last_updated_sequence_number = std::nullopt; + std::optional first_row_id = std::nullopt; + ColumnPtr materialized_row_ids = {}; + ColumnPtr materialized_last_updated_sequence_numbers = {}; }; void addRequestedFileLikeStorageVirtualsToChunk( diff --git a/tests/integration/test_storage_iceberg_with_spark/test_row_lineage.py b/tests/integration/test_storage_iceberg_with_spark/test_row_lineage.py new file mode 100644 index 000000000000..3fd5125a93ff --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/test_row_lineage.py @@ -0,0 +1,278 @@ +import pytest + +from helpers.iceberg_utils import ( + create_iceberg_table, + default_upload_directory, + drop_iceberg_table, + get_creation_expression, + get_uuid_str, +) + + +def _spark_lineage(spark, table_name): + rows = spark.sql( + f"SELECT id, _row_id, _last_updated_sequence_number FROM {table_name}" + ).collect() + return { + row["id"]: (row["_row_id"], row["_last_updated_sequence_number"]) for row in rows + } + + +def _clickhouse_lineage(instance, table_expression, where="", settings=None): + raw = instance.query( + f"SELECT id, _row_id, _last_updated_sequence_number FROM {table_expression} {where} FORMAT TSV", + settings=settings, + ) + + def parse(value): + return None if value == "\\N" else int(value) + + lineage = {} + for line in raw.strip().split("\n"): + if not line: + continue + row_key, row_id, sequence_number = line.split("\t") + lineage[int(row_key)] = (parse(row_id), parse(sequence_number)) + return lineage + + +def _row_ids(lineage): + return {row_key: row_id for row_key, (row_id, _) in lineage.items()} + + +def _publish(started_cluster, storage_type, table_name): + default_upload_directory( + started_cluster, + storage_type, + f"/iceberg_data/default/{table_name}/", + f"/iceberg_data/default/{table_name}/", + ) + + +@pytest.mark.parametrize("run_on_cluster", [False, True]) +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_lineage_inherited_from_manifest( + started_cluster_iceberg_with_spark, storage_type, run_on_cluster +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_lineage_inherited_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '3')" + ) + for lo in range(0, 40, 10): + spark.sql( + f"INSERT INTO {TABLE_NAME} select id, char(id + ascii('a')) from range({lo}, {lo + 10})" + ) + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = get_creation_expression( + storage_type, + TABLE_NAME, + started_cluster_iceberg_with_spark, + table_function=True, + run_on_cluster=run_on_cluster, + ) + + assert int(instance.query(f"SELECT count() FROM {table_expression}")) == 40 + + spark_lineage = _spark_lineage(spark, TABLE_NAME) + + assert sorted(row_id for row_id, _ in spark_lineage.values()) == list(range(40)) + for row_key, (row_id, sequence_number) in spark_lineage.items(): + assert row_id == row_key + assert sequence_number == row_key // 10 + 1 + + assert _clickhouse_lineage(instance, table_expression) == spark_lineage + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_id_inherited_for_several_files_in_one_manifest( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_id_one_manifest_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, part int) USING iceberg " + f"PARTITIONED BY (part) TBLPROPERTIES ('format-version' = '3')" + ) + spark.sql(f"INSERT INTO {TABLE_NAME} select id, id % 4 from range(0, 20)") + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = get_creation_expression( + storage_type, + TABLE_NAME, + started_cluster_iceberg_with_spark, + table_function=True, + ) + + assert int(instance.query(f"SELECT count() FROM {table_expression}")) == 20 + + spark_lineage = _spark_lineage(spark, TABLE_NAME) + + assert sorted(row_id for row_id, _ in spark_lineage.values()) == list(range(20)) + assert all(sequence_number == 1 for _, sequence_number in spark_lineage.values()) + + assert _clickhouse_lineage(instance, table_expression) == spark_lineage + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_id_is_not_affected_by_filter_pushdown( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_id_filter_pushdown_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '3', 'write.parquet.row-group-size-bytes' = '100')" + ) + spark.sql(f"INSERT INTO {TABLE_NAME} select id, 'a' from range(0, 10)") + spark.sql(f"INSERT INTO {TABLE_NAME} select id, 'b' from range(10, 20)") + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = get_creation_expression( + storage_type, + TABLE_NAME, + started_cluster_iceberg_with_spark, + table_function=True, + ) + + assert _row_ids( + _clickhouse_lineage(instance, table_expression, where="WHERE id >= 15") + ) == {row_key: row_key for row_key in range(15, 20)} + + assert _row_ids( + _clickhouse_lineage(instance, table_expression, where="WHERE id % 7 = 3") + ) == {3: 3, 10: 10, 17: 17} + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_lineage_materialized_after_update( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_lineage_materialized_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '3', 'write.update.mode' = 'copy-on-write')" + ) + spark.sql(f"INSERT INTO {TABLE_NAME} select id, 'a' from range(0, 4)") + spark.sql(f"UPDATE {TABLE_NAME} SET data = 'z' WHERE id = 1") + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = get_creation_expression( + storage_type, + TABLE_NAME, + started_cluster_iceberg_with_spark, + table_function=True, + ) + + assert int(instance.query(f"SELECT count() FROM {table_expression}")) == 4 + + spark_lineage = _spark_lineage(spark, TABLE_NAME) + + assert spark_lineage == {0: (0, 1), 1: (1, 2), 2: (2, 1), 3: (3, 1)} + + assert _clickhouse_lineage(instance, table_expression) == spark_lineage + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_id_survives_copy_on_write_delete( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_id_after_delete_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '3', 'write.delete.mode' = 'copy-on-write')" + ) + spark.sql(f"INSERT INTO {TABLE_NAME} select id, 'a' from range(0, 4)") + spark.sql(f"DELETE FROM {TABLE_NAME} WHERE id = 1") + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = get_creation_expression( + storage_type, + TABLE_NAME, + started_cluster_iceberg_with_spark, + table_function=True, + ) + + assert int(instance.query(f"SELECT count() FROM {table_expression}")) == 3 + + spark_lineage = _spark_lineage(spark, TABLE_NAME) + + assert _row_ids(spark_lineage) == {0: 0, 2: 2, 3: 3} + + assert _clickhouse_lineage(instance, table_expression) == spark_lineage + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_lineage_is_null_for_v2_table( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_lineage_v2_null_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '2')" + ) + spark.sql(f"INSERT INTO {TABLE_NAME} select id, 'a' from range(0, 4)") + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = get_creation_expression( + storage_type, + TABLE_NAME, + started_cluster_iceberg_with_spark, + table_function=True, + ) + + assert _clickhouse_lineage(instance, table_expression) == { + row_key: (None, None) for row_key in range(4) + } + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_first_row_id_in_system_iceberg_files( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_first_row_id_system_table_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '3')" + ) + for lo in range(0, 40, 10): + spark.sql(f"INSERT INTO {TABLE_NAME} select id, 'a' from range({lo}, {lo + 10})") + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + create_iceberg_table( + storage_type, + instance, + TABLE_NAME, + started_cluster_iceberg_with_spark, + format_version=3, + ) + + # Spark writes the entries without an explicit first_row_id and it is assigned from the manifest + # list at read time, so the system table is where the resolved value can be observed. + assert instance.query( + f"SELECT first_row_id FROM system.iceberg_files " + f"WHERE database = currentDatabase() AND table = '{TABLE_NAME}' AND content = 'DATA' " + f"ORDER BY first_row_id FORMAT TSV" + ).split() == ["0", "10", "20", "30"] + + drop_iceberg_table(instance, TABLE_NAME) diff --git a/tests/integration/test_storage_iceberg_with_spark/test_row_lineage_pruning.py b/tests/integration/test_storage_iceberg_with_spark/test_row_lineage_pruning.py new file mode 100644 index 000000000000..8ac5e032b268 --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/test_row_lineage_pruning.py @@ -0,0 +1,352 @@ +"""Pruning by the Iceberg v3 row lineage columns (`_row_id`, `_last_updated_sequence_number`). + +Both columns are derived from file-level metadata, so a filter on them is answerable from the +manifest alone: a data file holds the row ids `[first_row_id, first_row_id + record_count)` and a +single `_last_updated_sequence_number` (its data sequence number). That makes the incremental "give +me everything changed after sequence number N" read - the way row lineage is meant to be consumed - +a metadata-only file skip, and a `_row_id` point lookup a single-file read. + +The trap these tests guard is the copy-on-write rewrite: a file whose rows carry materialized row +ids does NOT hold the contiguous range its manifest entry advertises, so pruning it by that range +drops live rows. See `test_row_lineage.py` for how the values themselves are derived. + +NOTE: row lineage is written only by `iceberg-spark-runtime` 1.10.0 and later, see the version +pinned in `ci/docker/integration/runner/Dockerfile`. +""" + +import uuid + +import pytest + +from helpers.iceberg_utils import ( + check_validity_and_get_prunned_files_general, + default_upload_directory, + get_creation_expression, + get_uuid_str, +) + +# Both runs must read the same data, so only the metadata-level skip differs between them. +PRUNING_DISABLED = { + "use_iceberg_partition_pruning": 0, + "input_format_parquet_filter_push_down": 0, + "input_format_parquet_bloom_filter_push_down": 0, +} +PRUNING_ENABLED = { + "use_iceberg_partition_pruning": 1, + "input_format_parquet_filter_push_down": 0, + "input_format_parquet_bloom_filter_push_down": 0, +} + + +def _publish(started_cluster, storage_type, table_name): + default_upload_directory( + started_cluster, + storage_type, + f"/iceberg_data/default/{table_name}/", + f"/iceberg_data/default/{table_name}/", + ) + + +def _table_function(started_cluster, storage_type, table_name): + return get_creation_expression( + storage_type, table_name, started_cluster, table_function=True + ) + + +def _pruned_files(instance, table_name, select_expression): + """Number of data files skipped by the manifest-level filter, cross-checked for equal results.""" + return check_validity_and_get_prunned_files_general( + instance, + table_name, + PRUNING_DISABLED, + PRUNING_ENABLED, + "IcebergMinMaxIndexPrunedFiles", + select_expression, + ) + + +def _read_rows(instance, select_expression, settings=None): + query_id = f"row-lineage-pruning-{uuid.uuid4()}" + instance.query(select_expression, query_id=query_id, settings=settings) + instance.query("SYSTEM FLUSH LOGS") + return int( + instance.query( + f"SELECT read_rows FROM system.query_log WHERE query_id = '{query_id}' AND type = 'QueryFinish'" + ) + ) + + +def _create_table_with_five_appends(spark, table_name, file_format="parquet"): + """Five appends of ten rows: file k holds row ids [10k, 10k + 10) and sequence number k + 1.""" + spark.sql( + f"CREATE TABLE {table_name} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '3', 'write.format.default' = '{file_format}')" + ) + for lo in range(0, 50, 10): + spark.sql( + f"INSERT INTO {table_name} select id, 'a' from range({lo}, {lo + 10})" + ) + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_id_filter_prunes_files(started_cluster_iceberg_with_spark, storage_type): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_id_pruning_" + storage_type + "_" + get_uuid_str() + + _create_table_with_five_appends(spark, TABLE_NAME) + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = _table_function( + started_cluster_iceberg_with_spark, storage_type, TABLE_NAME + ) + + assert ( + _pruned_files(instance, TABLE_NAME, f"SELECT id FROM {table_expression} ORDER BY ALL") + == 0 + ) + + # A point lookup touches the one file whose row id range contains the value. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _row_id = 25 ORDER BY ALL", + ) + == 4 + ) + + # A half-open range keeps the two files above it. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _row_id >= 35 ORDER BY ALL", + ) + == 3 + ) + + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _row_id < 10 ORDER BY ALL", + ) + == 4 + ) + + # A range spanning everything prunes nothing. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _row_id >= 0 ORDER BY ALL", + ) + == 0 + ) + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_incremental_read_by_sequence_number_prunes_files( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_sequence_number_pruning_" + storage_type + "_" + get_uuid_str() + + _create_table_with_five_appends(spark, TABLE_NAME) + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = _table_function( + started_cluster_iceberg_with_spark, storage_type, TABLE_NAME + ) + + # "Everything that changed after the sequence number I consumed last time" - the whole point of + # the column - must not open the files it is going to discard. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number > 3 ORDER BY ALL", + ) + == 3 + ) + + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number = 2 ORDER BY ALL", + ) + == 4 + ) + + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number > 0 ORDER BY ALL", + ) + == 0 + ) + + # An incremental consumer reads exactly the rows of the two newest files, not the whole table. + assert ( + _read_rows( + instance, + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number > 3 FORMAT Null", + PRUNING_ENABLED, + ) + == 20 + ) + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_pruning_without_column_statistics( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_lineage_pruning_no_stats_" + storage_type + "_" + get_uuid_str() + + # Avro data files carry no per-column statistics, so the manifest cannot tell whether a file + # materializes row lineage - and materialized values are always carried over from an earlier + # write, so only the upper bound of the inherited range is known to hold. + _create_table_with_five_appends(spark, TABLE_NAME, file_format="avro") + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = _table_function( + started_cluster_iceberg_with_spark, storage_type, TABLE_NAME + ) + + # An incremental read still skips everything written before the consumed sequence number. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number > 3 ORDER BY ALL", + ) + == 3 + ) + + # A point lookup keeps every file that could hold an older materialized value, so only the one + # file whose whole range is below the requested value is skipped. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number = 2 ORDER BY ALL", + ) + == 1 + ) + + # The same for row ids: everything above the inherited block is still skipped ... + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _row_id >= 35 ORDER BY ALL", + ) + == 3 + ) + + # ... while nothing below it is, because an older materialized id could be there. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _row_id < 10 ORDER BY ALL", + ) + == 0 + ) + + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _row_id = 25 ORDER BY ALL", + ) + == 2 + ) + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_row_id_pruning_is_skipped_for_orc( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_lineage_pruning_orc_" + storage_type + "_" + get_uuid_str() + + _create_table_with_five_appends(spark, TABLE_NAME, file_format="orc") + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = _table_function( + started_cluster_iceberg_with_spark, storage_type, TABLE_NAME + ) + + # The ORC reader reports no physical row numbers, so `_row_id` is NULL for every row and the + # inherited range describes nothing: pruning by it would drop the rows this query asks for. + assert ( + instance.query( + f"SELECT count() FROM {table_expression} WHERE _row_id IS NULL", + settings=PRUNING_ENABLED, + ).strip() + == "50" + ) + + # The sequence number does not depend on row numbers, so it prunes as usual. + assert ( + _pruned_files( + instance, + TABLE_NAME, + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number > 3 ORDER BY ALL", + ) + == 3 + ) + + +@pytest.mark.parametrize("storage_type", ["s3"]) +def test_materialized_row_ids_are_not_pruned_away( + started_cluster_iceberg_with_spark, storage_type +): + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + TABLE_NAME = "test_row_id_pruning_materialized_" + storage_type + "_" + get_uuid_str() + + spark.sql( + f"CREATE TABLE {TABLE_NAME} (id bigint, data string) USING iceberg " + f"TBLPROPERTIES ('format-version' = '3', 'write.update.mode' = 'copy-on-write')" + ) + spark.sql(f"INSERT INTO {TABLE_NAME} select id, 'a' from range(0, 4)") + spark.sql(f"UPDATE {TABLE_NAME} SET data = 'z' WHERE id = 1") + + _publish(started_cluster_iceberg_with_spark, storage_type, TABLE_NAME) + table_expression = _table_function( + started_cluster_iceberg_with_spark, storage_type, TABLE_NAME + ) + + # The rewritten file keeps the original row ids 0..3, while its manifest entry advertises the + # freshly reserved block 4..7. Pruning by that block would drop every live row, so a file with + # materialized row ids must be read no matter what the filter says. + for row_key in range(4): + assert ( + instance.query( + f"SELECT id FROM {table_expression} WHERE _row_id = {row_key}", + settings=PRUNING_ENABLED, + ).strip() + == str(row_key) + ) + + assert ( + instance.query( + f"SELECT id FROM {table_expression} WHERE _row_id >= 4 ORDER BY ALL", + settings=PRUNING_ENABLED, + ).strip() + == "" + ) + + # The same for the sequence number: the updated row carries a materialized value of its own. + assert ( + instance.query( + f"SELECT id FROM {table_expression} WHERE _last_updated_sequence_number = 1 ORDER BY ALL", + settings=PRUNING_ENABLED, + ).strip() + == "0\n2\n3" + ) diff --git a/tests/queries/0_stateless/04203_system_iceberg_files.reference b/tests/queries/0_stateless/04203_system_iceberg_files.reference index 3c3a9f40d26e..1d81d09b8894 100644 --- a/tests/queries/0_stateless/04203_system_iceberg_files.reference +++ b/tests/queries/0_stateless/04203_system_iceberg_files.reference @@ -15,6 +15,7 @@ null_value_counts Map(Int32, Int64) column_sizes Map(Int32, Int64) value_counts Map(Int32, Int64) equality_ids Array(Int32) +first_row_id Nullable(UInt64) --- per-file columns (empty table) --- --- per-file columns --- Row 1: From a01afdf15040f9a577e56520530673390764d8a5 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Wed, 2 Sep 2026 23:36:52 +0200 Subject: [PATCH 2/3] Add DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_CDC_READING and DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION --- src/Core/ProtocolDefines.h | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/Core/ProtocolDefines.h b/src/Core/ProtocolDefines.h index 84379f04d80e..e86e2aaabd70 100644 --- a/src/Core/ProtocolDefines.h +++ b/src/Core/ProtocolDefines.h @@ -40,7 +40,8 @@ static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_FILE_BUCKETS static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_EXCLUDED_ROWS = 5; static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_FILE_STATS = 6; static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_ABSOLUTE_PATH = 9; -static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION = DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_ABSOLUTE_PATH; +static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_CDC_READING = 10; +static constexpr auto DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION = DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_ICEBERG_CDC_READING; static constexpr auto DATA_LAKE_TABLE_STATE_SNAPSHOT_PROTOCOL_VERSION = 1; From c9a0c51d9485250976107f4e000ef8255a391132 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 3 Sep 2026 17:28:21 +0200 Subject: [PATCH 3/3] Fixed gtest_iceberg_count_shortcuts test --- ci/docker/integration/runner/Dockerfile | 2 +- .../DataLakes/Iceberg/tests/gtest_iceberg_count_shortcuts.cpp | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/ci/docker/integration/runner/Dockerfile b/ci/docker/integration/runner/Dockerfile index 493c6011589e..91f16e3d68c3 100644 --- a/ci/docker/integration/runner/Dockerfile +++ b/ci/docker/integration/runner/Dockerfile @@ -77,7 +77,7 @@ RUN curl -fsSL -O https://archive.apache.org/dist/spark/spark-3.5.5/spark-3.5.5- # if you change packages, don't forget to update them in tests/integration/helpers/cluster.py RUN packages="io.delta:delta-spark_2.12:3.1.0,\ org.apache.hudi:hudi-spark3.5-bundle_2.12:1.0.1,\ -org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,\ +org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.10.0,\ org.apache.hadoop:hadoop-aws:3.3.4,\ com.amazonaws:aws-java-sdk-bundle:1.12.262,\ org.apache.hadoop:hadoop-azure:3.3.4,\ diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_count_shortcuts.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_count_shortcuts.cpp index 1c1a894e8018..5823889ba884 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_count_shortcuts.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_count_shortcuts.cpp @@ -125,6 +125,7 @@ ProcessedManifestFileEntryPtr makeDataEntryForRecordCount( ManifestEntryStatus::ADDED, /*written_sequence_number=*/std::nullopt, /*written_snapshot_id=*/std::nullopt, + /*written_first_row_id=*/std::nullopt, DB::Row{}, std::move(columns_infos), std::unordered_map>{},