Export partition - allow non matching partition expressions iff monotonicity is proved and data is not repartitioned

docs/en/antalya/partition_export.md

@@ -24,6 +24,19 @@ The manifest file produced by the commit contains a summary field `clickhouse.ex
 
 The Iceberg manifest files contain statistics about the data. Exporting a merge tree partition is a non ephemeral long running task, in which nodes can be turned off and turned on. This means the stats of individual files need to be persisted somewhere in order to produce the final manifest. This is implemented through sidecars. Each data file exported will contain a "sibling" sidecar file named `<data_file_name>_clickhouse_export_part_sidecar.avro`. ClickHouse does not clean up these files, and they can be safely deleted once the data is comitted.
 
+#### Source partition key compatibility
+
+Because the commit writes a single partition tuple per exported partition, every destination Iceberg partition field must be single-valued across the exported source partition. A source `PARTITION BY` field is accepted when either:
+
+- It structurally matches the destination transform on the same column. The functions with a direct Iceberg equivalent are `identity` (a bare column), `toYearNumSinceEpoch` (`year`), `toMonthNumSinceEpoch` (`month`), `toRelativeDayNum` (`day`), `toRelativeHourNum` (`hour`), `icebergTruncate` (`truncate`), and `icebergBucket` (`bucket`).
+- Or the destination transform is proven constant over the exported partition's actual `[min, max]`. This accepts other equivalent or finer keys - for example `PARTITION BY toDate(ts)` or `toYYYYMM(ts)` or `toStartOfHour(ts)` into a destination partitioned by `day(ts)` / `month(ts)` / `hour(ts)`, a bare `Date` column into a `day` transform, or a source that adds extra partition columns on top of the destination's.
+
+The proof uses each part's min/max statistics, so it is data-dependent: a partition whose rows would span more than one destination partition (for example a monthly source partition exported into a daily destination that actually contains several days) is rejected with a `BAD_ARGUMENTS` error at `EXPORT` time.
+
+`bucket` is a hash and is not order-preserving, so it can only be matched structurally: the source must be partitioned by `icebergBucket(N, col)` with the same `N`.
+
+Lossy partition-column casts (allowed via `export_merge_tree_part_allow_lossy_cast`) are supported as long as the cast stays order-preserving over the partition's actual values; a partition whose values cross the destination type's overflow boundary is rejected. A `Nullable` partition column is only accepted through a structural match, because a `NULL` forms its own Iceberg partition.
+
 ### On plain object storage exports:
 
 Each MergeTree part will become a separate file with the following name convention: `<table_directory>/<partitioning>/<data_part_name>_<merge_tree_part_checksum>.<format>`. To ensure atomicity, a commit file containing the relative paths of all exported parts is also shipped. A data file should only be considered part of the dataset if a commit file references it. The commit file will be named using the following convention: `<table_directory>/commit_<partition_id>_<transaction_id>`.

src/Storages/MergeTree/ExportPartitionUtils.h

@@ -8,6 +8,7 @@
 #include <Common/ZooKeeper/ZooKeeper.h>
 #include "Storages/IStorage.h"
 #include <Storages/StorageInMemoryMetadata.h>
+#include <Storages/MergeTree/MergeTreeData.h>
 #include <config.h>
 
 #if USE_AVRO
@@ -100,12 +101,20 @@ namespace ExportPartitionUtils
         const ContextPtr & context);
 
 #if USE_AVRO
-    /// Verifies the source MergeTree partition key matches the destination Iceberg
-    /// partition spec (source-ids and transforms in order). Throws BAD_ARGUMENTS on
-    /// mismatch.
+    /// Verifies the source MergeTree partition key is compatible with the destination Iceberg
+    /// partition spec: every destination partition field must be single-valued across the exported
+    /// source partition (which the commit path requires - it writes one partition tuple per export).
+    /// A field is proven either structurally (the source key already applies the matching transform
+    /// on that column) or dynamically, by checking the destination transform is constant over the
+    /// partition's actual [min, max] folded across `parts`. `bucket` is non-monotonic and can only be
+    /// matched structurally. Throws BAD_ARGUMENTS when a field cannot be proven.
     void verifyIcebergPartitionCompatibility(
         const Poco::JSON::Object::Ptr & metadata_object,
-        const ASTPtr & partition_key_ast);
+        const StorageMetadataPtr & source_metadata,
+        const StorageMetadataPtr & destination_metadata,
+        const MergeTreeData::DataPartsVector & parts,
+        const String & partition_id,
+        const ContextPtr & context);
 #endif
 }
 

src/Storages/MergeTree/MergeTreeData.cpp

@@ -6700,6 +6700,12 @@ void MergeTreeData::exportPartToTable(
     auto source_metadata_ptr = getInMemoryMetadataPtr();
     auto destination_metadata_ptr = dest_storage->getInMemoryMetadataPtr();
 
+    auto part = getPartIfExists(part_name, {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated});
+
+    if (!part)
+        throw Exception(ErrorCodes::NO_SUCH_DATA_PART, "No such data part '{}' to export in table '{}'",
+                        part_name, getStorageID().getFullTableName());
+
     std::string iceberg_metadata_json;
 
     if (dest_storage->isDataLake())
@@ -6745,7 +6751,11 @@ void MergeTreeData::exportPartToTable(
 
             ExportPartitionUtils::verifyIcebergPartitionCompatibility(
                 metadata_object,
-                source_metadata_ptr->getPartitionKeyAST());
+                source_metadata_ptr,
+                destination_metadata_ptr,
+                {part},
+                part->info.getPartitionId(),
+                query_context);
         }
 #else
         (void)iceberg_metadata_json_;
@@ -6765,12 +6775,6 @@ void MergeTreeData::exportPartToTable(
             throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key");
     }
 
-    auto part = getPartIfExists(part_name, {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated});
-
-    if (!part)
-        throw Exception(ErrorCodes::NO_SUCH_DATA_PART, "No such data part '{}' to export in table '{}'",
-                        part_name, getStorageID().getFullTableName());
-
     if (part->getState() == MergeTreeDataPartState::Outdated && !allow_outdated_parts)
         throw Exception(
             ErrorCodes::BAD_ARGUMENTS,

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

@@ -676,10 +676,10 @@ Poco::JSON::Object::Ptr getPartitionField(
                 field = identifier->name();
             }
             const auto * literal = expression_list_child->as<ASTLiteral>();
-            if (literal)
-            {
+            /// A String literal is an optional timezone argument (e.g. toRelativeDayNum(col, 'UTC')) and
+            /// does not affect the transform; only an integer literal is the bucket/truncate width.
+            if (literal && literal->value.getType() != Field::Types::String)
                 param = literal->value.safeGet<Int64>();
-            }
         }
     }
     if (!field)

src/Storages/StorageReplicatedMergeTree.cpp

@@ -8565,7 +8565,11 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand &
 
         ExportPartitionUtils::verifyIcebergPartitionCompatibility(
             metadata_object,
-            src_snapshot->getPartitionKeyAST());
+            src_snapshot,
+            destination_snapshot,
+            parts,
+            partition_id,
+            query_context);
 
         std::ostringstream oss;     // STYLE_CHECK_ALLOW_STD_STRING_STREAM
         oss.exceptions(std::ios::failbit);