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,13 @@ 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
+
+The source partition must not be split in the destination. This is validated at schedule time through two mechanisms:
+
+1. Identical expressions; or
+2. Destination partition expression is monotonically increasing in the min/max column ranges and the destination expression columns are a subset of the source expression;
+
 ### 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/ExportReplicatedMergeTreePartitionManifest.h

@@ -249,6 +249,11 @@ struct ExportReplicatedMergeTreePartitionManifest
     std::optional<UInt64> parquet_row_group_size;
     std::optional<UInt64> parquet_row_group_size_bytes;
 
+    /// this is a controversial setting. As far as I can infer from the iceberg docs, the transforms are always UTC.
+    /// this setting allows to specify different timezones. Since it is already implemented, we must respect it.
+    /// At the same time, we don't allow transforms with timezones, so this is very weird.
+    std::optional<String> iceberg_partition_timezone;
+
     std::string toJsonString() const
     {
         Poco::JSON::Object json;
@@ -290,6 +295,8 @@ struct ExportReplicatedMergeTreePartitionManifest
             json.set("parquet_row_group_size", *parquet_row_group_size);
         if (parquet_row_group_size_bytes)
             json.set("parquet_row_group_size_bytes", *parquet_row_group_size_bytes);
+        if (iceberg_partition_timezone)
+            json.set("iceberg_partition_timezone", *iceberg_partition_timezone);
         std::ostringstream oss;     // STYLE_CHECK_ALLOW_STD_STRING_STREAM
         oss.exceptions(std::ios::failbit);
         Poco::JSON::Stringifier::stringify(json, oss);
@@ -378,6 +385,11 @@ struct ExportReplicatedMergeTreePartitionManifest
             manifest.parquet_row_group_size_bytes = json->getValue<UInt64>("parquet_row_group_size_bytes");
         }
 
+        if (json->has("iceberg_partition_timezone"))
+        {
+            manifest.iceberg_partition_timezone = json->getValue<String>("iceberg_partition_timezone");
+        }
+
         return manifest;
     }
 };

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
@@ -29,16 +30,9 @@ namespace ExportPartitionUtils
 
     ContextPtr getContextCopyWithTaskSettings(const ContextPtr & context, const ExportReplicatedMergeTreePartitionManifest & manifest);
 
-    /// Returns the representative source partition-key columns (the first active local part's
-    /// minmax block) for the given partition_id. The destination recomputes the Iceberg partition
-    /// tuple from this block by casting to its column types and applying the partition transform.
-    ///
-    /// Edge case: if the partition was dropped after export started, or this replica
-    /// has not yet received any part for this partition (extreme replication lag on a
-    /// recovery path), no active part will be found and the commit will fail. The task
-    /// will be retried on the next poll cycle or picked up by a different replica.
+    /// Get the min/max values from the partition expression columns
     Block getPartitionSourceBlockForIcebergCommit(
-        MergeTreeData & storage, const String & partition_id);
+        MergeTreeData & storage, const String & partition_id, const std::vector<String> & exported_part_names);
 
     void commit(
         const ExportReplicatedMergeTreePartitionManifest & manifest,
@@ -89,23 +83,34 @@ namespace ExportPartitionUtils
         const std::string & exception_message,
         const LoggerPtr & log);
 
-    /// Validates that source columns can be exported into the destination with the
-    /// same positional CAST matching as `INSERT INTO dest SELECT * FROM src`. Lossy
-    /// casts are rejected unless `export_merge_tree_part_allow_lossy_cast` is set.
-    /// Throws BAD_ARGUMENTS on any violation.
     void verifyExportSchemaCastable(
         const StorageMetadataPtr & source_metadata,
         const StorageMetadataPtr & destination_metadata,
         const StorageID & destination_storage_id,
         const ContextPtr & context);
 
+    void verifyPlainPartitionCompatibility(
+        const StorageMetadataPtr & source_metadata,
+        const StorageMetadataPtr & destination_metadata,
+        const MergeTreeData::DataPartsVector & parts,
+        const String & partition_id,
+        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

@@ -6692,25 +6692,29 @@ void MergeTreeData::exportPartToTable(
     if (!dest_storage->supportsImport(query_context))
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Destination storage {} does not support MergeTree parts or uses unsupported partitioning", dest_storage->getName());
 
-    auto query_to_string = [] (const ASTPtr & ast)
-    {
-        return ast ? ast->formatWithSecretsOneLine() : "";
-    };
-
     auto source_metadata_ptr = getInMemoryMetadataPtr();
     auto destination_metadata_ptr = dest_storage->getInMemoryMetadataPtr();
 
+    if (dest_storage->isDataLake() && !query_context->getSettingsRef()[Setting::allow_insert_into_iceberg])
+    {
+        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
+            "Iceberg writes are experimental. "
+            "To allow its usage, enable the setting `allow_insert_into_iceberg`.");
+    }
+
+    ExportPartitionUtils::verifyExportSchemaCastable(
+        source_metadata_ptr, destination_metadata_ptr, dest_storage->getStorageID(), query_context);
+
+    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())
     {
-        if (!query_context->getSettingsRef()[Setting::allow_insert_into_iceberg])
-        {
-            throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
-                "Iceberg writes are experimental. "
-                "To allow its usage, enable the setting `allow_insert_into_iceberg`.");
-        }
-
 #if USE_AVRO
         if (iceberg_metadata_json_)
         {
@@ -6745,32 +6749,30 @@ 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_;
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Data lake export requires Avro support");
 #endif
     }
-
-    /// Positional CAST matching, like `INSERT INTO dest SELECT * FROM src`.
-    ExportPartitionUtils::verifyExportSchemaCastable(
-        source_metadata_ptr, destination_metadata_ptr, dest_storage->getStorageID(), query_context);
-
-    /// Iceberg partition compatibility is checked above; here we only need the
-    /// partition-key ASTs to match (partition-column types follow the lossy-cast gate).
-    if (!dest_storage->isDataLake())
+    else
     {
-        if (query_to_string(source_metadata_ptr->getPartitionKeyAST()) != query_to_string(destination_metadata_ptr->getPartitionKeyAST()))
-            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key");
+        /// Plain (hive) object storage writes every row of the part to the one directory computed from
+        /// the destination PARTITION BY on the part's min row, so the source partition must map to a
+        /// single destination partition. Equivalent or finer source keys are accepted.
+        ExportPartitionUtils::verifyPlainPartitionCompatibility(
+            source_metadata_ptr,
+            destination_metadata_ptr,
+            {part},
+            part->info.getPartitionId(),
+            query_context);
     }
 
-    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/StorageReplicatedMergeTree.cpp

@@ -235,6 +235,7 @@ namespace Setting
     extern const SettingsBool allow_insert_into_iceberg;
     extern const SettingsUInt64 iceberg_insert_max_bytes_in_data_file;
     extern const SettingsUInt64 iceberg_insert_max_rows_in_data_file;
+    extern const SettingsTimezone iceberg_partition_timezone;
 }
 
 
@@ -8393,26 +8394,13 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand &
     if (!dest_storage->supportsImport(query_context))
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Destination storage {} does not support MergeTree parts or uses unsupported partitioning", dest_storage->getName());
 
-    auto query_to_string = [] (const ASTPtr & ast)
-    {
-        return ast ? ast->formatWithSecretsOneLine() : "";
-    };
-
     auto src_snapshot = getInMemoryMetadataPtr();
     auto destination_snapshot = dest_storage->getInMemoryMetadataPtr();
 
     /// Positional CAST matching, like `INSERT INTO dest SELECT * FROM src`.
     ExportPartitionUtils::verifyExportSchemaCastable(
         src_snapshot, destination_snapshot, dest_storage->getStorageID(), query_context);
 
-    /// Iceberg partition compatibility is checked below; here we only need the
-    /// partition-key ASTs to match (partition-column types follow the lossy-cast gate).
-    if (!dest_storage->isDataLake())
-    {
-        if (query_to_string(src_snapshot->getPartitionKeyAST()) != query_to_string(destination_snapshot->getPartitionKeyAST()))
-            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key");
-    }
-
     zkutil::ZooKeeperPtr zookeeper = getZooKeeperAndAssertNotReadonly();
 
     const String partition_id = getPartitionIDFromQuery(command.partition, query_context);
@@ -8530,6 +8518,7 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand &
     manifest.filename_pattern = query_context->getSettingsRef()[Setting::export_merge_tree_part_filename_pattern].value;
     manifest.write_full_path_in_iceberg_metadata = query_context->getSettingsRef()[Setting::write_full_path_in_iceberg_metadata];
     manifest.allow_lossy_cast = query_context->getSettingsRef()[Setting::export_merge_tree_part_allow_lossy_cast];
+    manifest.iceberg_partition_timezone = query_context->getSettingsRef()[Setting::iceberg_partition_timezone].toString();
 
     if (dest_storage->isDataLake())
     {
@@ -8564,7 +8553,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);
@@ -8578,6 +8571,15 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand &
         throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Data lake export requires Avro support");
 #endif
     }
+    else
+    {
+        ExportPartitionUtils::verifyPlainPartitionCompatibility(
+            src_snapshot,
+            destination_snapshot,
+            parts,
+            partition_id,
+            query_context);
+    }
 
     ops.emplace_back(zkutil::makeCreateRequest(
         fs::path(partition_exports_path) / "metadata.json", 

tests/integration/helpers/export_partition_helpers.py

@@ -137,15 +137,17 @@ def make_rmt(
     partition_by,
     replica_name="r1",
     order_by="tuple()",
+    extra_settings="",
 ):
     """Create a ReplicatedMergeTree table with block-number settings."""
+    settings = f"{_BLOCK_SETTINGS}, {extra_settings}" if extra_settings else _BLOCK_SETTINGS
     node.query(
         f"""
         CREATE TABLE {name} ({columns})
         ENGINE = ReplicatedMergeTree('/clickhouse/tables/{name}', '{replica_name}')
         PARTITION BY {partition_by}
         ORDER BY {order_by}
-        SETTINGS {_BLOCK_SETTINGS}
+        SETTINGS {settings}
         """
     )