Antalya 26.6 - backport of 115603 - iceberg v3 - row_lineage

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,\

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;
 

src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp

@@ -169,6 +169,13 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
         }
     }
 
+    std::optional<UInt64> 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<Int64>();
+    }
 
     const auto file_path_from_metadata = IcebergPathFromMetadata::deserialize(
         getValueFromRowByName(row_index, c_data_file_file_path, TypeIndex::String).safeGet<String>());
@@ -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,

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

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

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.h

@@ -1,6 +1,8 @@
 #pragma once
 #include "config.h"
 
+#include <optional>
+
 #include <Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.h>
 #include <Interpreters/Context_fwd.h>
 #include <Storages/ObjectStorage/IObjectIterator.h>
@@ -10,6 +12,7 @@
 
 #include <Core/Field.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergPath.h>
+#include <base/types.h>
 
 #include <string_view>
 
@@ -41,6 +44,7 @@ struct IcebergObjectSerializableInfo
     std::vector<Iceberg::EqualityDeleteObject> equality_deletes_objects;
     std::optional<Int64> record_count;
     std::optional<Int64> file_size_in_bytes;
+    std::optional<UInt64> first_row_id;
 
     /// Set to true by the coordinator when the file is outside of the table location
     bool requires_external_storage = false;

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp

@@ -198,6 +198,7 @@ std::optional<ProcessedManifestFileEntryPtr> 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);

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)
     {

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h

@@ -52,6 +52,7 @@ struct IcebergFileRecord
     std::map<Int32, Int64> column_sizes;
     std::map<Int32, Int64> value_counts;
     std::vector<Int32> equality_ids;
+    std::optional<UInt64> first_row_id;
 };
 
 class IcebergMetadata : public IDataLakeMetadata

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadataFilesCache.h

@@ -1,5 +1,6 @@
 #pragma once
 #include <chrono>
+#include <optional>
 #include <IO/CompressionMethod.h>
 #include <base/defines.h>
 #include "config.h"
@@ -59,6 +60,7 @@ struct ManifestFileCacheKey
     Int64 added_sequence_number;
     Int64 added_snapshot_id;
     Iceberg::ManifestFileContentType content_type;
+    std::optional<UInt64> first_row_id;
 };
 
 using ManifestFileCacheKeys = std::vector<ManifestFileCacheKey>;

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h

@@ -86,6 +86,7 @@ struct ParsedManifestFileEntry : boost::noncopyable
     ManifestEntryStatus status;
     std::optional<Int64> parsed_sequence_number;
     std::optional<Int64> parsed_snapshot_id;
+    std::optional<UInt64> parsed_first_row_id;
 
     DB::Row partition_key_value;
     std::unordered_map<Int32, ColumnInfo> columns_infos;
@@ -121,6 +122,7 @@ struct ParsedManifestFileEntry : boost::noncopyable
         ManifestEntryStatus status_,
         std::optional<Int64> written_sequence_number_,
         std::optional<Int64> written_snapshot_id_,
+        std::optional<UInt64> written_first_row_id_,
         DB::Row partition_key_value_,
         std::unordered_map<Int32, ColumnInfo> columns_infos_,
         std::unordered_map<Int32, std::pair<Field, Field>> 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<UInt64> first_row_id;
 
     String dumpDeletesMatchingInfo() const;
 };

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFileIterator.cpp

@@ -20,6 +20,7 @@
 #include <Core/Settings.h>
 #include <Core/TypeId.h>
 #include <DataTypes/DataTypesDecimal.h>
+#include <DataTypes/DataTypesNumber.h>
 #include <Poco/JSON/Parser.h>
 #include <Storages/ColumnsDescription.h>
 #include <Parsers/ASTFunction.h>
@@ -49,6 +50,83 @@ extern const Event IcebergPartitionPrunedFiles;
 extern const Event IcebergMinMaxIndexPrunedFiles;
 };
 
+namespace
+{
+
+using namespace DB::Iceberg;
+
+std::optional<DB::Range> 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<DB::DataTypeUInt64>();
+    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<Int32, DB::Range> & 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<UInt64>(entry.sequence_number);
+    const UInt64 last_inherited_row_id = *entry.first_row_id + static_cast<UInt64>(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> ManifestFileIterator::create(
     IcebergSchemaProcessor & schema_processor,
     Int64 inherited_sequence_number_,
     Int64 inherited_snapshot_id_,
+    std::optional<UInt64> inherited_first_row_id_,
     DB::ContextPtr context_,
     std::shared_ptr<const ActionsDAG> filter_dag_,
     Int32 table_snapshot_schema_id_)
@@ -217,6 +296,7 @@ std::shared_ptr<ManifestFileIterator> ManifestFileIterator::create(
         schema_processor,
         inherited_sequence_number_,
         inherited_snapshot_id_,
+        inherited_first_row_id_,
         context_,
         manifest_schema_id,
         std::make_shared<const PartitionSpecification>(std::move(partition_spec_vec)),
@@ -234,6 +314,7 @@ ManifestFileIterator::ManifestFileIterator(
     IcebergSchemaProcessor & schema_processor,
     Int64 inherited_sequence_number_,
     Int64 inherited_snapshot_id_,
+    std::optional<UInt64> inherited_first_row_id_,
     DB::ContextPtr context_,
     Int32 manifest_schema_id_,
     std::shared_ptr<const PartitionSpecification> 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<UInt64>(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<ProcessedManifestFileEntry>(
         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);