From bfa15409a756ca1f4f1e3fade942423c98e6edda Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Wed, 12 Aug 2026 20:53:42 +0200 Subject: [PATCH 1/2] Cherry-pick of https://github.com/Altinity/ClickHouse/pull/2125 with unresolved conflict markers (resolution in next commit) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Original cherry-pick message follows: Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655 Antalya 26.6: Implement TRUNCATE TABLE for Iceberg Engine (REST … # Conflicts: # src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp # src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp # src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h --- .../DataLakes/IDataLakeMetadata.h | 6 ++ .../DataLakes/Iceberg/IcebergMetadata.cpp | 94 ++++++++++++++++++- .../DataLakes/Iceberg/IcebergMetadata.h | 2 + .../DataLakes/Iceberg/IcebergWrites.cpp | 57 +++++++++++ .../DataLakes/Iceberg/MetadataGenerator.cpp | 37 ++++++++ .../DataLakes/Iceberg/MetadataGenerator.h | 4 + .../ObjectStorage/StorageObjectStorage.cpp | 10 +- .../test_iceberg_truncate.py | 80 ++++++++++++++++ 8 files changed, 284 insertions(+), 6 deletions(-) create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_iceberg_truncate.py diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h index 8347164b9a46..2bf6ed0f136d 100644 --- a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h @@ -181,6 +181,12 @@ class IDataLakeMetadata : boost::noncopyable throwNotImplemented(fmt::format("EXECUTE {}", command_name)); } + virtual bool supportsTruncate() const { return false; } + virtual void truncate(ContextPtr /*context*/, std::shared_ptr /*catalog*/, const StorageID & /*storage_id*/) + { + throwNotImplemented("truncate"); + } + virtual void drop(ContextPtr) { } virtual ObjectStorageType getObjectStorageType() const { return ObjectStorageType::None; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 2af675f6cf10..7f3f45c5259d 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -72,6 +72,7 @@ #include #include #include +#include #include #include #include @@ -120,7 +121,11 @@ extern const int ICEBERG_SPECIFICATION_VIOLATION; extern const int S3_ERROR; extern const int TABLE_ALREADY_EXISTS; extern const int SUPPORT_IS_DISABLED; +<<<<<<< HEAD extern const int FILE_ALREADY_EXISTS; +======= +extern const int INCORRECT_DATA; +>>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) } namespace Setting @@ -294,15 +299,17 @@ void IcebergMetadata::backgroundMetadataPrefetcherThread() /// first, we fetch the latest metadata version and cache it; /// as a part of the same method, we download metadata.json of the latest metadata version /// and after parsing it, we fetch manifest lists, parse and cache them - auto ctx = Context::getGlobalContextInstance()->getBackgroundContext(); + auto ctx = Context::createCopy(Context::getGlobalContextInstance()); auto [actual_data_snapshot, actual_table_state_snapshot] = getRelevantState(ctx, true); if (actual_data_snapshot) { for (const auto & entry : actual_data_snapshot->manifest_list_entries) { /// second, we fetch, parse and cache each manifest file - auto manifest_file_ptr = getManifestFileEntriesHandle( - object_storage, persistent_components, ctx, log, entry, actual_table_state_snapshot.schema_id); + auto manifest_file_ptr = Iceberg::getManifestFile( + object_storage, persistent_components, ctx, log, + entry.manifest_file_path, + entry.manifest_file_byte_size); } } @@ -684,6 +691,7 @@ void IcebergMetadata::mutate( catalog); } +<<<<<<< HEAD void IcebergMetadata::checkTableRootIsQueriedPath(std::string_view operation) const { if (!persistent_components.table_root_was_derived) @@ -699,6 +707,86 @@ void IcebergMetadata::checkTableRootIsQueriedPath(std::string_view operation) co persistent_components.table_path); } +======= +void IcebergMetadata::truncate(ContextPtr context, std::shared_ptr catalog, const StorageID & storage_id) +{ + if (!context->getSettingsRef()[Setting::allow_insert_into_iceberg].value) + throw Exception( + ErrorCodes::SUPPORT_IS_DISABLED, + "Iceberg truncate requires the allow_insert_into_iceberg setting to be enabled."); + + auto [actual_data_snapshot, actual_table_state_snapshot] = getRelevantState(context); + auto metadata_object = getMetadataJSONObject( + actual_table_state_snapshot.metadata_file_path, + object_storage, + persistent_components.metadata_cache, + context, + log, + persistent_components.metadata_compression_method, + persistent_components.table_uuid); + + // Use -1 as the Iceberg spec sentinel for "no parent snapshot" + // (distinct from snapshot ID 0 which is a valid snapshot). + Int64 parent_snapshot_id = actual_table_state_snapshot.snapshot_id.value_or(-1); + + // On antalya-26.6 all metadata paths flow through the table's IcebergPathResolver: + // FileNamesGenerator produces IcebergPathFromMetadata values (relative to the table + // location), and resolver.resolve() / resolveForCatalog() turn those into the storage + // path for I/O and the fully-qualified path the catalog expects. This mirrors the + // write path in IcebergStorageSink (see IcebergWrites.cpp) so transactional (REST) and + // non-transactional catalogs are handled uniformly. + const auto & resolver = persistent_components.path_resolver; + + bool is_transactional = (catalog != nullptr && catalog->isTransactional()); + + FileNamesGenerator filename_generator( + resolver.getTableLocation(), + is_transactional, + persistent_components.metadata_compression_method, + write_format); + + Int32 new_metadata_version = actual_table_state_snapshot.metadata_version + 1; + filename_generator.setVersion(new_metadata_version); + + auto metadata_info = filename_generator.generateMetadataPathWithInfo(); + + auto [new_snapshot, manifest_list_path] = MetadataGenerator(metadata_object).generateNextMetadata( + filename_generator, metadata_info.path, parent_snapshot_id, + /* added_files */ 0, /* added_records */ 0, /* added_files_size */ 0, + /* num_partitions */ 0, /* added_delete_files */ 0, /* num_deleted_rows */ 0, + std::nullopt, std::nullopt, /*is_truncate=*/true); + + auto storage_manifest_list_name = resolver.resolve(manifest_list_path); + + auto write_settings = context->getWriteSettings(); + auto buf = object_storage->writeObject( + StoredObject(storage_manifest_list_name), + WriteMode::Rewrite, std::nullopt, + DBMS_DEFAULT_BUFFER_SIZE, write_settings); + + // Truncate writes a metadata-only overwrite snapshot: an empty manifest list + // (no manifest entries, no sizes) that supersedes all previous snapshots. + generateManifestList(resolver, metadata_object, object_storage, + context, {}, new_snapshot, {}, *buf, Iceberg::FileContentType::DATA, /*use_previous_snapshots=*/false); + buf->finalize(); + + String metadata_content = dumpMetadataObjectToString(metadata_object); + writeMessageToFile(metadata_content, resolver.resolve(metadata_info.path), object_storage, + context, "*", "", persistent_components.metadata_compression_method); + + if (catalog) + { + String catalog_filename = resolver.resolveForCatalog(metadata_info.path); + + const auto & [namespace_name, table_name] = DataLake::parseTableName(storage_id.getTableName()); + if (!catalog->updateMetadata(namespace_name, table_name, catalog_filename, new_snapshot)) + throw Exception(ErrorCodes::INCORRECT_DATA, + "Failed to commit Iceberg truncate update to catalog."); + } +} + + +>>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) void IcebergMetadata::checkMutationIsPossible(const MutationCommands & commands) { checkTableRootIsQueriedPath("Mutation"); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h index efd9c0d60029..60455c182fdc 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h @@ -106,6 +106,8 @@ class IcebergMetadata : public IDataLakeMetadata bool supportsUpdate() const override { return true; } bool supportsWrites() const override { return true; } bool supportsParallelInsert() const override { return true; } + bool supportsTruncate() const override { return true; } + void truncate(ContextPtr context, std::shared_ptr catalog, const StorageID & storage_id) override; IcebergHistory getHistory(ContextPtr local_context) const; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index 728f8c17bc51..69f61ca28bca 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -701,6 +701,31 @@ void generateManifestFile( writer.close(); } +// Avro uses zigzag encoding for integers to efficiently represent small negative +// numbers. Positive n maps to 2n, negative n maps to 2(-n)-1, keeping small +// magnitudes compact regardless of sign. The value is then serialized as a +// variable-length base-128 integer (little-endian), where the high bit of each +// byte signals whether more bytes follow. +// See: https://avro.apache.org/docs/1.11.1/specification/#binary-encoding +static void writeAvroLong(WriteBuffer & out, int64_t val) +{ + uint64_t n = (static_cast(val) << 1) ^ static_cast(val >> 63); + while (n & ~0x7fULL) + { + char c = static_cast((n & 0x7f) | 0x80); + out.write(&c, 1); + n >>= 7; + } + char c = static_cast(n); + out.write(&c, 1); +} + +static void writeAvroBytes(WriteBuffer & out, const String & s) +{ + writeAvroLong(out, static_cast(s.size())); + out.write(s.data(), s.size()); +} + void generateManifestList( const Iceberg::IcebergPathResolver & path_resolver, Poco::JSON::Object::Ptr metadata, @@ -740,6 +765,38 @@ void generateManifestList( else schema_representation = manifest_list_v2_schema; + // For empty manifest list (e.g. TRUNCATE), write a valid Avro container + // file manually so we can embed the full schema JSON with field-ids intact, + // without triggering the DataFileWriter constructor's eager writeHeader() + // which commits encoder state before we can override avro.schema. + if (manifest_entry_names.empty() && !use_previous_snapshots) + { + // For an empty manifest list (e.g. after TRUNCATE), we write a minimal valid + // Avro Object Container File manually rather than using avro::DataFileWriter. + // The reason: DataFileWriter calls writeHeader() eagerly in its constructor, + // committing the binary encoder state. Post-construction setMetadata() calls + // corrupt StreamWriter::next_ causing a NULL dereference on close(). Writing + // the OCF header directly ensures the full schema JSON (with Iceberg field-ids) + // is embedded intact — the Avro C++ library strips unknown field properties + // like field-id during schema node serialization. + // Avro OCF format: [magic(4)] [metadata_map] [sync_marker(16)] [no data blocks] + buf.write("Obj\x01", 4); + + writeAvroLong(buf, 2); // 2 metadata entries + writeAvroBytes(buf, "avro.codec"); + writeAvroBytes(buf, "null"); + writeAvroBytes(buf, "avro.schema"); + writeAvroBytes(buf, schema_representation); // full JSON with field-ids intact + + writeAvroLong(buf, 0); // end of metadata map + + static const char sync_marker[16] = {}; + buf.write(sync_marker, 16); + + buf.finalize(); + return; + } + auto schema = avro::compileJsonSchemaFromString(schema_representation); // NOLINT auto adapter = std::make_unique(buf); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index 812a6c62ef03..5706090f3573 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -176,7 +176,11 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( Int64 num_deleted_rows, std::optional user_defined_snapshot_id, std::optional user_defined_timestamp, +<<<<<<< HEAD SnapshotOperation operation) +======= + bool is_truncate) +>>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) { int format_version = metadata_object->getValue(Iceberg::f_format_version); @@ -220,6 +224,7 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( auto parent_snapshot = getParentSnapshot(parent_snapshot_id); Poco::JSON::Object::Ptr summary = new Poco::JSON::Object; +<<<<<<< HEAD /// A merge-on-read DELETE writes position-delete files (num_deleted_rows != 0): per the Iceberg /// spec that snapshot is an `overwrite`, not an `append`. Compaction passes `Replace` explicitly. const char * operation_name = Iceberg::f_append; @@ -233,12 +238,25 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( summary->set(Iceberg::f_added_files_size, std::to_string(added_files_size)); summary->set(Iceberg::f_changed_partition_count, std::to_string(num_partitions)); if (num_deleted_rows != 0) +======= + if (is_truncate) + { + summary->set(Iceberg::f_operation, Iceberg::f_overwrite); + Int32 prev_total_records = parent_snapshot && parent_snapshot->has(Iceberg::f_summary) && parent_snapshot->getObject(Iceberg::f_summary)->has(Iceberg::f_total_records) ? std::stoi(parent_snapshot->getObject(Iceberg::f_summary)->getValue(Iceberg::f_total_records)) : 0; + Int32 prev_total_data_files = parent_snapshot && parent_snapshot->has(Iceberg::f_summary) && parent_snapshot->getObject(Iceberg::f_summary)->has(Iceberg::f_total_data_files) ? std::stoi(parent_snapshot->getObject(Iceberg::f_summary)->getValue(Iceberg::f_total_data_files)) : 0; + + summary->set(Iceberg::f_deleted_records, std::to_string(prev_total_records)); + summary->set(Iceberg::f_deleted_data_files, std::to_string(prev_total_data_files)); + } + else if (num_deleted_rows == 0) +>>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) { summary->set(Iceberg::f_added_delete_files, std::to_string(added_delete_files)); summary->set(Iceberg::f_added_position_delete_files, std::to_string(added_delete_files)); summary->set(Iceberg::f_added_position_deletes, std::to_string(num_deleted_rows)); } +<<<<<<< HEAD setSnapshotTotals( summary, parent_snapshot, @@ -248,6 +266,25 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( /*added_delete_files=*/added_delete_files, /*added_position_deletes=*/num_deleted_rows, /*added_equality_deletes=*/0); +======= + auto sum_with_parent_snapshot = [&](const char * field_name, Int64 snapshot_value) + { + if (is_truncate) + { + summary->set(field_name, std::to_string(0)); + return; + } + Int64 prev_value = parent_snapshot && parent_snapshot->has(Iceberg::f_summary) && parent_snapshot->getObject(Iceberg::f_summary)->has(field_name) ? parse(parent_snapshot->getObject(Iceberg::f_summary)->getValue(field_name)) : 0; + summary->set(field_name, std::to_string(prev_value + snapshot_value)); + }; + + sum_with_parent_snapshot(Iceberg::f_total_records, added_records); + sum_with_parent_snapshot(Iceberg::f_total_files_size, added_files_size); + sum_with_parent_snapshot(Iceberg::f_total_data_files, added_files); + sum_with_parent_snapshot(Iceberg::f_total_delete_files, added_delete_files); + sum_with_parent_snapshot(Iceberg::f_total_position_deletes, num_deleted_rows); + sum_with_parent_snapshot(Iceberg::f_total_equality_deletes, 0); +>>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) new_snapshot->set(Iceberg::f_summary, summary); new_snapshot->set(Iceberg::f_schema_id, metadata_object->getValue(Iceberg::f_current_schema_id)); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h index 5182d76921ff..c7772d0bc55b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h @@ -44,6 +44,7 @@ class MetadataGenerator Int64 num_deleted_rows, std::optional user_defined_snapshot_id = std::nullopt, std::optional user_defined_timestamp = std::nullopt, +<<<<<<< HEAD SnapshotOperation operation = SnapshotOperation::Append); /// Create a manifest-only rewrite snapshot (`replace` operation) carrying `total-*` counters forward so `OPTIMIZE ... MANIFEST` is idempotent. @@ -51,6 +52,9 @@ class MetadataGenerator FileNamesGenerator & generator, const Iceberg::IcebergPathFromMetadata & metadata_file_path, Int64 parent_snapshot_id); +======= + bool is_truncate = false); +>>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) void generateAddColumnMetadata(const String & column_name, DataTypePtr type); void generateDropColumnMetadata(const String & column_name); diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index 8a76fa0ea73b..4d9c1472a43f 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -830,7 +830,7 @@ bool StorageObjectStorage::optimize( void StorageObjectStorage::truncate( const ASTPtr & /* query */, const StorageMetadataPtr & /* metadata_snapshot */, - ContextPtr /* context */, + ContextPtr local_context, TableExclusiveLockHolder & /* table_holder */) { const auto path = configuration->getRawPath(); @@ -844,8 +844,12 @@ void StorageObjectStorage::truncate( if (configuration->isDataLakeConfiguration()) { - throw Exception(ErrorCodes::NOT_IMPLEMENTED, - "Truncate is not supported for data lake engine"); + auto * data_lake_metadata = getExternalMetadata(local_context); + if (!data_lake_metadata || !data_lake_metadata->supportsTruncate()) + throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Truncate is not supported for this data lake engine"); + + data_lake_metadata->truncate(local_context, catalog, getStorageID()); + return; } if (path.hasGlobsIgnorePlaceholders()) diff --git a/tests/integration/test_storage_iceberg_no_spark/test_iceberg_truncate.py b/tests/integration/test_storage_iceberg_no_spark/test_iceberg_truncate.py new file mode 100644 index 000000000000..cf8d77ee6a54 --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_iceberg_truncate.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 + +from pyiceberg.catalog import load_catalog +from helpers.config_cluster import minio_secret_key, minio_access_key +import uuid +import pyarrow as pa +from pyiceberg.schema import Schema, NestedField +from pyiceberg.types import LongType, StringType +from pyiceberg.partitioning import PartitionSpec + +CATALOG_NAME = "demo" + +def load_catalog_impl(started_cluster): + return load_catalog( + CATALOG_NAME, + **{ + "uri": f"http://localhost:{started_cluster.iceberg_rest_catalog_port}", + "type": "rest", + "s3.endpoint": f"http://{started_cluster.minio_ip}:{started_cluster.minio_port}", + "s3.access-key-id": minio_access_key, + "s3.secret-access-key": minio_secret_key, + }, + ) + + +def test_iceberg_truncate_restart(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + catalog = load_catalog_impl(started_cluster_iceberg_no_spark) + + namespace = f"clickhouse_truncate_restart_{uuid.uuid4().hex}" + catalog.create_namespace(namespace) + + schema = Schema( + NestedField(field_id=1, name="id", field_type=LongType(), required=False), + NestedField(field_id=2, name="val", field_type=StringType(), required=False), + ) + table_name = "test_truncate_restart" + catalog.create_table( + identifier=f"{namespace}.{table_name}", + schema=schema, + location=f"s3://warehouse-rest/{namespace}.{table_name}", + partition_spec=PartitionSpec(), + ) + + ch_table_identifier = f"`{namespace}.{table_name}`" + + instance.query(f"DROP DATABASE IF EXISTS {namespace}") + instance.query( + f""" + CREATE DATABASE {namespace} ENGINE = DataLakeCatalog('http://rest:8181/v1', 'minio', '{minio_secret_key}') + SETTINGS + catalog_type='rest', + warehouse='demo', + storage_endpoint='http://minio1:9001/warehouse-rest'; + """, + settings={"allow_database_iceberg": 1} + ) + + # 1. Insert initial data and truncate + df = pa.Table.from_pylist([{"id": 1, "val": "A"}, {"id": 2, "val": "B"}]) + catalog.load_table(f"{namespace}.{table_name}").append(df) + + assert int(instance.query(f"SELECT count() FROM {namespace}.{ch_table_identifier}").strip()) == 2 + + instance.query( + f"TRUNCATE TABLE {namespace}.{ch_table_identifier}", + settings={"allow_experimental_insert_into_iceberg": 1} + ) + assert int(instance.query(f"SELECT count() FROM {namespace}.{ch_table_identifier}").strip()) == 0 + + # 2. Restart ClickHouse and verify table is still readable (count = 0) + instance.restart_clickhouse() + assert int(instance.query(f"SELECT count() FROM {namespace}.{ch_table_identifier}").strip()) == 0 + + # 3. Insert new data after restart and verify it's readable + new_df = pa.Table.from_pylist([{"id": 3, "val": "C"}]) + catalog.load_table(f"{namespace}.{table_name}").append(new_df) + assert int(instance.query(f"SELECT count() FROM {namespace}.{ch_table_identifier}").strip()) == 1 + + instance.query(f"DROP DATABASE {namespace}") From 398cf3d57e2605060defb5fc23a46f11edcaea78 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Tue, 15 Sep 2026 19:22:34 +0200 Subject: [PATCH 2/2] Resolve conflicts in cherry-pick of #2125 Kept antalya-26.8's `SnapshotOperation` parameter, `setSnapshotTotals` helper and `generateManifestOnlySnapshot`, and layered the PR's `is_truncate` handling on top of them. Adapted: `generateNextMetadata` gained `bool is_truncate` after antalya-26.8's `SnapshotOperation operation` parameter (existing callers pass `operation` positionally); the truncate call site passes `SnapshotOperation::Append` explicitly Adapted: the PR's truncate `overwrite` operation and zeroed `total-*` counters expressed in antalya-26.8's `operation_name` / `setSnapshotTotals` shape --- .../DataLakes/Iceberg/IcebergMetadata.cpp | 9 +-- .../DataLakes/Iceberg/MetadataGenerator.cpp | 61 +++++++------------ .../DataLakes/Iceberg/MetadataGenerator.h | 7 +-- 3 files changed, 25 insertions(+), 52 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 7f3f45c5259d..e6b42429c724 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -121,11 +121,8 @@ extern const int ICEBERG_SPECIFICATION_VIOLATION; extern const int S3_ERROR; extern const int TABLE_ALREADY_EXISTS; extern const int SUPPORT_IS_DISABLED; -<<<<<<< HEAD extern const int FILE_ALREADY_EXISTS; -======= extern const int INCORRECT_DATA; ->>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) } namespace Setting @@ -691,7 +688,6 @@ void IcebergMetadata::mutate( catalog); } -<<<<<<< HEAD void IcebergMetadata::checkTableRootIsQueriedPath(std::string_view operation) const { if (!persistent_components.table_root_was_derived) @@ -707,7 +703,6 @@ void IcebergMetadata::checkTableRootIsQueriedPath(std::string_view operation) co persistent_components.table_path); } -======= void IcebergMetadata::truncate(ContextPtr context, std::shared_ptr catalog, const StorageID & storage_id) { if (!context->getSettingsRef()[Setting::allow_insert_into_iceberg].value) @@ -754,7 +749,7 @@ void IcebergMetadata::truncate(ContextPtr context, std::shared_ptr>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) void IcebergMetadata::checkMutationIsPossible(const MutationCommands & commands) { checkTableRootIsQueriedPath("Mutation"); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index 5706090f3573..bc80cd67c9fc 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -176,11 +176,8 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( Int64 num_deleted_rows, std::optional user_defined_snapshot_id, std::optional user_defined_timestamp, -<<<<<<< HEAD - SnapshotOperation operation) -======= + SnapshotOperation operation, bool is_truncate) ->>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) { int format_version = metadata_object->getValue(Iceberg::f_format_version); @@ -224,67 +221,53 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( auto parent_snapshot = getParentSnapshot(parent_snapshot_id); Poco::JSON::Object::Ptr summary = new Poco::JSON::Object; -<<<<<<< HEAD /// A merge-on-read DELETE writes position-delete files (num_deleted_rows != 0): per the Iceberg /// spec that snapshot is an `overwrite`, not an `append`. Compaction passes `Replace` explicitly. + /// A TRUNCATE drops every data file, which per the spec is an `overwrite` as well. const char * operation_name = Iceberg::f_append; if (operation == SnapshotOperation::Replace) operation_name = Iceberg::f_replace; - else if (num_deleted_rows != 0) + else if (is_truncate || num_deleted_rows != 0) operation_name = Iceberg::f_overwrite; summary->set(Iceberg::f_operation, operation_name); summary->set(Iceberg::f_added_data_files, std::to_string(added_files)); summary->set(Iceberg::f_added_records, std::to_string(added_records)); summary->set(Iceberg::f_added_files_size, std::to_string(added_files_size)); summary->set(Iceberg::f_changed_partition_count, std::to_string(num_partitions)); - if (num_deleted_rows != 0) -======= if (is_truncate) { - summary->set(Iceberg::f_operation, Iceberg::f_overwrite); Int32 prev_total_records = parent_snapshot && parent_snapshot->has(Iceberg::f_summary) && parent_snapshot->getObject(Iceberg::f_summary)->has(Iceberg::f_total_records) ? std::stoi(parent_snapshot->getObject(Iceberg::f_summary)->getValue(Iceberg::f_total_records)) : 0; Int32 prev_total_data_files = parent_snapshot && parent_snapshot->has(Iceberg::f_summary) && parent_snapshot->getObject(Iceberg::f_summary)->has(Iceberg::f_total_data_files) ? std::stoi(parent_snapshot->getObject(Iceberg::f_summary)->getValue(Iceberg::f_total_data_files)) : 0; summary->set(Iceberg::f_deleted_records, std::to_string(prev_total_records)); summary->set(Iceberg::f_deleted_data_files, std::to_string(prev_total_data_files)); } - else if (num_deleted_rows == 0) ->>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) + if (num_deleted_rows != 0) { summary->set(Iceberg::f_added_delete_files, std::to_string(added_delete_files)); summary->set(Iceberg::f_added_position_delete_files, std::to_string(added_delete_files)); summary->set(Iceberg::f_added_position_deletes, std::to_string(num_deleted_rows)); } -<<<<<<< HEAD - setSnapshotTotals( - summary, - parent_snapshot, - /*added_records=*/added_records, - /*added_files_size=*/added_files_size, - /*added_data_files=*/added_files, - /*added_delete_files=*/added_delete_files, - /*added_position_deletes=*/num_deleted_rows, - /*added_equality_deletes=*/0); -======= - auto sum_with_parent_snapshot = [&](const char * field_name, Int64 snapshot_value) + if (is_truncate) { - if (is_truncate) - { - summary->set(field_name, std::to_string(0)); - return; - } - Int64 prev_value = parent_snapshot && parent_snapshot->has(Iceberg::f_summary) && parent_snapshot->getObject(Iceberg::f_summary)->has(field_name) ? parse(parent_snapshot->getObject(Iceberg::f_summary)->getValue(field_name)) : 0; - summary->set(field_name, std::to_string(prev_value + snapshot_value)); - }; - - sum_with_parent_snapshot(Iceberg::f_total_records, added_records); - sum_with_parent_snapshot(Iceberg::f_total_files_size, added_files_size); - sum_with_parent_snapshot(Iceberg::f_total_data_files, added_files); - sum_with_parent_snapshot(Iceberg::f_total_delete_files, added_delete_files); - sum_with_parent_snapshot(Iceberg::f_total_position_deletes, num_deleted_rows); - sum_with_parent_snapshot(Iceberg::f_total_equality_deletes, 0); ->>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) + summary->set(Iceberg::f_total_records, std::to_string(0)); + summary->set(Iceberg::f_total_files_size, std::to_string(0)); + summary->set(Iceberg::f_total_data_files, std::to_string(0)); + summary->set(Iceberg::f_total_delete_files, std::to_string(0)); + summary->set(Iceberg::f_total_position_deletes, std::to_string(0)); + summary->set(Iceberg::f_total_equality_deletes, std::to_string(0)); + } + else + setSnapshotTotals( + summary, + parent_snapshot, + /*added_records=*/added_records, + /*added_files_size=*/added_files_size, + /*added_data_files=*/added_files, + /*added_delete_files=*/added_delete_files, + /*added_position_deletes=*/num_deleted_rows, + /*added_equality_deletes=*/0); new_snapshot->set(Iceberg::f_summary, summary); new_snapshot->set(Iceberg::f_schema_id, metadata_object->getValue(Iceberg::f_current_schema_id)); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h index c7772d0bc55b..522a2792b80f 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h @@ -44,17 +44,14 @@ class MetadataGenerator Int64 num_deleted_rows, std::optional user_defined_snapshot_id = std::nullopt, std::optional user_defined_timestamp = std::nullopt, -<<<<<<< HEAD - SnapshotOperation operation = SnapshotOperation::Append); + SnapshotOperation operation = SnapshotOperation::Append, + bool is_truncate = false); /// Create a manifest-only rewrite snapshot (`replace` operation) carrying `total-*` counters forward so `OPTIMIZE ... MANIFEST` is idempotent. NextMetadataResult generateManifestOnlySnapshot( FileNamesGenerator & generator, const Iceberg::IcebergPathFromMetadata & metadata_file_path, Int64 parent_snapshot_id); -======= - bool is_truncate = false); ->>>>>>> 911f0dcb139 (Merge pull request #2125 from Altinity/feature/antalya-26.6/pr-1655) void generateAddColumnMetadata(const String & column_name, DataTypePtr type); void generateDropColumnMetadata(const String & column_name);