From d73e8821d6aec9cc348fcad446ebb6a8191b7a87 Mon Sep 17 00:00:00 2001 From: Andrew Xie Date: Mon, 12 Oct 2026 13:03:51 -0400 Subject: [PATCH] Implement additional Iceberg drop partition features Add support for catalog tables, purging data files, mixed manifests, equality deletes, and partition spec evolution for Iceberg `DROP PARTITION`. Signed-off-by: Andrew Xie --- docs/reference/statements/alter/partition.mdx | 16 + .../Iceberg/AlterDropPartitionExecutor.cpp | 630 +++++++++++--- .../Iceberg/AlterDropPartitionExecutor.h | 50 +- .../DataLakes/Iceberg/IcebergMetadata.cpp | 16 +- .../DataLakes/Iceberg/IcebergMetadata.h | 11 +- .../DataLakes/Iceberg/IcebergWrites.cpp | 31 +- .../DataLakes/Iceberg/IcebergWrites.h | 14 +- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 10 +- .../ObjectStorage/DataLakes/Iceberg/Utils.h | 3 +- .../registerStorageObjectStorage.cpp | 8 +- .../test_drop_partition.py | 779 ++++++++++++++++++ .../test_drop_partition_whole_manifest.py | 37 +- 12 files changed, 1479 insertions(+), 126 deletions(-) create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_drop_partition.py diff --git a/docs/reference/statements/alter/partition.mdx b/docs/reference/statements/alter/partition.mdx index d135b7b220ecf..6a54c4177ddf5 100644 --- a/docs/reference/statements/alter/partition.mdx +++ b/docs/reference/statements/alter/partition.mdx @@ -68,6 +68,22 @@ ALTER TABLE mt DROP PARTITION '2020-11-21'; ALTER TABLE mt DROP PART 'all_4_4_0'; ``` +On Iceberg tables, `DROP PARTITION` drops the partition by writing a new snapshot with the partition data files excluded. + +By default, no parquet files are physically deleted from object storage. +The `data_lake_delete_data_on_drop` option must be enabled for data files to be purged. + +:::note +Purging data files will break time travel to older snapshots that reference that partition. +::: + +Example purging data files: + +```sql +ALTER TABLE mt DROP PARTITION tuple(toRelativeDayNum(toDate('2026-01-01'))) +SETTINGS data_lake_delete_data_on_drop = 1; +``` + ## DROP DETACHED PARTITION\|PART {#drop-detached-partitionpart} ```sql diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp index faf3b30e2a176..6a006e5e95650 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp @@ -17,8 +17,11 @@ #include #include #include +#include +#include #include #include +#include #include #include #include @@ -26,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -35,6 +39,7 @@ #include #include +#include #include @@ -55,12 +60,15 @@ extern const int LOGICAL_ERROR; extern const int NOT_IMPLEMENTED; extern const int ICEBERG_SPECIFICATION_VIOLATION; extern const int CONCURRENT_ACCESS_NOT_SUPPORTED; +extern const int METADATA_MISMATCH; +extern const int ACCESS_DENIED; } namespace Setting { extern const SettingsInt64 iceberg_snapshot_id; extern const SettingsInt64 iceberg_timestamp_ms; +extern const SettingsBool data_lake_delete_data_on_drop; } namespace DataLakeStorageSetting @@ -84,15 +92,56 @@ static constexpr auto MAX_TRANSACTION_RETRIES = 100; namespace { -bool partitionEquals(const Row & lhs, const Row & rhs) +/// Find the partition spec object with the given spec-id inside a metadata JSON document. +/// Throws METADATA_MISMATCH if the spec is not found (indicates metadata/spec-id mismatch). +Poco::JSON::Object::Ptr lookupPartitionSpec(const Poco::JSON::Object::Ptr & meta, Int64 spec_id) { - if (lhs.size() != rhs.size()) - return false; - /// `icebergBucket` returns `UInt32`, but manifest bucket values are `Int32`. - for (size_t i = 0; i < lhs.size(); ++i) - if (!accurateEquals(lhs[i], rhs[i])) - return false; - return true; + auto specs = meta->getArray(f_partition_specs); + for (size_t i = 0; i < specs->size(); ++i) + { + auto spec = specs->getObject(static_cast(i)); + if (spec->getValue(f_spec_id) == spec_id) + return spec; + } + throw Exception(ErrorCodes::METADATA_MISMATCH, + "Partition spec with id {} not found in table metadata", spec_id); +} + +/// One term per partition field in spec order +PartitionSpecSignature getPartitionSpecSignature(const Poco::JSON::Array::Ptr & spec_fields) +{ + PartitionSpecSignature signature; + signature.reserve(spec_fields->size()); + for (UInt32 i = 0; i < spec_fields->size(); ++i) + { + auto spec_field = spec_fields->getObject(i); + signature.emplace_back( + spec_field->getValue(f_source_id), + Poco::toLower(spec_field->getValue(f_partition_transform))); + } + return signature; +} + +constexpr size_t PARTITION_FIELD_NOT_IN_SPEC = std::numeric_limits::max(); + +/// Position of every target spec field inside a file's own partition tuple +std::vector mapTargetFieldsToSpecPositions( + const PartitionSpecification & entry_spec, const PartitionSpecSignature & signature) +{ + std::vector positions(signature.size(), PARTITION_FIELD_NOT_IN_SPEC); + for (size_t i = 0; i < signature.size(); ++i) + { + for (size_t j = 0; j < entry_spec.size(); ++j) + { + if (entry_spec[j].source_id == signature[i].column_id + && Poco::toLower(entry_spec[j].transform_name) == signature[i].transform) + { + positions[i] = j; + break; + } + } + } + return positions; } void validateDropPartitionAST(const ASTPartition & ast, const PartitionCommand & command) @@ -168,6 +217,51 @@ Row parsePartitionTuple(const IAST & value_ast, const DataTypes & partition_type return out; } +Block getPartitionSourceHeader( + const Poco::JSON::Array::Ptr & spec_fields, + const Poco::JSON::Array::Ptr & schema_fields, + const IcebergSchemaProcessor & schema_processor, + Int32 schema_id) +{ + std::unordered_map column_name_by_source_id; + for (UInt32 i = 0; i < schema_fields->size(); ++i) + { + auto schema_field = schema_fields->getObject(i); + column_name_by_source_id[schema_field->getValue(f_id)] = schema_field->getValue(f_name); + } + + ColumnsWithTypeAndName columns; + std::unordered_set used_column_names; + for (UInt32 i = 0; i < spec_fields->size(); ++i) + { + auto spec_field = spec_fields->getObject(i); + auto source_id = spec_field->getValue(f_source_id); + + auto name_it = column_name_by_source_id.find(source_id); + if (name_it == column_name_by_source_id.end()) + throw Exception(ErrorCodes::NOT_IMPLEMENTED, + "DROP PARTITION is not supported for a partition spec whose source column (field id {}) is not a " + "top-level column of the current table schema", source_id); + + const auto & column_name = name_it->second; + + if (!used_column_names.insert(column_name).second) + throw Exception(ErrorCodes::NOT_IMPLEMENTED, + "DROP PARTITION is not supported for a partition spec that uses column '{}' in more than one " + "partition field", column_name); + + auto name_and_type = schema_processor.tryGetFieldCharacteristics(schema_id, source_id); + if (!name_and_type) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "Partition source column '{}' (field id {}) is missing from Iceberg schema {}", + column_name, source_id, schema_id); + + columns.emplace_back(name_and_type->type, column_name); + } + + return Block(std::move(columns)); +} + DataTypes resolvePartitionTypes( const Poco::JSON::Object & partition_spec, const Poco::JSON::Object & current_schema, @@ -176,36 +270,104 @@ DataTypes resolvePartitionTypes( ContextPtr context) { auto partition_fields = partition_spec.getArray(f_fields); + auto schema_fields = current_schema.getArray(f_fields); - std::vector source_ids; - for (size_t i = 0; i < partition_fields->size(); ++i) + if (!schema_fields || schema_fields->size() == 0) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Could not find key '{}' in schema {} or fields is empty", f_fields, schema_id); + + SharedHeader sample_block + = std::make_shared(getPartitionSourceHeader(partition_fields, schema_fields, schema_processor, schema_id)); + + ChunkPartitioner partitioner(partition_fields, schema_fields, context, sample_block); + return partitioner.getResultTypes(); +} + +struct ManifestRewriteResult +{ + Int32 surviving_entries = 0; + Int64 surviving_rows = 0; + Int64 removed_entries = 0; + Int64 min_sequence_number = 0; +}; + +/// Copy a manifest file without the entries whose `data_file.file_path` is in excluded_file_paths +ManifestRewriteResult rewriteManifestFileExcludingFiles( + ReadBuffer & source, + const std::unordered_set & excluded_file_paths, + Int64 inherited_sequence_number, + Int64 inherited_snapshot_id, + WriteBuffer & out) +{ + auto input_stream = std::make_unique(source); + auto reader_base = std::make_unique(std::move(input_stream), MAX_AVRO_SCHEMA_DEPTH); + + auto * reader_base_ptr = reader_base.get(); + avro::DataFileReader reader(std::move(reader_base)); + + const avro::ValidSchema & schema = reader.dataSchema(); + + auto adapter = std::make_unique(out); + avro::DataFileWriter writer(std::move(adapter), schema); + + for (const auto & [key, value] : reader_base_ptr->metadata()) { - auto field_object = partition_fields->getObject(static_cast(i)); - auto field_source_id = field_object->getValue(f_source_id); - source_ids.emplace_back(field_source_id); + if (key.starts_with("avro.") && key != "avro.schema") + continue; + writer.setMetadata(key, String(value.begin(), value.end())); } - auto names_and_types = schema_processor.tryGetFieldsCharacteristics(schema_id, source_ids); - if (names_and_types.size() != source_ids.size()) - throw Exception( - ErrorCodes::BAD_ARGUMENTS, - "Could not resolve all partition source columns against schema {} (got {}/{} fields)", - schema_id, - names_and_types.size(), - source_ids.size()); + ManifestRewriteResult result; + Int64 min_sequence_number = std::numeric_limits::max(); + + avro::GenericDatum datum(schema); + while (reader.read(datum)) + { + avro::GenericRecord & entry = datum.value(); + avro::GenericRecord & data_file = entry.field(f_data_file).value(); - Block block; - for (const auto & [name, type] : names_and_types) - block.insert(ColumnWithTypeAndName{nullptr, type, name}); + /// Copying a DELETED entry would revive it, since every entry below is restamped EXISTING. + if (entry.field(f_status).value() == static_cast(ManifestEntryStatus::DELETED)) + continue; - SharedHeader sample_block = std::make_shared(std::move(block)); - auto schema_fields = current_schema.getArray(f_fields); + if (excluded_file_paths.contains(data_file.field(f_file_path).value())) + { + ++result.removed_entries; + continue; + } - if (!schema_fields || schema_fields->size() == 0) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Could not find key '{}' in schema {} or fields is empty", f_fields, schema_id); + auto materialize = [&](const String & field_name, Int64 inherited_value) -> Int64 + { + if (!entry.hasField(field_name)) + return inherited_value; - ChunkPartitioner partitioner(partition_fields, schema_fields, context, sample_block); - return partitioner.getResultTypes(); + auto & field = entry.field(field_name); + if (!field.isUnion()) + return field.value(); + if (field.unionBranch() != 0) + return field.value(); + + field.selectBranch(1); + field.value() = inherited_value; + return inherited_value; + }; + + const Int64 sequence_number = materialize(f_sequence_number, inherited_sequence_number); + materialize(f_file_sequence_number, inherited_sequence_number); + materialize(f_snapshot_id, inherited_snapshot_id); + + entry.field(f_status) = avro::GenericDatum(static_cast(ManifestEntryStatus::EXISTING)); + + ++result.surviving_entries; + result.surviving_rows += data_file.field(f_record_count).value(); + min_sequence_number = std::min(min_sequence_number, sequence_number); + + writer.write(datum); + } + + writer.close(); + + result.min_sequence_number = result.surviving_entries > 0 ? min_sequence_number : inherited_sequence_number; + return result; } } @@ -219,7 +381,9 @@ AlterDropPartitionExecutor::AlterDropPartitionExecutor( const PersistentTableComponents & components_, const DataLakeStorageSettings & data_lake_settings_, String write_format_, - LoggerPtr log_) + LoggerPtr log_, + std::shared_ptr catalog_, + const StorageID & storage_id_) : command(command_) , metadata(metadata_) , context(context_) @@ -229,6 +393,9 @@ AlterDropPartitionExecutor::AlterDropPartitionExecutor( , data_lake_settings(data_lake_settings_) , write_format(std::move(write_format_)) , log(std::move(log_)) + , catalog(std::move(catalog_)) + , storage_id(storage_id_) + , purge(context->getSettingsRef()[Setting::data_lake_delete_data_on_drop]) { } @@ -237,7 +404,19 @@ std::optional AlterDropPartitionExecu SnapshotState state; { - auto [snapshot, table_state] = metadata.getRelevantState(context, /*force_fetch_latest_metadata=*/true); + const auto [metadata_version, metadata_file_path, compression_method] = getLatestMetadataFileAndVersionWithCatalog( + object_storage, + catalog, + storage_id.getTableName(), + components.table_path, + data_lake_settings, + components.metadata_cache, + context, + log.get(), + components.table_uuid, + components.metadata_compression_method); + + auto [snapshot, table_state] = metadata.getState(context, metadata_file_path, metadata_version); if (!snapshot) return std::nullopt; @@ -275,21 +454,7 @@ std::optional AlterDropPartitionExecu state.schema_id = static_cast(state.snapshot->schema_id_on_snapshot_commit); state.partition_spec_id = metadata_object->getValue(f_default_spec_id); - /// TODO: support different specs - if (specs->size() > 1) - throw Exception( - ErrorCodes::NOT_IMPLEMENTED, - "DROP PARTITION is not supported on Iceberg tables with evolved partition specs " - "({} specs in metadata)", - specs->size()); - - auto partition_spec = specs->getObject(0); - if (!partition_spec || partition_spec->getValue(f_spec_id) != state.partition_spec_id) - throw Exception( - ErrorCodes::BAD_ARGUMENTS, - "Default partition spec {} not found in metadata {}", - state.partition_spec_id, - state.table_state.metadata_file_path); + auto partition_spec = lookupPartitionSpec(metadata_object, state.partition_spec_id); if (!partition_spec->has(f_fields)) throw Exception( @@ -312,6 +477,8 @@ std::optional AlterDropPartitionExecu if (state.partition_columns.empty()) throw Exception(ErrorCodes::BAD_ARGUMENTS, "DROP PARTITION is not supported on unpartitioned Iceberg tables"); + state.spec_signature = getPartitionSpecSignature(partition_fields); + auto schemas = metadata_object->getArray(f_schemas); if (!schemas || schemas->size() == 0) throw Exception( @@ -320,11 +487,13 @@ std::optional AlterDropPartitionExecu f_schemas, state.table_state.metadata_file_path); + const auto current_schema_id = IcebergMetadata::parseTableSchema(metadata_object, *components.schema_processor, context, log); + Poco::JSON::Object::Ptr current_schema; for (size_t i = 0; schemas && i < schemas->size(); ++i) { auto schema = schemas->getObject(static_cast(i)); - if (!schema || schema->getValue(f_schema_id) != state.schema_id) + if (!schema || schema->getValue(f_schema_id) != current_schema_id) continue; current_schema = schema; @@ -335,11 +504,11 @@ std::optional AlterDropPartitionExecu throw Exception( ErrorCodes::BAD_ARGUMENTS, "Iceberg schema '{}' not found in metadata {} or empty", - state.schema_id, + current_schema_id, state.table_state.metadata_file_path); state.partition_types - = resolvePartitionTypes(*state.partition_spec, *current_schema, *components.schema_processor, state.schema_id, context); + = resolvePartitionTypes(*state.partition_spec, *current_schema, *components.schema_processor, current_schema_id, context); if (state.partition_types.size() != state.partition_columns.size()) throw Exception(ErrorCodes::LOGICAL_ERROR, "Partitions types count doesn't match with number of partition columns"); @@ -347,10 +516,63 @@ std::optional AlterDropPartitionExecu return state; } +const std::vector & AlterDropPartitionExecutor::getTargetFieldPositions( + const ProcessedManifestFileEntry & entry, const SnapshotState & state, SpecFieldPositions & spec_field_positions) const +{ + const auto & entry_spec = entry.common_partition_specification; + if (!entry_spec) + throw Exception(ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "File {} has no partition specification", entry.parsed_entry->file_path_key.serialize()); + + const auto & partition_key_value = entry.parsed_entry->partition_key_value; + if (partition_key_value.size() != entry_spec->size()) + throw Exception(ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "File {} has {} partition values, but its partition spec has {} fields", + entry.parsed_entry->file_path_key.serialize(), partition_key_value.size(), entry_spec->size()); + + auto [it, inserted] = spec_field_positions.try_emplace(entry_spec.get()); + if (inserted) + it->second = mapTargetFieldsToSpecPositions(*entry_spec, state.spec_signature); + + return it->second; +} + +bool AlterDropPartitionExecutor::isInTargetPartition( + const ProcessedManifestFileEntry & entry, + const SnapshotState & state, + const Row & target_partition, + SpecFieldPositions & spec_field_positions) const +{ + const auto & positions = getTargetFieldPositions(entry, state, spec_field_positions); + const auto & partition_key_value = entry.parsed_entry->partition_key_value; + + bool constrains_every_target_field = true; + for (size_t i = 0; i < state.spec_signature.size(); ++i) + { + const auto position = positions[i]; + if (position == PARTITION_FIELD_NOT_IN_SPEC) + { + constrains_every_target_field = false; + continue; + } + if (!accurateEquals(partition_key_value[position], target_partition[i])) + return false; + } + + if (!constrains_every_target_field) + throw Exception(ErrorCodes::NOT_IMPLEMENTED, + "DROP PARTITION is not supported for Iceberg table {}: file {} was written under a partition spec " + "that does not contain every field of the current spec ({}), so it spans more than the dropped partition", + components.table_path, entry.parsed_entry->file_path_key.serialize(), state.partition_spec_id); + + return true; +} + AlterDropPartitionExecutor::TargetFilePaths AlterDropPartitionExecutor::discoverTargetFilePaths(const SnapshotState & state, const Row & target_partition) const { TargetFilePaths targets; + SpecFieldPositions spec_field_positions; auto collect = [&](const std::vector & entries) { for (const auto & entry : entries) @@ -360,7 +582,7 @@ AlterDropPartitionExecutor::discoverTargetFilePaths(const SnapshotState & state, if (!parsed) throw Exception(ErrorCodes::LOGICAL_ERROR, "Manifest file entry is not parsed"); - if (partitionEquals(parsed->partition_key_value, target_partition)) + if (isInTargetPartition(*entry, state, target_partition, spec_field_positions)) targets.emplace(components.path_resolver.resolve(parsed->file_path_key)); } }; @@ -372,16 +594,7 @@ AlterDropPartitionExecutor::discoverTargetFilePaths(const SnapshotState & state, collect(handle.getFilesWithoutDeleted(FileContentType::DATA)); collect(handle.getFilesWithoutDeleted(FileContentType::POSITION_DELETE)); - - for (const auto & entry : handle.getFilesWithoutDeleted(FileContentType::EQUALITY_DELETE)) - { - if (!entry->parsed_entry) - throw Exception(ErrorCodes::LOGICAL_ERROR, "Manifest file entry is not parsed"); - - if (partitionEquals(entry->parsed_entry->partition_key_value, target_partition)) - throw Exception( - ErrorCodes::NOT_IMPLEMENTED, "DROP PARTITION is not supported when the selected partition has equality deletes"); - } + collect(handle.getFilesWithoutDeleted(FileContentType::EQUALITY_DELETE)); } return targets; @@ -398,32 +611,43 @@ AlterDropPartitionExecutor::buildDropPlan(const SnapshotState & state, const Tar UInt64 removed_files_size = 0; UInt64 removed_position_deletes = 0; UInt64 removed_position_delete_files = 0; + UInt64 removed_equality_deletes = 0; + UInt64 removed_equality_delete_files = 0; /// Paranoia. Partition's key values are converted from iceberg's manifest to clickhouse's representation /// We're trying to protect ourselves against ambiguous conversions which might lead to several partitions matches a single clickhouse's defintion Row partition_key_value_for_assert; - auto assert_on_ambigious_partition_key = [&](const auto & parsed_entry) + SpecFieldPositions spec_field_positions; + auto assert_on_ambigious_partition_key = [&](const ProcessedManifestFileEntry & entry) { - if (!partitionEquals(parsed_entry.partition_key_value, target_partition)) + const auto & parsed_entry = *entry.parsed_entry; + if (!isInTargetPartition(entry, state, target_partition, spec_field_positions)) throw Exception( ErrorCodes::LOGICAL_ERROR, "Manifest file entry partition value ({}) is different from target ({})", fmt::join(parsed_entry.partition_key_value, ", "), fmt::join(target_partition, ", ")); + Row target_field_values; + for (const auto position : getTargetFieldPositions(entry, state, spec_field_positions)) + target_field_values.push_back(parsed_entry.partition_key_value[position]); + if (partition_key_value_for_assert.empty()) - partition_key_value_for_assert = parsed_entry.partition_key_value; - else if (!std::ranges::equal(partition_key_value_for_assert, parsed_entry.partition_key_value)) + partition_key_value_for_assert = std::move(target_field_values); + else if (!std::ranges::equal(partition_key_value_for_assert, target_field_values)) { throw Exception( ErrorCodes::LOGICAL_ERROR, "Manifest file entry partition value ({}) is different from previous one ({})", fmt::join(partition_key_value_for_assert, ", "), - fmt::join(parsed_entry.partition_key_value, ", ")); + fmt::join(target_field_values, ", ")); } }; - auto process_entries = [&](const auto & entries, size_t & entries_to_keep, size_t & entries_to_remove) + auto process_entries = [&](const auto & entries, + size_t & entries_to_keep, + size_t & entries_to_remove, + std::unordered_set & excluded_file_paths) { for (const auto & entry : entries) { @@ -438,10 +662,13 @@ AlterDropPartitionExecutor::buildDropPlan(const SnapshotState & state, const Tar continue; } - assert_on_ambigious_partition_key(parsed_entry); + assert_on_ambigious_partition_key(*entry); ++entries_to_remove; unprocessed_target_file_paths.erase(storage_path); + excluded_file_paths.insert(parsed_entry.file_path_key.serialize()); + if (purge) + result.files_to_be_purged.push_back(parsed_entry.file_path_key); switch (parsed_entry.content_type) { case FileContentType::DATA: @@ -455,8 +682,10 @@ AlterDropPartitionExecutor::buildDropPlan(const SnapshotState & state, const Tar removed_files_size += parsed_entry.file_size_in_bytes; break; case FileContentType::EQUALITY_DELETE: - throw Exception( - ErrorCodes::BAD_ARGUMENTS, "DROP PARTITION encountered an equality-delete entry, which is not supported"); + ++removed_equality_delete_files; + removed_equality_deletes += parsed_entry.record_count; + removed_files_size += parsed_entry.file_size_in_bytes; + break; } } }; @@ -468,21 +697,18 @@ AlterDropPartitionExecutor::buildDropPlan(const SnapshotState & state, const Tar size_t entries_to_keep = 0; size_t entries_to_remove = 0; + std::unordered_set excluded_file_paths; - process_entries(handle.getFilesWithoutDeleted(FileContentType::DATA), entries_to_keep, entries_to_remove); - process_entries(handle.getFilesWithoutDeleted(FileContentType::POSITION_DELETE), entries_to_keep, entries_to_remove); + process_entries(handle.getFilesWithoutDeleted(FileContentType::DATA), entries_to_keep, entries_to_remove, excluded_file_paths); + process_entries( + handle.getFilesWithoutDeleted(FileContentType::POSITION_DELETE), entries_to_keep, entries_to_remove, excluded_file_paths); + process_entries( + handle.getFilesWithoutDeleted(FileContentType::EQUALITY_DELETE), entries_to_keep, entries_to_remove, excluded_file_paths); if (entries_to_remove == 0) continue; - /// Removing a manifest with equality deletes would silently resurrect rows. - if (!handle.getFilesWithoutDeleted(FileContentType::EQUALITY_DELETE).empty()) - throw Exception( - ErrorCodes::NOT_IMPLEMENTED, - "DROP PARTITION is not supported on Iceberg tables where an affected manifest '{}' contains equality deletes", - manifest_key.manifest_file_path); - - TargetManifest target_manifest{.manifest_key = manifest_key}; + TargetManifest target_manifest{.manifest_key = manifest_key, .excluded_file_paths = std::move(excluded_file_paths)}; if (entries_to_keep == 0) result.target_manifests.fully_matched.emplace_back(std::move(target_manifest)); else @@ -501,6 +727,8 @@ AlterDropPartitionExecutor::buildDropPlan(const SnapshotState & state, const Tar .removed_files_size = removed_files_size, .removed_position_delete_files = removed_position_delete_files, .removed_position_deletes = removed_position_deletes, + .removed_equality_delete_files = removed_equality_delete_files, + .removed_equality_deletes = removed_equality_deletes, .num_partitions = 1}; return result; } @@ -566,6 +794,8 @@ AlterDropPartitionExecutor::ManifestListWriteResult AlterDropPartitionExecutor:: carry_forward_manifest_paths.insert(path); } + auto manifest_actions = rewritePartiallyMatchedManifests(plan, filename_generator, files_for_cleanup); + LOG_TRACE(log, "ALTER DROP PARTITION writing new manifest list {}", storage_manifest_list_path); auto buf = object_storage->writeObject( @@ -589,30 +819,166 @@ AlterDropPartitionExecutor::ManifestListWriteResult AlterDropPartitionExecutor:: /*use_previous_snapshots=*/false, /*per_entry_content_types=*/{}, /*entry_counts=*/{}, - carry_forward_manifest_paths); + carry_forward_manifest_paths, + /*entry_partition_spec_ids=*/{}, + /*entry_partition_summaries=*/{}, + manifest_actions); buf->finalize(); - return ManifestListWriteResult{.metadata_info = metadata_info}; + return ManifestListWriteResult{.metadata_info = metadata_info, .new_snapshot = new_snapshot}; +} + +PreviousManifestActions AlterDropPartitionExecutor::rewritePartiallyMatchedManifests( + const DropPlan & plan, FileNamesGenerator & filename_generator, std::vector & files_for_cleanup) +{ + PreviousManifestActions manifest_actions; + const auto & resolver = components.path_resolver; + + for (const auto & [source_manifest, excluded_file_paths] : plan.target_manifests.partially_matched) + { + auto rewritten_path = filename_generator.generateManifestEntryName(); + auto rewritten_storage_path = resolver.resolve(rewritten_path); + files_for_cleanup.push_back(rewritten_storage_path); + + auto [source_storage, source_key] = resolveObjectStorageForPath( + components.table_location, + source_manifest.manifest_file_path.serialize(), + object_storage, + external_storages, + context, + resolver); + + RelativePathWithMetadata source_object_info(source_key); + auto source_buf = createReadBuffer(source_object_info, source_storage, context, log); + auto rewritten_buf = object_storage->writeObject( + StoredObject(rewritten_storage_path), + WriteMode::Rewrite, std::nullopt, + DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings()); + + auto rewrite_result = rewriteManifestFileExcludingFiles( + *source_buf, + excluded_file_paths, + source_manifest.added_sequence_number, + source_manifest.added_snapshot_id, + *rewritten_buf); + rewritten_buf->finalize(); + + if (rewrite_result.removed_entries != static_cast(excluded_file_paths.size())) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "Rewriting manifest {} removed {} entries, expected {}", + source_manifest.manifest_file_path.serialize(), rewrite_result.removed_entries, + excluded_file_paths.size()); + + auto rewritten_bytes = rewritten_buf->count(); + if (rewritten_bytes == 0) + { + rewritten_bytes = object_storage->getObjectMetadata(rewritten_storage_path, /*with_tags=*/false).size_bytes; + } + + LOG_TRACE(log, "Rewrote manifest {} as {}: {} entries kept, {} removed", + source_manifest.manifest_file_path.serialize(), rewritten_path.serialize(), + rewrite_result.surviving_entries, rewrite_result.removed_entries); + + manifest_actions.emplace( + source_manifest.manifest_file_path.serialize(), + PreviousManifestAction{ + .new_path = rewritten_path, + .new_length = static_cast(rewritten_bytes), + .new_existing_files_count = rewrite_result.surviving_entries, + .new_existing_rows_count = rewrite_result.surviving_rows, + .new_min_sequence_number = rewrite_result.min_sequence_number, + }); + } + + return manifest_actions; } bool AlterDropPartitionExecutor::commitMetadataJSON( - SnapshotState & state, FileNamesGenerator & filename_generator, const GeneratedMetadataFileWithInfo & metadata_info) + SnapshotState & state, + FileNamesGenerator & filename_generator, + const ManifestListWriteResult & manifest_list_result, + std::vector & files_for_cleanup, + bool & published, + bool * version_hint_names_metadata_file) { + const auto & metadata_info = manifest_list_result.metadata_info; std::string json_representation = stringifyJSON(state.metadata_object, 4); fiu_do_on(FailPoints::iceberg_writes_cleanup, { throw Exception(ErrorCodes::BAD_ARGUMENTS, "Failpoint for cleanup enabled"); }); auto hint_path = filename_generator.generateVersionHint(); - return writeMetadataFileAndVersionHint( - components.path_resolver, - metadata_info, - json_representation, - hint_path, + const bool catalog_writes_metadata_file = catalog && catalog->isTransactional(); + if (!catalog_writes_metadata_file) + { + if (!writeMetadataFileAndVersionHint( + components.path_resolver, + metadata_info, + json_representation, + hint_path, + object_storage, + context, + data_lake_settings[DataLakeStorageSetting::iceberg_use_version_hint], + version_hint_names_metadata_file)) + return false; + } + + if (!catalog) + return true; + + if (!catalog_writes_metadata_file) + files_for_cleanup.push_back(components.path_resolver.resolve(metadata_info.path)); + + const auto & [namespace_name, table_name] = DataLake::parseTableName(storage_id.getTableName()); + const auto new_snapshot_id = manifest_list_result.new_snapshot->getValue(f_metadata_snapshot_id); + + /// Assume readers can reach the snapshot until the catalog says it rejected the commit. + published = true; + if (!catalog->updateMetadata( + namespace_name, table_name, components.path_resolver.resolveForCatalog(metadata_info.path), manifest_list_result.new_snapshot)) + { + if (!catalogCommittedSnapshot(new_snapshot_id)) + { + published = false; + return false; + } + + LOG_INFO(log, "Iceberg DROP PARTITION reported a failed catalog commit that the catalog accepted"); + } + + if (version_hint_names_metadata_file) + *version_hint_names_metadata_file = true; + + return true; +} + +bool AlterDropPartitionExecutor::catalogCommittedSnapshot(Int64 snapshot_id) const +{ + const auto [metadata_version, metadata_file_path, compression_method] = getLatestMetadataFileAndVersionWithCatalog( object_storage, + catalog, + storage_id.getTableName(), + components.table_path, + data_lake_settings, + components.metadata_cache, context, - data_lake_settings[DataLakeStorageSetting::iceberg_use_version_hint]); + log.get(), + components.table_uuid, + components.metadata_compression_method); + + auto metadata_object = getMetadataJSONObject( + metadata_file_path, object_storage, components.metadata_cache, context, log, compression_method, components.table_uuid); + + auto snapshots = metadata_object->getArray(f_snapshots); + for (size_t i = 0; snapshots && i < snapshots->size(); ++i) + { + auto snapshot = snapshots->getObject(static_cast(i)); + if (snapshot && snapshot->getValue(f_metadata_snapshot_id) == snapshot_id) + return true; + } + + return false; } void AlterDropPartitionExecutor::cleanupNotCommited(std::vector files) @@ -636,21 +1002,30 @@ bool AlterDropPartitionExecutor::tryCommit(SnapshotState & state, const DropPlan /// table can move from uncompressed to e.g. `vN.gz.metadata.json` (external writer or changed /// setting), and the replacement metadata must follow the same convention. const auto compression_method = DB::Iceberg::getCompressionMethodFromMetadataFile(state.table_state.metadata_file_path); - FileNamesGenerator filename_generator(components.path_resolver.getTableLocation(), false, compression_method, write_format); + FileNamesGenerator filename_generator( + components.path_resolver.getTableLocation(), + catalog != nullptr && catalog->isTransactional(), + compression_method, + write_format); filename_generator.setVersion(state.table_state.metadata_version + 1); filename_generator.setCompressionMethod(compression_method); std::vector files_for_cleanup; bool committed = false; + /// Set once the new snapshot is reachable by readers, after which its files must survive. + bool published = false; SCOPE_EXIT({ - if (!committed) + if (!committed && !published) cleanupNotCommited(std::move(files_for_cleanup)); }); - auto [metadata_info] = writeManifestList(state, plan, filename_generator, files_for_cleanup); + auto manifest_list_result = writeManifestList(state, plan, filename_generator, files_for_cleanup); + const auto & metadata_info = manifest_list_result.metadata_info; - committed = commitMetadataJSON(state, filename_generator, metadata_info); + bool version_hint_names_metadata_file = false; + committed = commitMetadataJSON( + state, filename_generator, manifest_list_result, files_for_cleanup, published, &version_hint_names_metadata_file); if (!committed) return false; @@ -661,12 +1036,73 @@ bool AlterDropPartitionExecutor::tryCommit(SnapshotState & state, const DropPlan plan.snapshot_summary_update.removed_records, plan.snapshot_summary_update.removed_position_delete_files); + if (purge) + purgeDroppedFiles(plan, metadata_info, version_hint_names_metadata_file); + return true; } +void AlterDropPartitionExecutor::purgeDroppedFiles( + const DropPlan & plan, const GeneratedMetadataFileWithInfo & metadata_info, bool version_hint_names_metadata_file) +{ + if (!version_hint_names_metadata_file) + { + LOG_WARNING(log, "Skipping the purge after DROP PARTITION: version-hint.text does not name {}, so readers " + "that follow it still reference the dropped files", components.path_resolver.resolve(metadata_info.path)); + return; + } + + std::unordered_map batches; + for (const auto & file_to_purge : plan.files_to_be_purged) + { + auto [storage, key] = resolveObjectStorageForPath( + components.table_location, + file_to_purge.serialize(), + object_storage, + external_storages, + context, + components.path_resolver); + + batches[storage].emplace_back(key); + } + + for (auto & [storage, batch] : batches) + { + try + { + storage->removeObjectsIfExist(batch); + } + catch (...) + { + Strings surviving_paths; + surviving_paths.reserve(batch.size()); + for (const auto & object : batch) + surviving_paths.push_back(object.remote_path); + + tryLogCurrentException(log, fmt::format( + "Failed to purge data files after DROP PARTITION, these files may still exist in storage {} and " + "are not retried by re-running the query: {}", + storage->getDescription(), fmt::join(surviving_paths, ", "))); + } + } +} + +void AlterDropPartitionExecutor::checkPurgeIsGranted(const DropPlan & plan) const +{ + for (const auto & file_to_purge : plan.files_to_be_purged) + { + if (!isPathWriteGranted(components.table_location, file_to_purge.serialize(), object_storage, context)) + throw Exception( + ErrorCodes::ACCESS_DENIED, + "Not enough privileges to drop the Iceberg partition: `data_lake_delete_data_on_drop` deletes the file '{}', " + "which requires the `WRITE` grant on it", + file_to_purge); + } +} + void AlterDropPartitionExecutor::run() { - if (data_lake_settings[DataLakeStorageSetting::iceberg_metadata_file_path].changed) + if (!catalog && data_lake_settings[DataLakeStorageSetting::iceberg_metadata_file_path].changed) throw Exception(ErrorCodes::BAD_ARGUMENTS, "DROP PARTITION does not support the iceberg_metadata_file_path setting"); const auto & settings = context->getSettingsRef(); @@ -704,10 +1140,8 @@ void AlterDropPartitionExecutor::run() } auto plan = buildDropPlan(state, targets, target_partition); - if (!plan.target_manifests.partially_matched.empty()) - throw Exception( - ErrorCodes::NOT_IMPLEMENTED, - "DROP PARTITION would require rewriting a manifest that also contains files from other partitions"); + if (purge) + checkPurgeIsGranted(plan); if (tryCommit(state, plan)) return; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h index a462b94962a57..563f98ee455d0 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h @@ -6,12 +6,15 @@ #if USE_AVRO #include +#include #include #include +#include #include #include #include #include +#include #include #include #include @@ -35,6 +38,14 @@ struct ExternalStorageCache; namespace Iceberg { +struct PartitionTerm +{ + Int32 column_id; + String transform; +}; + +using PartitionSpecSignature = std::vector; + class AlterDropPartitionExecutor { public: @@ -47,7 +58,9 @@ class AlterDropPartitionExecutor const PersistentTableComponents & components_, const DataLakeStorageSettings & data_lake_settings_, String write_format_, - LoggerPtr log_); + LoggerPtr log_, + std::shared_ptr catalog_, + const StorageID & storage_id_); void run(); @@ -62,11 +75,15 @@ class AlterDropPartitionExecutor Int32 schema_id = 0; std::vector partition_columns; DataTypes partition_types; + PartitionSpecSignature spec_signature; }; + using SpecFieldPositions = std::unordered_map>; + struct TargetManifest { ManifestFileCacheKey manifest_key; + std::unordered_set excluded_file_paths; }; struct TargetManifests @@ -81,12 +98,21 @@ class AlterDropPartitionExecutor { TargetManifests target_manifests; Iceberg::SnapshotSummaryUpdateDelete snapshot_summary_update; + std::vector files_to_be_purged; }; using TargetFilePaths = std::unordered_set; std::optional fetchSnapshotState(); + const std::vector & getTargetFieldPositions( + const ProcessedManifestFileEntry & entry, const SnapshotState & state, SpecFieldPositions & spec_field_positions) const; + bool isInTargetPartition( + const ProcessedManifestFileEntry & entry, + const SnapshotState & state, + const Row & target_partition, + SpecFieldPositions & spec_field_positions) const; + TargetFilePaths discoverTargetFilePaths(const SnapshotState & state, const Row & target_partition) const; DropPlan buildDropPlan(const SnapshotState & state, const TargetFilePaths & targets, const Row & target_partition) const; bool tryCommit(SnapshotState & state, const DropPlan & plan); @@ -94,16 +120,31 @@ class AlterDropPartitionExecutor struct ManifestListWriteResult { GeneratedMetadataFileWithInfo metadata_info; + Poco::JSON::Object::Ptr new_snapshot; }; ManifestListWriteResult writeManifestList( SnapshotState & state, const DropPlan & plan, FileNamesGenerator & filename_generator, std::vector & files_for_cleanup); - bool - commitMetadataJSON(SnapshotState & state, FileNamesGenerator & filename_generator, const GeneratedMetadataFileWithInfo & metadata_info); + PreviousManifestActions rewritePartiallyMatchedManifests( + const DropPlan & plan, FileNamesGenerator & filename_generator, std::vector & files_for_cleanup); + + bool commitMetadataJSON( + SnapshotState & state, + FileNamesGenerator & filename_generator, + const ManifestListWriteResult & manifest_list_result, + std::vector & files_for_cleanup, + bool & published, + bool * version_hint_names_metadata_file); + + bool catalogCommittedSnapshot(Int64 snapshot_id) const; void cleanupNotCommited(std::vector files); + void checkPurgeIsGranted(const DropPlan & plan) const; + + void purgeDroppedFiles(const DropPlan & plan, const GeneratedMetadataFileWithInfo & metadata_info, bool version_hint_names_metadata_file); + const PartitionCommand & command; const IcebergMetadata & metadata; ContextPtr context; @@ -113,6 +154,9 @@ class AlterDropPartitionExecutor const DataLakeStorageSettings & data_lake_settings; String write_format; LoggerPtr log; + std::shared_ptr catalog; + StorageID storage_id; + bool purge = false; }; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 17a2a27a726c2..ce5900b05cce6 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -940,7 +940,7 @@ Pipe IcebergMetadata::alterPartition( const PartitionCommands & commands, ContextPtr context, std::shared_ptr catalog, - StorageID /*storage_id*/) + StorageID storage_id) { if (!context->getSettingsRef()[Setting::allow_insert_into_iceberg].value) { @@ -948,8 +948,6 @@ Pipe IcebergMetadata::alterPartition( ErrorCodes::SUPPORT_IS_DISABLED, "Alter iceberg is experimental. To allow its usage, enable setting allow_insert_into_iceberg"); } - if (catalog) - throw Exception(ErrorCodes::NOT_IMPLEMENTED, "DROP PARTITION is not supported for catalog-backed Iceberg tables"); if (commands.size() != 1) throw Exception( ErrorCodes::BAD_ARGUMENTS, @@ -961,12 +959,16 @@ Pipe IcebergMetadata::alterPartition( if (command.part || command.detach) throw Exception(ErrorCodes::NOT_IMPLEMENTED, "{} is not supported by Iceberg", command.typeToString()); - alterPartitionDropImpl(command, context); + alterPartitionDropImpl(command, context, std::move(catalog), storage_id); persistent_components.invalidateMetadataCache(); return {}; } -void IcebergMetadata::alterPartitionDropImpl(const PartitionCommand & command, ContextPtr context) +void IcebergMetadata::alterPartitionDropImpl( + const PartitionCommand & command, + ContextPtr context, + std::shared_ptr catalog, + const StorageID & storage_id) { Iceberg::AlterDropPartitionExecutor executor( command, @@ -977,7 +979,9 @@ void IcebergMetadata::alterPartitionDropImpl(const PartitionCommand & command, C persistent_components, data_lake_settings, write_format, - log); + log, + std::move(catalog), + storage_id); executor.run(); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h index 391c1c2edb5b6..952bf95e46297 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h @@ -246,6 +246,9 @@ class IcebergMetadata : public IDataLakeMetadata const ContextPtr & context, bool force_fetch_latest_metadata = false) const; + std::pair + getState(const ContextPtr & local_context, const String & metadata_path, Int32 metadata_version) const; + const DB::Iceberg::PersistentTableComponents & getPersistentComponents() const { return persistent_components; @@ -270,8 +273,6 @@ class IcebergMetadata : public IDataLakeMetadata Iceberg::IcebergDataSnapshotPtr createIcebergDataSnapshotFromSnapshotJSON(Poco::JSON::Object::Ptr snapshot_object, Int64 snapshot_id, ContextPtr local_context) const; std::pair getStateImpl(const ContextPtr & local_context, Poco::JSON::Object::Ptr metadata_object) const; - std::pair - getState(const ContextPtr & local_context, const String & metadata_path, Int32 metadata_version) const; Iceberg::IcebergDataSnapshotPtr getRelevantDataSnapshotFromTableStateSnapshot(Iceberg::TableStateSnapshot table_state_snapshot, ContextPtr local_context) const; StorageObjectStorageConfigurationPtr getConfiguration() const; @@ -326,7 +327,11 @@ class IcebergMetadata : public IDataLakeMetadata void backgroundMetadataPrefetcherThread(); - void alterPartitionDropImpl(const PartitionCommand & command, ContextPtr context); + void alterPartitionDropImpl( + const PartitionCommand & command, + ContextPtr context, + std::shared_ptr catalog, + const StorageID & storage_id); }; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index f279cfa20528e..c46a646b68aad 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -1063,7 +1063,8 @@ void generateManifestList( const std::vector & entry_counts, const std::unordered_set & carry_forward_manifest_paths, const std::vector & entry_partition_spec_ids, - const std::vector>> & entry_partition_summaries) + const std::vector>> & entry_partition_summaries, + const PreviousManifestActions & previous_manifest_actions) { chassert( per_entry_content_types.empty() || per_entry_content_types.size() == manifest_entry_names.size(), @@ -1326,6 +1327,20 @@ void generateManifestList( if (!carry_forward_manifest_paths.empty() && !carry_forward_manifest_paths.contains(old_manifest_path.value())) return; + + const PreviousManifestAction * action = nullptr; + if (!previous_manifest_actions.empty()) + { + auto action_it = previous_manifest_actions.find( + old_entry.field(Iceberg::f_manifest_path).value()); + if (action_it != previous_manifest_actions.end()) + action = &action_it->second; + } + + if (action && version == 1) + throw Exception(ErrorCodes::NOT_IMPLEMENTED, + "Replacing a manifest of a format version 1 table is not supported"); + avro::GenericDatum new_datum(schema.root()); avro::GenericRecord & new_entry = new_datum.value(); @@ -1378,6 +1393,20 @@ void generateManifestList( add_field_to_datum(Iceberg::f_sequence_number); add_field_to_datum(Iceberg::f_min_sequence_number); } + + if (action) + { + new_entry.field(Iceberg::f_manifest_path) = action->new_path.serialize(); + new_entry.field(Iceberg::f_manifest_length) = action->new_length; + new_entry.field(Iceberg::f_added_files_count) = static_cast(0); + new_entry.field(Iceberg::f_existing_files_count) = action->new_existing_files_count; + new_entry.field(Iceberg::f_deleted_files_count) = static_cast(0); + new_entry.field(Iceberg::f_added_rows_count) = static_cast(0); + new_entry.field(Iceberg::f_existing_rows_count) = action->new_existing_rows_count; + new_entry.field(Iceberg::f_deleted_rows_count) = static_cast(0); + new_entry.field(Iceberg::f_min_sequence_number) = action->new_min_sequence_number; + } + writer.write(new_datum); }); break; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h index 2f0bd02a4771d..1b1769757a7a6 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h @@ -158,6 +158,17 @@ struct ManifestListEntryCounts std::optional added_sequence_number; }; +struct PreviousManifestAction +{ + Iceberg::IcebergPathFromMetadata new_path; + Int64 new_length = 0; + Int32 new_existing_files_count = 0; + Int64 new_existing_rows_count = 0; + Int64 new_min_sequence_number = 0; +}; + +using PreviousManifestActions = std::unordered_map; + void generateManifestList( const Iceberg::IcebergPathResolver & path_resolver, Poco::JSON::Object::Ptr metadata, @@ -174,7 +185,8 @@ void generateManifestList( const std::vector & entry_counts = {}, const std::unordered_set & carry_forward_manifest_paths = {}, const std::vector & entry_partition_spec_ids = {}, - const std::vector>> & entry_partition_summaries = {}); + const std::vector>> & entry_partition_summaries = {}, + const PreviousManifestActions & previous_manifest_actions = {}); std::string getIcebergExportPartSidecarStoragePath(const String & data_file_storage_path); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 1512d060a03c9..ddb450c9a59a2 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -342,8 +342,11 @@ bool writeMetadataFileAndVersionHint( const IcebergPathFromMetadata & version_hint_path, DB::ObjectStoragePtr object_storage, DB::ContextPtr context, - bool try_write_version_hint) + bool try_write_version_hint, + bool * version_hint_names_metadata_file) { + bool hint_names_metadata_file = false; + auto storage_metadata_path = resolver.resolve(metadata_file_info.path); auto storage_version_hint_path = resolver.resolve(version_hint_path); try @@ -397,6 +400,7 @@ bool writeMetadataFileAndVersionHint( else if (!try_write_version_hint) { /// The file does not exist and this writer was not asked to create it. + hint_names_metadata_file = true; break; } @@ -424,6 +428,7 @@ bool writeMetadataFileAndVersionHint( context, write_if_none_match, /* write-if-match */ etag); + hint_names_metadata_file = true; break; } catch (...) @@ -460,6 +465,9 @@ bool writeMetadataFileAndVersionHint( metadata_file_info.version, MAX_TRANSACTION_RETRIES); + if (version_hint_names_metadata_file) + *version_hint_names_metadata_file = hint_names_metadata_file; + return true; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h index 434ff0e8618e4..5351d4ca36bf1 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h @@ -66,7 +66,8 @@ bool writeMetadataFileAndVersionHint( const IcebergPathFromMetadata & version_hint_path, DB::ObjectStoragePtr object_storage, DB::ContextPtr context, - bool try_write_version_hint); + bool try_write_version_hint, + bool * version_hint_names_metadata_file = nullptr); struct TransformAndArgument { diff --git a/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp b/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp index 5073a40917e49..596c520fb6b51 100644 --- a/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/registerStorageObjectStorage.cpp @@ -1281,11 +1281,11 @@ ClickHouse supports partition pruning during SELECT queries for Iceberg tables, ## `DROP PARTITION` {#drop-partition} -`ALTER TABLE ... DROP PARTITION ` removes every data file belonging to a single partition and creates a new snapshot that no longer references them. It is currently supported for local and object-storage Iceberg tables, but not for catalog-backed tables. +`ALTER TABLE ... DROP PARTITION ` removes every data file belonging to a single partition and creates a new snapshot that no longer references them. It is supported for local, object-storage and catalog-backed Iceberg tables. Enable `allow_insert_into_iceberg` to use this operation. -The operation is supported only for Iceberg `format-version` 2 tables with a single, non-evolved partition spec. Each manifest containing the selected partition must contain no files from other partitions. If a manifest is shared by the selected partition and another partition, the operation fails without changing the table. The operation also rejects affected manifests containing equality-delete files. +The operation is supported only for Iceberg `format-version` 2 tables. The partition value refers to the table's current partition spec. Files written under an earlier partition spec are dropped when their partition values match every field of the current spec; if such a file's spec does not contain every field of the current spec, the operation fails without changing the table. Manifests that also contain files from other partitions are rewritten without the dropped files. Position-delete and equality-delete files belonging to the partition are dropped together with its data files. The partition value follows the same rules as for `MergeTree`. For a single-column partition, pass a scalar literal; for a multi-column partition, pass a tuple of values: @@ -1301,9 +1301,9 @@ ALTER TABLE iceberg_table DROP PARTITION 0; ALTER TABLE iceberg_table DROP PARTITION tuple(icebergBucket(4, 'apple')); ``` -The operation rejects explicitly set `iceberg_snapshot_id`, `iceberg_timestamp_ms`, or `iceberg_metadata_file_path` settings. It modifies the current table state, not a historical snapshot or an explicitly selected metadata version. +The operation rejects explicitly set `iceberg_snapshot_id` or `iceberg_timestamp_ms` settings, and the `iceberg_metadata_file_path` setting on tables that are not catalog-backed. It modifies the current table state, not a historical snapshot or an explicitly selected metadata version. -The `DROP PARTITION ID '...'` and `DROP PARTITION ALL` forms are not supported. Dropping a partition that does not exist is a no-op. The operation does not physically delete the data files. Earlier snapshots retain access to the removed rows and remain available to time-travel queries until those snapshots expire and their files are cleaned up. +The `DROP PARTITION ID '...'` and `DROP PARTITION ALL` forms are not supported. Dropping a partition that does not exist is a no-op. By default, the operation does not physically delete the data files. Earlier snapshots retain access to the removed rows and remain available to time-travel queries until those snapshots expire and their files are cleaned up. Enable `data_lake_delete_data_on_drop` to delete the dropped files from storage after the new snapshot is committed; this breaks time travel to earlier snapshots that reference them. ## Time travel {#time-travel} diff --git a/tests/integration/test_storage_iceberg_no_spark/test_drop_partition.py b/tests/integration/test_storage_iceberg_no_spark/test_drop_partition.py new file mode 100644 index 0000000000000..18b608dcd4caf --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_drop_partition.py @@ -0,0 +1,779 @@ +#!/usr/bin/env python3 + +import json +import uuid +from datetime import datetime + +import pyarrow as pa +from pyiceberg.catalog import load_catalog +from pyiceberg.partitioning import PartitionField, PartitionSpec +from pyiceberg.schema import Schema +from pyiceberg.transforms import DayTransform, IdentityTransform +from pyiceberg.types import ( + DoubleType, + LongType, + NestedField, + StringType, + TimestampType, +) + +from helpers.config_cluster import minio_access_key, minio_secret_key +from helpers.iceberg_utils import create_iceberg_table, get_uuid_str +from helpers.s3_tools import list_s3_objects + +CATALOG_NAME = "demo" + +WRITE_SETTINGS = { + "allow_experimental_insert_into_iceberg": 1, + "write_full_path_in_iceberg_metadata": 1 +} + +ROWS = [ + {"ts": datetime(2024, 1, 15, 8, 0, 0), "country": "US", "id": 1, "amount": 10.5}, + {"ts": datetime(2024, 1, 15, 12, 30, 0), "country": "DE", "id": 2, "amount": 20.0}, + {"ts": datetime(2024, 1, 15, 23, 0, 0), "country": "US", "id": 3, "amount": 30.25}, + {"ts": datetime(2024, 1, 16, 9, 0, 0), "country": "US", "id": 4, "amount": 60.0}, + {"ts": datetime(2024, 1, 16, 18, 45, 0), "country": "DE", "id": 5, "amount": 70.75}, + {"ts": datetime(2024, 1, 17, 11, 11, 11), "country": "JP", "id": 6, "amount": 80.0}, +] + + +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 create_day_partitioned_table(catalog, namespace, table_name): + """A table partitioned by `day(ts)` filled by a single append that spans three days""" + catalog.create_namespace(namespace) + + schema = Schema( + NestedField(field_id=1, name="ts", field_type=TimestampType(), required=False), + NestedField(field_id=2, name="country", field_type=StringType(), required=False), + NestedField(field_id=3, name="id", field_type=LongType(), required=False), + NestedField(field_id=4, name="amount", field_type=DoubleType(), required=False), + ) + partition_spec = PartitionSpec( + PartitionField( + source_id=1, field_id=1000, transform=DayTransform(), name="ts_day" + ) + ) + + table = catalog.create_table( + identifier=f"{namespace}.{table_name}", + schema=schema, + location=f"s3://warehouse-rest/{namespace}.{table_name}", + partition_spec=partition_spec, + ) + table.append(pa.Table.from_pylist(ROWS, schema=schema.as_arrow())) + + manifests = table.current_snapshot().manifests(table.io) + assert len(manifests) == 1, ( + f"this test needs one manifest holding several partitions, pyiceberg wrote {len(manifests)}" + ) + return table + + +def create_iceberg_database(instance, namespace): + instance.query(f"DROP DATABASE IF EXISTS {namespace}") + instance.query( + f""" + CREATE DATABASE {namespace} ENGINE = DataLakeCatalog('http://rest:8181/v1', '{minio_access_key}', '{minio_secret_key}') + SETTINGS + catalog_type='rest', + warehouse='demo', + storage_endpoint='http://minio1:9001/warehouse-rest'; + """, + settings={"allow_database_iceberg": 1}, + ) + + +def current_snapshot_history(instance, namespace, table_name): + """`operation` and `summary` of the table's current snapshot""" + operation, summary = ( + instance.query( + f"SELECT operation, toJSONString(summary) FROM system.iceberg_history " + f"WHERE database = '{namespace}' AND table = '{namespace}.{table_name}' " + f"ORDER BY made_current_at DESC, snapshot_id DESC LIMIT 1 FORMAT TSV" + ) + .strip() + .split("\t", 1) + ) + return operation, json.loads(summary) + + +def create_partitioned_table(instance, namespace, table_name, columns, partition_by): + """Create an Iceberg table through ClickHouse and return its qualified name.""" + ch_table = f"{namespace}.`{namespace}.{table_name}`" + instance.query( + f"CREATE TABLE {ch_table} ({columns}) " + f"ENGINE = IcebergS3('http://minio1:9001/warehouse-rest/{namespace}/{table_name}', " + f"'{minio_access_key}', '{minio_secret_key}') " + f"PARTITION BY {partition_by}", + settings=WRITE_SETTINGS, + ) + return ch_table + + +def setup(instance, table_name, columns, partition_by): + namespace = f"clickhouse_drop_partition_{uuid.uuid4().hex}" + create_iceberg_database(instance, namespace) + ch_table = create_partitioned_table( + instance, namespace, table_name, columns, partition_by + ) + return namespace, ch_table + + +def test_drop_partition_separate_manifest(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + + namespace = f"clickhouse_drop_partition_{uuid.uuid4().hex}" + table_name = "separate_manifest" + + create_iceberg_database(instance, namespace) + + ch_table = f"{namespace}.`{namespace}.{table_name}`" + + instance.query( + f"CREATE TABLE {ch_table} (" + f"ts DateTime64(6)," + f"country String," + f"id Int64," + f"amount Float64) " + f"ENGINE = IcebergS3('http://minio1:9001/warehouse-rest/{namespace}/{table_name}', '{minio_access_key}', '{minio_secret_key}') " + f"PARTITION BY toRelativeDayNum(ts)", + settings = WRITE_SETTINGS + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"(toDateTime64('2024-01-15 08:00:00', 6), 'US', 1, 10.5), " + f"(toDateTime64('2024-01-15 12:30:00', 6), 'DE', 2, 20.0), " + f"(toDateTime64('2024-01-15 23:59:59', 6), 'US', 3, 30.25)", + settings = WRITE_SETTINGS + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"(toDateTime64('2024-01-15 01:00:00', 6), 'FR', 4, 40.0), " + f"(toDateTime64('2024-01-15 06:15:00', 6), 'US', 5, 50.5)", + settings = WRITE_SETTINGS + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"(toDateTime64('2024-01-16 09:00:00', 6), 'US', 6, 60.0), " + f"(toDateTime64('2024-01-16 18:45:00', 6), 'DE', 7, 70.75)", + settings = WRITE_SETTINGS + ) + + instance.query(f"INSERT INTO {ch_table} VALUES (toDateTime64('2024-01-17 11:11:11', 6), 'JP', 8, 80.0)", settings = WRITE_SETTINGS) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "8" + + assert instance.query( + f"SELECT toDate(ts) AS d, count() FROM {ch_table} GROUP BY d ORDER BY d FORMAT TSV" + ) == "2024-01-15\t5\n2024-01-16\t2\n2024-01-17\t1\n" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-15', 6)))", + settings = WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "3" + + assert instance.query( + f"SELECT toDate(ts) AS d, count() FROM {ch_table} GROUP BY d ORDER BY d FORMAT TSV" + ) == "2024-01-16\t2\n2024-01-17\t1\n" + + # Drop the rest of the partitions + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-16', 6)))", + settings = WRITE_SETTINGS, + ) + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-17', 6)))", + settings = WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "0" + + + +def test_drop_partition_rewrites_mixed_manifest(started_cluster_iceberg_no_spark): + """DROP PARTITION on a manifest that also holds files of other partitions must rewrite + that manifest without the dropped files, rather than unreferencing it completely.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + catalog = load_catalog_impl(started_cluster_iceberg_no_spark) + + namespace = f"clickhouse_drop_partition_{uuid.uuid4().hex}" + table_name = "mixed_manifest" + create_day_partitioned_table(catalog, namespace, table_name) + + create_iceberg_database(instance, namespace) + ch_table = f"{namespace}.`{namespace}.{table_name}`" + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "6" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-15', 6)))", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "3" + assert instance.query( + f"SELECT id, country FROM {ch_table} ORDER BY id FORMAT TSV" + ) == "4\tUS\n5\tDE\n6\tJP\n" + assert instance.query( + f"SELECT toDate(ts) AS d, count() FROM {ch_table} GROUP BY d ORDER BY d FORMAT TSV" + ) == "2024-01-16\t2\n2024-01-17\t1\n" + + # Only the file of the dropped partition is gone while the two that shared its manifest + # stay + assert instance.query( + f"SELECT partition, record_count, sequence_number FROM system.iceberg_files " + f"WHERE database = '{namespace}' AND table = '{namespace}.{table_name}' " + f"AND content = 'DATA' ORDER BY partition FORMAT TSV" + ) == "{19738}\t2\t1\n{19739}\t1\t1\n" + + operation, summary = current_snapshot_history(instance, namespace, table_name) + assert operation == "DELETE" + assert summary["deleted-data-files"] == "1" + assert summary["deleted-records"] == "3" + assert summary["total-data-files"] == "2" + assert summary["total-records"] == "3" + assert summary["changed-partition-count"] == "1" + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_without_matching_files_is_noop(started_cluster_iceberg_no_spark): + """A partition value that no file belongs to must leave the table untouched, with no + new snapshot committed.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + catalog = load_catalog_impl(started_cluster_iceberg_no_spark) + + namespace = f"clickhouse_drop_partition_noop_{uuid.uuid4().hex}" + table_name = "mixed_manifest" + create_day_partitioned_table(catalog, namespace, table_name) + + create_iceberg_database(instance, namespace) + ch_table = f"{namespace}.`{namespace}.{table_name}`" + + snapshots_before = instance.query( + f"SELECT count() FROM system.iceberg_history " + f"WHERE database = '{namespace}' AND table = '{namespace}.{table_name}'" + ).strip() + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-02-20', 6)))", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "6" + assert ( + instance.query( + f"SELECT count() FROM system.iceberg_history " + f"WHERE database = '{namespace}' AND table = '{namespace}.{table_name}'" + ).strip() + == snapshots_before + ) + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_tuple_of_identities(started_cluster_iceberg_no_spark): + """A partition spec of several identity fields: the values are given as a tuple, one per + partition field.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + namespace, ch_table = setup( + instance, + "tuple_of_identities", + "country String, region String, id Int64", + "(country, region)", + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"('US', 'west', 1), ('US', 'east', 2), ('DE', 'west', 3)", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "3" + + error = instance.query_and_get_error( + f"ALTER TABLE {ch_table} DROP PARTITION 'US'", settings=WRITE_SETTINGS + ) + assert "Wrong number of fields in the partition expression: 1, must be: 2" in error + + error = instance.query_and_get_error( + f"ALTER TABLE {ch_table} DROP PARTITION ('US', 'west', 'extra')", + settings=WRITE_SETTINGS, + ) + assert "Wrong number of fields in the partition expression: 3, must be: 2" in error + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION ('US', 'west')", settings=WRITE_SETTINGS + ) + + assert ( + instance.query( + f"SELECT country, region, id FROM {ch_table} ORDER BY id FORMAT TSV" + ) + == "US\teast\t2\nDE\twest\t3\n" + ) + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_month_transform(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + namespace, ch_table = setup( + instance, + "month_transform", + "ts DateTime64(6), id Int64", + "toMonthNumSinceEpoch(ts)", + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"(toDateTime64('2024-01-02 00:00:00', 6), 1), " + f"(toDateTime64('2024-01-30 23:59:59', 6), 2), " + f"(toDateTime64('2024-02-10 12:00:00', 6), 3), " + f"(toDateTime64('2024-02-28 23:59:59', 6), 4)", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "4" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toMonthNumSinceEpoch(toDateTime64('2024-01-15 08:30:00', 6)))", + settings=WRITE_SETTINGS, + ) + + assert ( + instance.query(f"SELECT id FROM {ch_table} ORDER BY id FORMAT TSV") == "3\n4\n" + ) + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toMonthNumSinceEpoch(toDateTime64('2024-02-01', 6)))", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "0" + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_day_transform(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + namespace, ch_table = setup( + instance, + "day_transform", + "ts DateTime64(6), id Int64", + "toRelativeDayNum(ts)", + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"(toDateTime64('2024-01-15 08:00:00', 6), 1), " + f"(toDateTime64('2024-01-15 23:59:59', 6), 2), " + f"(toDateTime64('2024-01-16 09:00:00', 6), 3), " + f"(toDateTime64('2024-01-17 09:00:00', 6), 4)", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "4" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-15 12:00:00', 6)))", + settings=WRITE_SETTINGS, + ) + assert instance.query(f"SELECT id FROM {ch_table} ORDER BY id FORMAT TSV") == "3\n4\n" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-16', 6)))", + settings=WRITE_SETTINGS, + ) + assert instance.query(f"SELECT id FROM {ch_table} FORMAT TSV") == "4\n" + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_int_identity(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + namespace, ch_table = setup( + instance, "int_identity", "key Int64, val String", "key" + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES (10, 'a'), (10, 'b'), (20, 'c'), (30, 'd')", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "4" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION 10", settings=WRITE_SETTINGS + ) + assert ( + instance.query(f"SELECT key, val FROM {ch_table} ORDER BY key FORMAT TSV") + == "20\tc\n30\td\n" + ) + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION '20'", settings=WRITE_SETTINGS + ) + assert instance.query(f"SELECT key FROM {ch_table} FORMAT TSV") == "30\n" + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_date_identity(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + namespace, ch_table = setup( + instance, "date_identity", "event_date Date, id Int64", "event_date" + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"('2024-01-15', 1), ('2024-01-15', 2), ('2024-01-16', 3)", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "3" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION '2024-01-15'", settings=WRITE_SETTINGS + ) + + assert ( + instance.query( + f"SELECT event_date, id FROM {ch_table} ORDER BY id FORMAT TSV" + ) + == "2024-01-16\t3\n" + ) + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_unpartitioned_table(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + + namespace = f"clickhouse_drop_partition_{uuid.uuid4().hex}" + table_name = "unpartitioned" + ch_table = f"{namespace}.`{namespace}.{table_name}`" + + create_iceberg_database(instance, namespace) + + instance.query( + f"CREATE TABLE {ch_table} (id Int64, country String) " + f"ENGINE = IcebergS3('http://minio1:9001/warehouse-rest/{namespace}/{table_name}', " + f"'{minio_access_key}', '{minio_secret_key}')", + settings=WRITE_SETTINGS, + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"(1, 'CA'), (2, 'PT'), (3, 'MX')", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "3" + + error = instance.query_and_get_error( + f"ALTER TABLE {ch_table} DROP PARTITION 'CA'", settings=WRITE_SETTINGS + ) + assert "DROP PARTITION is not supported on unpartitioned Iceberg tables" in error + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "3" + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_without_catalog(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + + table_name = "test_drop_partition_no_catalog_" + get_uuid_str() + + create_iceberg_table( + "s3", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(ts DateTime64(6), id Int64)", + format_version=2, + partition_by="toRelativeDayNum(ts)", + use_version_hint=True + ) + + instance.query( + f"INSERT INTO {table_name} VALUES " + f"(toDateTime64('2024-01-15 08:00:00', 6), 1), " + f"(toDateTime64('2024-01-15 23:59:59', 6), 2), " + f"(toDateTime64('2024-01-16 09:00:00', 6), 3), " + f"(toDateTime64('2024-01-17 09:00:00', 6), 4)", + settings=WRITE_SETTINGS + ) + + assert instance.query(f"SELECT count() FROM {table_name}").strip() == "4" + + instance.query( + f"ALTER TABLE {table_name} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-15', 6)))", + settings=WRITE_SETTINGS + ) + + assert instance.query(f"SELECT count() FROM {table_name}").strip() == "2" + + +def test_drop_partition_with_position_deletes(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + + table_name = "position_deletes" + namespace, ch_table = setup( + instance, table_name, "event_date Date, id Int64", "event_date" + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"('2024-01-15', 1), ('2024-01-15', 2), ('2024-01-16', 3), ('2024-01-17', 4)", + settings=WRITE_SETTINGS + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "4" + + instance.query( + f"DELETE FROM {ch_table} WHERE event_date = '2024-01-16'", + settings=WRITE_SETTINGS + ) + + instance.query( + f"DELETE FROM {ch_table} WHERE id = 1", + settings=WRITE_SETTINGS + ) + + assert instance.query( + f"SELECT count() FROM system.iceberg_files " + f"WHERE database = '{namespace}' AND table = '{namespace}.{table_name}' " + f"AND content = 'POSITION_DELETE'" + ).strip() == "2" + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "2" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION '2024-01-15'", + settings=WRITE_SETTINGS + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "1" + + assert instance.query( + f"SELECT content, count() FROM system.iceberg_files " + f"WHERE database = '{namespace}' AND table = '{namespace}.{table_name}' " + f"GROUP BY content ORDER BY content FORMAT TSV" + ) == "DATA\t2\nPOSITION_DELETE\t1\n" + + operation, summary = current_snapshot_history(instance, namespace, table_name) + assert operation == "DELETE" + assert summary["deleted-data-files"] == "1" + assert summary["removed-position-delete-files"] == "1" + assert summary["removed-position-deletes"] == "1" + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_hour_transform(started_cluster_iceberg_no_spark): + instance = started_cluster_iceberg_no_spark.instances["node1"] + + namespace, ch_table = setup( + instance, "hour_transform", "ts DateTime64(6), id Int64", "toRelativeHourNum(ts)" + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"(toDateTime64('2024-01-15 08:15:00', 6, 'UTC'), 1), " + f"(toDateTime64('2024-01-15 08:45:00', 6, 'UTC'), 2), " + f"(toDateTime64('2024-01-15 09:30:00', 6, 'UTC'), 3), " + f"(toDateTime64('2024-01-15 10:00:00', 6, 'UTC'), 4)", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "4" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeHourNum(toDateTime64('2024-01-15 09:30:00', 6, 'UTC')))", + settings=WRITE_SETTINGS, + ) + + assert ( + instance.query(f"SELECT id FROM {ch_table} ORDER BY id FORMAT TSV") + == "1\n2\n4\n" + ) + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeHourNum(toDateTime64('2024-01-15 08:59:59', 6, 'UTC')))", + settings=WRITE_SETTINGS, + ) + + assert instance.query(f"SELECT id FROM {ch_table} ORDER BY id FORMAT TSV") == "4\n" + + instance.query(f"DROP DATABASE {namespace}") + + +def test_drop_partition_after_partition_spec_evolution(started_cluster_iceberg_no_spark): + """Files written under `day(ts), country` are partitioned more finely than the current + `day(ts)` spec, so a drop by day must take them along with the files of the current spec.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + catalog = load_catalog_impl(started_cluster_iceberg_no_spark) + + namespace = f"clickhouse_drop_partition_{uuid.uuid4().hex}" + table_name = "spec_evolution" + catalog.create_namespace(namespace) + + schema = Schema( + NestedField(field_id=1, name="ts", field_type=TimestampType(), required=False), + NestedField(field_id=2, name="country", field_type=StringType(), required=False), + NestedField(field_id=3, name="id", field_type=LongType(), required=False), + ) + table = catalog.create_table( + identifier=f"{namespace}.{table_name}", + schema=schema, + location=f"s3://warehouse-rest/{namespace}.{table_name}", + partition_spec=PartitionSpec( + PartitionField( + source_id=1, field_id=1000, transform=DayTransform(), name="ts_day" + ), + PartitionField( + source_id=2, field_id=1001, transform=IdentityTransform(), name="country" + ), + ), + ) + table.append( + pa.Table.from_pylist( + [ + {"ts": datetime(2024, 1, 15, 8, 0, 0), "country": "US", "id": 1}, + {"ts": datetime(2024, 1, 15, 12, 0, 0), "country": "DE", "id": 2}, + {"ts": datetime(2024, 1, 16, 9, 0, 0), "country": "US", "id": 3}, + ], + schema=schema.as_arrow(), + ) + ) + + with table.update_spec() as update: + update.remove_field("country") + + table.append( + pa.Table.from_pylist( + [ + {"ts": datetime(2024, 1, 15, 20, 0, 0), "country": "JP", "id": 4}, + {"ts": datetime(2024, 1, 17, 10, 0, 0), "country": "US", "id": 5}, + ], + schema=schema.as_arrow(), + ) + ) + + create_iceberg_database(instance, namespace) + ch_table = f"{namespace}.`{namespace}.{table_name}`" + + assert instance.query(f"SELECT count() FROM {ch_table}").strip() == "5" + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION " + f"tuple(toRelativeDayNum(toDateTime64('2024-01-15', 6)))", + settings=WRITE_SETTINGS, + ) + + assert ( + instance.query(f"SELECT id FROM {ch_table} ORDER BY id FORMAT TSV") == "3\n5\n" + ) + + operation, summary = current_snapshot_history(instance, namespace, table_name) + assert operation == "DELETE" + assert summary["deleted-data-files"] == "3" + assert summary["deleted-records"] == "3" + + instance.query(f"DROP DATABASE {namespace}") + + +def data_files_in_storage(started_cluster, namespace, table_name): + return sorted( + name + for name in list_s3_objects( + started_cluster.minio_client, + "warehouse-rest", + prefix=f"{namespace}/{table_name}/", + ) + if name.endswith(".parquet") + ) + + +def test_drop_partition_purges_data_files(started_cluster_iceberg_no_spark): + """`data_lake_delete_data_on_drop` decides whether the dropped files leave object storage. + """ + instance = started_cluster_iceberg_no_spark.instances["node1"] + + table_name = "purge_data_files" + namespace, ch_table = setup( + instance, table_name, "event_date Date, id Int64", "event_date" + ) + + instance.query( + f"INSERT INTO {ch_table} VALUES " + f"('2024-01-15', 1), ('2024-01-15', 2), ('2024-01-16', 3), ('2024-01-17', 4)", + settings=WRITE_SETTINGS, + ) + + after_insert = data_files_in_storage( + started_cluster_iceberg_no_spark, namespace, table_name + ) + assert len(after_insert) == 3 + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION '2024-01-16'", + settings=dict(WRITE_SETTINGS, data_lake_delete_data_on_drop=0), + ) + + assert instance.query(f"SELECT id FROM {ch_table} ORDER BY id FORMAT TSV") == "1\n2\n4\n" + assert ( + data_files_in_storage(started_cluster_iceberg_no_spark, namespace, table_name) + == after_insert + ) + + instance.query( + f"ALTER TABLE {ch_table} DROP PARTITION '2024-01-15'", + settings=dict(WRITE_SETTINGS, data_lake_delete_data_on_drop=1), + ) + + assert instance.query(f"SELECT id FROM {ch_table} ORDER BY id FORMAT TSV") == "4\n" + + after_purge = data_files_in_storage( + started_cluster_iceberg_no_spark, namespace, table_name + ) + assert len(after_purge) == 2 + assert set(after_purge) < set(after_insert) + + operation, summary = current_snapshot_history(instance, namespace, table_name) + assert operation == "DELETE" + assert summary["deleted-data-files"] == "1" + + instance.query(f"DROP DATABASE {namespace}") diff --git a/tests/integration/test_storage_iceberg_with_spark/test_drop_partition_whole_manifest.py b/tests/integration/test_storage_iceberg_with_spark/test_drop_partition_whole_manifest.py index a67f7479ab461..9c4bdc3608e2c 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_drop_partition_whole_manifest.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_drop_partition_whole_manifest.py @@ -146,7 +146,7 @@ def spark_query(query): @pytest.mark.parametrize("storage_type", ["s3", "local"]) -def test_drop_partition_shared_spark_manifest_is_rejected( +def test_drop_partition_shared_spark_manifest_is_rewritten( started_cluster_iceberg_with_spark, storage_type ): cluster = started_cluster_iceberg_with_spark @@ -171,19 +171,24 @@ def spark_query(query): ) create_iceberg_table(storage_type, instance, table_name, cluster) - error = instance.query_and_get_error( + instance.query( f"ALTER TABLE {table_name} DROP PARTITION 1", settings={"allow_insert_into_iceberg": 1}, ) - assert "NOT_IMPLEMENTED" in error, error - assert "rewriting a manifest" in error, error - expected = "1\tdrop\n2\tkeep-a\n3\tkeep-b\n" + expected = "2\tkeep-a\n3\tkeep-b\n" assert instance.query(f"SELECT * FROM {table_name} ORDER BY tag") == expected + local_dir = update_spark_version_hint(cluster, storage_type, table_name) + spark_rows = spark.read.format("iceberg").load(local_dir).collect() + assert sorted((row["tag"], row["value"]) for row in spark_rows) == [ + (2, "keep-a"), + (3, "keep-b"), + ] + @pytest.mark.parametrize("storage_type", ["s3", "local"]) -def test_drop_partition_with_spark_evolved_spec_is_rejected( +def test_drop_partition_with_spark_evolved_spec( started_cluster_iceberg_with_spark, storage_type ): cluster = started_cluster_iceberg_with_spark @@ -208,10 +213,26 @@ def spark_query(query): create_iceberg_table(storage_type, instance, table_name, cluster) + # Files written before the spec gained `key` span every `key` value of their `tag` error = instance.query_and_get_error( - f"ALTER TABLE {table_name} DROP PARTITION 1", + f"ALTER TABLE {table_name} DROP PARTITION (1, 'c')", settings={"allow_insert_into_iceberg": 1}, ) assert "NOT_IMPLEMENTED" in error, error - assert "evolved partition specs" in error, error + assert "does not contain every field of the current spec" in error, error assert instance.query(f"SELECT count() FROM {table_name}") == "4\n" + + instance.query( + f"ALTER TABLE {table_name} DROP PARTITION (3, 'd')", + settings={"allow_insert_into_iceberg": 1}, + ) + expected = "1\ta\t10\n1\tc\t30\n2\tb\t20\n" + assert instance.query(f"SELECT * FROM {table_name} ORDER BY tag, key") == expected + + local_dir = update_spark_version_hint(cluster, storage_type, table_name) + spark_rows = spark.read.format("iceberg").load(local_dir).collect() + assert sorted((row["tag"], row["key"], row["value"]) for row in spark_rows) == [ + (1, "a", 10), + (1, "c", 30), + (2, "b", 20), + ]