[WIP] Export partition - Positional column matching might lead to incorrect partitioning

src/Storages/MergeTree/ExportPartitionUtils.cpp

@@ -9,6 +9,7 @@
 #include <Storages/MergeTree/MergeTreeData.h>
 #include <filesystem>
 #include <thread>
+#include <unordered_map>
 #include <unordered_set>
 #include <Core/Block.h>
 #include <Core/Settings.h>
@@ -634,6 +635,20 @@ namespace ExportPartitionUtils
     }
 #endif
 
+    void verifyMergeTreePartitionCompatibility(
+        const StorageMetadataPtr & source_metadata,
+        const StorageMetadataPtr & destination_metadata)
+    {
+        constexpr auto query_to_string = [] (const ASTPtr & ast)
+        {
+            return ast ? ast->formatWithSecretsOneLine() : "";
+        };
+
+        if (query_to_string(source_metadata->getPartitionKeyAST()) != query_to_string(destination_metadata->getPartitionKeyAST()))
+            throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                "Cannot export partition: source and destination tables have different `PARTITION BY` expressions");
+    }
+
     void verifyExportSchemaCastable(
         const StorageMetadataPtr & source_metadata,
         const StorageMetadataPtr & destination_metadata,
@@ -657,15 +672,34 @@ namespace ExportPartitionUtils
             ActionsDAG::MatchColumnsMode::Position,
             context);
 
-        /// Lossy casts may silently change values, so reject them unless the user opts in.
-        if (context->getSettingsRef()[Setting::export_merge_tree_part_allow_lossy_cast])
-            return;
+        auto partition_key_columns = source_metadata->getColumnsRequiredForPartitionKey();
+        const std::unordered_set<String> partition_key_column_set(
+            std::make_move_iterator(partition_key_columns.begin()),
+            std::make_move_iterator(partition_key_columns.end()));
+
+        const bool allow_lossy_cast = context->getSettingsRef()[Setting::export_merge_tree_part_allow_lossy_cast];
 
         const size_t num_columns = std::min(source_columns.size(), destination_columns.size());
         for (size_t i = 0; i < num_columns; ++i)
         {
             const auto & source_column = source_columns[i];
             const auto & destination_column = destination_columns[i];
+
+            if (partition_key_column_set.contains(source_column.name) && source_column.name != destination_column.name)
+                throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                    "Cannot export to {}: partition key column '{}' is at position {} in the source "
+                    "table, but the destination's column at that position is named '{}'. EXPORT "
+                    "PART/PARTITION matches columns by position, so partition key columns must be "
+                    "declared at the same position in both tables.",
+                    destination_storage_id.getFullTableName(),
+                    source_column.name,
+                    i,
+                    destination_column.name);
+
+            /// Lossy casts may silently change values, so reject them unless the user opts in.
+            if (allow_lossy_cast)
+                continue;
+
             if (!canBeSafelyCast(source_column.type, destination_column.type))
                 throw Exception(ErrorCodes::INCOMPATIBLE_COLUMNS,
                     "Cannot export to {}: column '{}' requires a lossy cast from {} to {}, "

src/Storages/MergeTree/ExportPartitionUtils.h

@@ -89,6 +89,10 @@ namespace ExportPartitionUtils
         const std::string & exception_message,
         const LoggerPtr & log);
 
+    void verifyMergeTreePartitionCompatibility(
+        const StorageMetadataPtr & source_metadata,
+        const StorageMetadataPtr & destination_metadata);
+
     /// 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.

src/Storages/MergeTree/MergeTreeData.cpp

@@ -6726,11 +6726,6 @@ 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();
 
@@ -6791,13 +6786,8 @@ void MergeTreeData::exportPartToTable(
     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())
-    {
-        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");
-    }
+        ExportPartitionUtils::verifyMergeTreePartitionCompatibility(source_metadata_ptr, destination_metadata_ptr);
 
     auto part = getPartIfExists(part_name, {MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated});
 

src/Storages/StorageReplicatedMergeTree.cpp

@@ -8408,25 +8408,15 @@ 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");
-    }
+        ExportPartitionUtils::verifyMergeTreePartitionCompatibility(src_snapshot, destination_snapshot);
 
     zkutil::ZooKeeperPtr zookeeper = getZooKeeperAndAssertNotReadonly();
 

tests/integration/test_export_merge_tree_part_to_iceberg/test.py

@@ -13,6 +13,12 @@
     test_export_part_with_year_transform_partition        – toYearNumSinceEpoch() partition expression
     test_export_part_with_bucket_partition                – icebergBucket(N, col) partition expression
     test_export_part_partition_key_mismatch_is_rejected   – mismatched partition spec rejected synchronously
+    test_export_part_same_partition_key_different_column_order_single_column – same 1-column partition key, different column order
+    test_export_part_same_partition_key_different_column_order_multi_column  – same 2-column partition key, different column order
+    test_export_part_multi_column_partition_key_success                     – composite (a, b) partition key round-trips
+    test_export_part_multi_column_partition_key_order_mismatch_is_rejected  – composite key fields swapped between src/dst
+    test_export_part_multi_column_partition_key_fewer_in_destination_is_rejected – dst has fewer partition fields than src
+    test_export_part_multi_column_partition_key_more_in_destination_is_rejected  – dst has more partition fields than src
 """
 
 import logging
@@ -462,6 +468,208 @@ def test_export_part_partition_key_mismatch_is_rejected(cluster):
     node.query(f"DROP TABLE IF EXISTS {iceberg}")
 
 
+def test_export_part_same_partition_key_different_column_order_single_column(cluster):
+    node = cluster.instances["node1"]
+    sfx = unique_suffix()
+    mt = f"mt_reordered_{sfx}"
+    iceberg = f"iceberg_reordered_{sfx}"
+
+    make_mt(node, mt, "a Int32, b Int32", "a")
+    make_iceberg_s3(node, iceberg, "b Int32, a Int32", "a")
+
+    node.query(f"INSERT INTO {mt} VALUES (1, 1), (1, 2)")
+
+    part_1 = get_part(node, mt, "1")
+
+    error = node.query_and_get_error(
+        f"ALTER TABLE {mt} EXPORT PART '{part_1}' TO TABLE {iceberg} "
+        f"SETTINGS allow_experimental_export_merge_tree_part = 1, "
+        f"allow_experimental_insert_into_iceberg = 1"
+    )
+    assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}"
+    assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error!r}"
+
+    count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
+    assert count == 0, f"Expected 0 rows in Iceberg table after rejected export, got {count}"
+
+    node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
+    node.query(f"DROP TABLE IF EXISTS {iceberg}")
+
+
+def test_export_part_same_partition_key_different_column_order_multi_column(cluster):
+    node = cluster.instances["node1"]
+    sfx = unique_suffix()
+    mt = f"mt_multi_reordered_{sfx}"
+    iceberg = f"iceberg_multi_reordered_{sfx}"
+
+    make_mt(node, mt, "a Int32, b Int32, c Int32, val String", "(a, b, c)")
+    make_iceberg_s3(node, iceberg, "c Int32, b Int32, a Int32, val String", "(a, b, c)")
+
+    node.query(f"INSERT INTO {mt} VALUES (1, 1, 1, 'x'), (1, 1, 1, 'y')")
+
+    pid = first_partition_id(node, mt)
+    part = get_part(node, mt, pid)
+
+    error = node.query_and_get_error(
+        f"ALTER TABLE {mt} EXPORT PART '{part}' TO TABLE {iceberg} "
+        f"SETTINGS allow_experimental_export_merge_tree_part = 1, "
+        f"allow_experimental_insert_into_iceberg = 1"
+    )
+    assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}"
+    assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error!r}"
+
+    count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
+    assert count == 0, f"Expected 0 rows in Iceberg table after rejected export, got {count}"
+
+    node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
+    node.query(f"DROP TABLE IF EXISTS {iceberg}")
+
+
+def test_export_part_multi_column_partition_key_success(cluster):
+    node = cluster.instances["node1"]
+    sfx = unique_suffix()
+    mt = f"mt_multi_pkey_ok_{sfx}"
+    iceberg = f"iceberg_multi_pkey_ok_{sfx}"
+
+    cols = "a Int32, b Int32, c Int32, val String"
+    make_mt(node, mt, cols, "(a, b, c)")
+    make_iceberg_s3(node, iceberg, cols, "(a, b, c)")
+
+    node.query(f"INSERT INTO {mt} VALUES (1, 1, 1, 'x'), (1, 1, 1, 'y')")
+
+    pid = first_partition_id(node, mt)
+    part = get_part(node, mt, pid)
+    export_part(node, mt, part, iceberg)
+    wait_for_export_part(node, mt, part)
+
+    count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
+    assert count == 2, f"Expected 2 rows in Iceberg table after export, got {count}"
+
+    result = node.query(f"SELECT a, b, c, val FROM {iceberg} ORDER BY val").strip()
+    assert result == "1\t1\t1\tx\n1\t1\t1\ty", f"Unexpected exported data:\n{result}"
+
+    assert_part_log(node, mt, part)
+
+    node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
+    node.query(f"DROP TABLE IF EXISTS {iceberg}")
+
+
+def test_export_part_multi_column_partition_key_order_mismatch_is_rejected(cluster):
+    node = cluster.instances["node1"]
+    sfx = unique_suffix()
+    mt = f"mt_multi_order_{sfx}"
+    iceberg = f"iceberg_multi_order_{sfx}"
+
+    cols = "a Int32, b Int32, c Int32, val String"
+    make_mt(node, mt, cols, "(a, b, c)")
+    make_iceberg_s3(node, iceberg, cols, "(c, b, a)")
+
+    node.query(f"INSERT INTO {mt} VALUES (1, 2, 3, 'x')")
+
+    pid = first_partition_id(node, mt)
+    part = get_part(node, mt, pid)
+
+    error = node.query_and_get_error(
+        f"ALTER TABLE {mt} EXPORT PART '{part}' TO TABLE {iceberg} "
+        f"SETTINGS allow_experimental_export_merge_tree_part = 1, "
+        f"allow_experimental_insert_into_iceberg = 1"
+    )
+    assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}"
+
+    count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
+    assert count == 0, f"Expected 0 rows in Iceberg table after rejected export, got {count}"
+
+    node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
+    node.query(f"DROP TABLE IF EXISTS {iceberg}")
+
+
+def test_export_part_multi_column_partition_key_fewer_in_destination_is_rejected(cluster):
+    node = cluster.instances["node1"]
+    sfx = unique_suffix()
+    mt = f"mt_multi_fewer_{sfx}"
+    iceberg = f"iceberg_multi_fewer_{sfx}"
+
+    cols = "a Int32, b Int32, c Int32, val String"
+    make_mt(node, mt, cols, "(a, b, c)")
+    make_iceberg_s3(node, iceberg, cols, "(a, b)")
+
+    node.query(f"INSERT INTO {mt} VALUES (1, 2, 3, 'x')")
+
+    pid = first_partition_id(node, mt)
+    part = get_part(node, mt, pid)
+
+    error = node.query_and_get_error(
+        f"ALTER TABLE {mt} EXPORT PART '{part}' TO TABLE {iceberg} "
+        f"SETTINGS allow_experimental_export_merge_tree_part = 1, "
+        f"allow_experimental_insert_into_iceberg = 1"
+    )
+    assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}"
+
+    count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
+    assert count == 0, f"Expected 0 rows in Iceberg table after rejected export, got {count}"
+
+    node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
+    node.query(f"DROP TABLE IF EXISTS {iceberg}")
+
+
+def test_export_part_multi_column_partition_key_more_in_destination_is_rejected(cluster):
+    node = cluster.instances["node1"]
+    sfx = unique_suffix()
+    mt = f"mt_multi_more_{sfx}"
+    iceberg = f"iceberg_multi_more_{sfx}"
+
+    cols = "a Int32, b Int32, c Int32, val String"
+    make_mt(node, mt, cols, "(a, b)")
+    make_iceberg_s3(node, iceberg, cols, "(a, b, c)")
+
+    node.query(f"INSERT INTO {mt} VALUES (1, 2, 3, 'x')")
+
+    pid = first_partition_id(node, mt)
+    part = get_part(node, mt, pid)
+
+    error = node.query_and_get_error(
+        f"ALTER TABLE {mt} EXPORT PART '{part}' TO TABLE {iceberg} "
+        f"SETTINGS allow_experimental_export_merge_tree_part = 1, "
+        f"allow_experimental_insert_into_iceberg = 1"
+    )
+    assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}"
+
+    count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
+    assert count == 0, f"Expected 0 rows in Iceberg table after rejected export, got {count}"
+
+    node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
+    node.query(f"DROP TABLE IF EXISTS {iceberg}")
+
+
+def test_export_part_transform_partition_key_different_column_order_is_rejected(cluster):
+    node = cluster.instances["node1"]
+    sfx = unique_suffix()
+    mt = f"mt_transform_reordered_{sfx}"
+    iceberg = f"iceberg_transform_reordered_{sfx}"
+
+    make_mt(node, mt, "other_id Int64, user_id Int64", "icebergBucket(8, user_id)")
+    make_iceberg_s3(node, iceberg, "user_id Int64, other_id Int64", "icebergBucket(8, user_id)")
+
+    node.query(f"INSERT INTO {mt} VALUES (1, 42)")
+
+    pid = first_partition_id(node, mt)
+    part = get_part(node, mt, pid)
+
+    error = node.query_and_get_error(
+        f"ALTER TABLE {mt} EXPORT PART '{part}' TO TABLE {iceberg} "
+        f"SETTINGS allow_experimental_export_merge_tree_part = 1, "
+        f"allow_experimental_insert_into_iceberg = 1"
+    )
+    assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error!r}"
+    assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error!r}"
+
+    count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
+    assert count == 0, f"Expected 0 rows in Iceberg table after rejected export, got {count}"
+
+    node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
+    node.query(f"DROP TABLE IF EXISTS {iceberg}")
+
+
 def test_export_part_with_bucket_partition(cluster):
     """
     Export a part from a MergeTree table partitioned by icebergBucket(8, user_id)