diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h index 8347164b9a464..2bf6ed0f136d9 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 2af675f6cf10a..e6b42429c7249 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 @@ -121,6 +122,7 @@ extern const int S3_ERROR; extern const int TABLE_ALREADY_EXISTS; extern const int SUPPORT_IS_DISABLED; extern const int FILE_ALREADY_EXISTS; +extern const int INCORRECT_DATA; } namespace Setting @@ -294,15 +296,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); } } @@ -699,6 +703,83 @@ 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, MetadataGenerator::SnapshotOperation::Append, /*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."); + } +} + 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 efd9c0d60029e..60455c182fdc3 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 a83d8b5f3278a..29159f71b951e 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -720,6 +720,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, @@ -759,6 +784,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 812a6c62ef038..bc80cd67c9fc3 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -176,7 +176,8 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( Int64 num_deleted_rows, std::optional user_defined_snapshot_id, std::optional user_defined_timestamp, - SnapshotOperation operation) + SnapshotOperation operation, + bool is_truncate) { int format_version = metadata_object->getValue(Iceberg::f_format_version); @@ -222,16 +223,25 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( Poco::JSON::Object::Ptr summary = new Poco::JSON::Object; /// 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 (is_truncate) + { + 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)); + } if (num_deleted_rows != 0) { summary->set(Iceberg::f_added_delete_files, std::to_string(added_delete_files)); @@ -239,15 +249,25 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( summary->set(Iceberg::f_added_position_deletes, std::to_string(num_deleted_rows)); } - 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); + if (is_truncate) + { + 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 5182d76921ff3..522a2792b80fb 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h @@ -44,7 +44,8 @@ class MetadataGenerator Int64 num_deleted_rows, std::optional user_defined_snapshot_id = std::nullopt, std::optional user_defined_timestamp = std::nullopt, - 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( diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index 3ff032164a9a6..8328f4b2adbaf 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -838,7 +838,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(); @@ -852,8 +852,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 0000000000000..cf8d77ee6a54d --- /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}")