Export partition - allow non matching partition expressions in case we can prove the destination expression does not split the data

docs/en/antalya/part_export.md

@@ -53,10 +53,10 @@ Source and destination tables must support positional schema conversion. The fol
 - **Column types** may differ, as long as the source type is safely castable to the destination type. Set `export_merge_tree_part_allow_lossy_cast = 1` to also permit lossy casts.
 - **`Tuple` element names** may differ if either the source or destination declares the tuple without named elements: an unnamed `Tuple` (e.g. `Tuple(Int32, Int32)`) is matched against the destination by element position and type only, not by name. For example, exporting from `t Tuple(Int32, Int32)` to `t Tuple(x Int32, y Int32)` is allowed as long as element types match positionally.
 
-The following must match between source and destination:
+The following requirements apply to the source and destination:
 
 1. **Column count** - source and destination must have the same number of columns by default. A mismatch in either direction throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode.
-2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must be representable as an Iceberg partition spec and must match the destination partition fields and transforms.
+2. **`PARTITION BY` expressions** - the whole part must land in a single destination partition. Identical expressions always satisfy this; otherwise the destination expression has to be computable from the values the source partition key pins, or be proven single-valued over the part's min/max range. The same requirement applies to the partition fields and transforms of an Apache Iceberg destination. See [Source partition key compatibility](/docs/en/antalya/partition_export.md#source-partition-key-compatibility).
 3. **The position of every column backing the partition key** - it is not enough for the `PARTITION BY` expressions to be textually identical: every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. If such a column contains a named `Tuple`, its element names must also be declared in the same order (an unnamed `Tuple` on either side is exempt from this, per the allowance above). This comparison is recursive through nested tuples and through container types such as `Array` and `Map`.
 
   For example, `CREATE TABLE src (a Int32, b Int32) ... PARTITION BY a` and `CREATE TABLE dst (b Int32, a Int32) ... PARTITION BY a` both have the expression `PARTITION BY a`, but `a` is at position 0 in `src` and position 1 in `dst`. The export is rejected with a `BAD_ARGUMENTS` exception whose message includes `Cannot export to <destination>: partition key column 'a' is at position 0 in the source table, but the destination's column at that position is named 'b'`.

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. Structural match: in case the source and destination are identical, the destination expression is a subset of the source expression or the destination expression can be entirely computed using only constants and the exact values guaranteed (pinned) by the source.
+2. Dynamic proof: the destination expression is monotonic over the source partition min/max range.
+
 ### 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>`.
@@ -45,10 +52,10 @@ TO TABLE [destination_database.]destination_table
 
 ## Requirements
 
-`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Column names may differ (columns are matched by position, not by name), and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following must match:
+`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Column names may differ (columns are matched by position, not by name), and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following requirements apply:
 
 1. **Column count** - source and destination must have the same number of columns by default. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode.
-2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must match the destination partition fields and transforms.
+2. **`PARTITION BY` expressions** - the whole source partition must land in a single destination partition. Identical expressions always satisfy this; otherwise the destination expression has to be computable from the values the source partition key pins, or be proven single-valued over the partition's min/max range. The same requirement applies to the partition fields and transforms of an Apache Iceberg destination. See [Source partition key compatibility](#source-partition-key-compatibility).
 3. **Partition key column positions and layouts** - every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. Named `Tuple` elements within such a column must also be declared in the same order, including tuples nested inside `Array` or `Map`. This applies even if both tables' `PARTITION BY` expressions are textually identical. See [`EXPORT PART` requirements](/docs/en/antalya/part_export.md#requirements) for a worked example and the corresponding exception message.
 
 ## Settings

src/Processors/QueryPlan/Optimizations/actionsDAGUtils.cpp

@@ -245,16 +245,7 @@ MatchedTrees::Matches matchTrees(const ActionsDAG::NodeRawConstPtrs & inner_dag,
 }
 
 
-struct PossiblyMonotonicChain
-{
-    const ActionsDAG::Node * input_node = nullptr;
-    std::vector<size_t> non_const_arg_pos;
-    bool changes_order = false;
-    bool is_strict = true;
-};
-
-/// Build a chain of functions which may be monotonic.
-static PossiblyMonotonicChain buildPossiblyMonitinicChain(const ActionsDAG::Node * node)
+PossiblyMonotonicChain buildPossiblyMonotonicChain(const ActionsDAG::Node * node)
 {
     std::vector<size_t> chain;
 
@@ -307,7 +298,6 @@ static PossiblyMonotonicChain buildPossiblyMonitinicChain(const ActionsDAG::Node
     return {node, std::move(chain)};
 }
 
-/// Check whether all the function in chain are monotonic
 bool isMonotonicChain(const ActionsDAG::Node * node, PossiblyMonotonicChain & chain)
 {
     auto it = chain.non_const_arg_pos.begin();
@@ -388,7 +378,7 @@ void applyActionsToSortDescription(
         if (output == output_to_skip)
             continue;
 
-        auto chain = buildPossiblyMonitinicChain(output);
+        auto chain = buildPossiblyMonotonicChain(output);
         if (!chain.input_node)
             break;
 

src/Processors/QueryPlan/Optimizations/actionsDAGUtils.h

@@ -48,6 +48,22 @@ struct MatchedTrees
 
 MatchedTrees::Matches matchTrees(const ActionsDAG::NodeRawConstPtrs & inner_dag, const ActionsDAG & outer_dag, bool check_monotonicity = true);
 
+/// A path from a node down to an input, where every function on the path has a single non-constant argument.
+/// `non_const_arg_pos` holds the position of that argument for every function on the path, top-down.
+struct PossiblyMonotonicChain
+{
+    const ActionsDAG::Node * input_node = nullptr;
+    std::vector<size_t> non_const_arg_pos;
+    bool changes_order = false;
+    bool is_strict = true;
+};
+
+/// Build a chain of functions which may be monotonic. `input_node` is nullptr if the node is not such a chain.
+PossiblyMonotonicChain buildPossiblyMonotonicChain(const ActionsDAG::Node * node);
+
+/// Check whether all the function in chain are monotonic
+bool isMonotonicChain(const ActionsDAG::Node * node, PossiblyMonotonicChain & chain);
+
 /// Update SortDescription (inplace) by applying ActionsDAG.
 ///
 /// Assuming that sorting properties are fulfilled for inputs, calculate sorting properties for the outputs.

src/Storages/ExportReplicatedMergeTreePartitionManifest.h

@@ -250,6 +250,11 @@ struct ExportReplicatedMergeTreePartitionManifest
     std::optional<UInt64> parquet_row_group_size_bytes;
     std::optional<MergeTreePartExportSchemaMismatchMode> schema_mismatch_mode;
 
+    /// 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;
@@ -291,6 +296,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);
         if (schema_mismatch_mode)
             json.set("schema_mismatch_mode", String(magic_enum::enum_name(*schema_mismatch_mode)));
         std::ostringstream oss;     // STYLE_CHECK_ALLOW_STD_STRING_STREAM
@@ -391,6 +398,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;
     }
 };