From 9209361362bdf9bedc094b10c018b592da939728 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Thu, 30 Jul 2026 15:08:41 +0200 Subject: [PATCH 1/4] add tests Signed-off-by: Konstantin Morozov --- .../test.py | 24 +++++++++ .../test.py | 49 +++++++++++++++++++ .../test.py | 23 +++++++++ .../test.py | 32 ++++++++++++ 4 files changed, 128 insertions(+) diff --git a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py index 486adf1f2b17..78a0b94d2414 100644 --- a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py +++ b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py @@ -13,6 +13,7 @@ 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 – same partition key, different column order """ import logging @@ -462,6 +463,29 @@ 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(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") + export_part(node, mt, part_1, iceberg) + wait_for_export_part(node, mt, part_1) + + result = node.query(f"SELECT a, b FROM {iceberg} ORDER BY a, b").strip() + expected = "1\t1\n1\t2" + assert result == expected, f"Expected:\n{expected}\nGot:\n{result}" + + 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) diff --git a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py index b8c15c26275f..10bd5959f3e1 100644 --- a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py +++ b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py @@ -312,3 +312,52 @@ def test_pending_patch_parts_skip_before_export(cluster): assert "1\n2\n3" in result, "Export should contain original data before patch" node.query(f"DROP TABLE {mt_table}") + + +def test_export_part_same_partition_key_different_column_order(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"reordered_part_mt_table_{postfix}" + s3_table = f"reordered_part_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32) + ENGINE = MergeTree() + PARTITION BY a + ORDER BY tuple() + SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1 + """) + + node.query(f""" + CREATE TABLE {s3_table} (b Int32, a Int32) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY a + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1), (1, 2)") + + part_name = node.query( + f"SELECT name FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + node.query(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") + + deadline = time.time() + 30 + while True: + node.query("SYSTEM FLUSH LOGS") + count = node.query( + f"SELECT count() FROM system.part_log WHERE event_type = 'ExportPart' " + f"AND database = currentDatabase() AND table = '{mt_table}' AND part_name = '{part_name}'" + ).strip() + if count != "0": + break + if time.time() > deadline: + raise TimeoutError(f"ExportPart event for part {part_name!r} did not appear within 30s") + time.sleep(0.5) + + dest_result = node.query(f"SELECT a, b FROM {s3_table} ORDER BY a, b") + expected = "1\t1\n1\t2\n" + assert dest_result == expected, f"Expected:\n{expected}\nGot:\n{dest_result}" diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index ad2deba8de19..844f79e85f7c 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -1324,6 +1324,29 @@ def test_export_partition_with_renamed_destination_column(cluster): ) +def test_export_partition_same_partition_key_different_column_order(cluster): + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_reordered_{uid}" + iceberg_table = f"iceberg_reordered_{uid}" + + make_rmt(node, mt_table, "a Int32, b Int32", "a", replica_name="replica1") + make_iceberg_s3(node, iceberg_table, "b Int32, a Int32", partition_by="a") + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1), (1, 2)") + + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '1' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + wait_for_export_status(node, mt_table, iceberg_table, "1", "COMPLETED") + + result = node.query(f"SELECT a, b FROM {iceberg_table} ORDER BY a, b").strip() + expected = "1\t1\n1\t2" + assert result == expected, f"Expected:\n{expected}\nGot:\n{result}" + + def test_export_partition_with_castable_widening(cluster): """A lossless widening of both a data column (id Int32 -> Int64) and the partition column (year Int32 -> Int64) round-trips.""" diff --git a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py index 8d4589292e3c..6a6e63e36bc1 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py @@ -1747,3 +1747,35 @@ def test_export_partition_all_failure_modes(cluster): f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}" f" SETTINGS export_merge_tree_partition_all_on_error = 'skip_conflicts'" ) + + +def test_export_partition_same_partition_key_different_column_order(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"reordered_mt_table_{postfix}" + s3_table = f"reordered_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32) + ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1') + PARTITION BY a + ORDER BY tuple() + """) + + node.query(f""" + CREATE TABLE {s3_table} (b Int32, a Int32) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY a + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1), (1, 2)") + + node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '1' TO TABLE {s3_table}") + wait_for_export_status(node, mt_table, s3_table, "1", "COMPLETED") + + dest_result = node.query(f"SELECT * FROM {s3_table} ORDER BY a, b") + + expected = "1\t1\n2\t1\n" + assert dest_result == expected, f"Expected:\n{expected}\nGot:\n{dest_result}" From 1a6d63c21c0376a60f858ac5e26b9786ea444814 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Thu, 30 Jul 2026 15:09:35 +0200 Subject: [PATCH 2/4] refactoring Signed-off-by: Konstantin Morozov --- src/Storages/MergeTree/ExportPartitionUtils.cpp | 13 +++++++++++++ src/Storages/MergeTree/ExportPartitionUtils.h | 4 ++++ src/Storages/MergeTree/MergeTreeData.cpp | 12 +----------- src/Storages/StorageReplicatedMergeTree.cpp | 12 +----------- 4 files changed, 19 insertions(+), 22 deletions(-) diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 3f85ebc0e1fb..616a3384ba1c 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -634,6 +634,19 @@ 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, "Tables have different partition key"); + } + void verifyExportSchemaCastable( const StorageMetadataPtr & source_metadata, const StorageMetadataPtr & destination_metadata, diff --git a/src/Storages/MergeTree/ExportPartitionUtils.h b/src/Storages/MergeTree/ExportPartitionUtils.h index 0bb8acb9bda4..dd1ae4c18094 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.h +++ b/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. diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index 0ed451b59655..ac8ad03c3b63 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/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}); diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index 5677a5e111a8..759736dc06be 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -8408,11 +8408,6 @@ 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(); @@ -8420,13 +8415,8 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & 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(); From 420f2e8ebf5cef375930bfca3307581a05a226c6 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Thu, 30 Jul 2026 16:55:57 +0200 Subject: [PATCH 3/4] add check names and position Signed-off-by: Konstantin Morozov --- .../MergeTree/ExportPartitionUtils.cpp | 29 ++++++++++++++++--- .../test.py | 15 ++++++---- .../test.py | 24 ++++----------- .../test.py | 10 +++---- .../test.py | 11 ++++--- 5 files changed, 51 insertions(+), 38 deletions(-) diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 616a3384ba1c..39256852b002 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -644,7 +645,8 @@ namespace ExportPartitionUtils }; if (query_to_string(source_metadata->getPartitionKeyAST()) != query_to_string(destination_metadata->getPartitionKeyAST())) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key"); + throw Exception(ErrorCodes::BAD_ARGUMENTS, + "Cannot export partition: source and destination tables have different `PARTITION BY` expressions"); } void verifyExportSchemaCastable( @@ -670,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 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 {}, " diff --git a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py index 78a0b94d2414..cefd25cc0796 100644 --- a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py +++ b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py @@ -475,12 +475,17 @@ def test_export_part_same_partition_key_different_column_order(cluster): node.query(f"INSERT INTO {mt} VALUES (1, 1), (1, 2)") part_1 = get_part(node, mt, "1") - export_part(node, mt, part_1, iceberg) - wait_for_export_part(node, mt, part_1) - result = node.query(f"SELECT a, b FROM {iceberg} ORDER BY a, b").strip() - expected = "1\t1\n1\t2" - assert result == expected, f"Expected:\n{expected}\nGot:\n{result}" + 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}") diff --git a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py index 10bd5959f3e1..e9b43c0141bb 100644 --- a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py +++ b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py @@ -343,21 +343,9 @@ def test_export_part_same_partition_key_different_column_order(cluster): f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" ).strip() - node.query(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") - - deadline = time.time() + 30 - while True: - node.query("SYSTEM FLUSH LOGS") - count = node.query( - f"SELECT count() FROM system.part_log WHERE event_type = 'ExportPart' " - f"AND database = currentDatabase() AND table = '{mt_table}' AND part_name = '{part_name}'" - ).strip() - if count != "0": - break - if time.time() > deadline: - raise TimeoutError(f"ExportPart event for part {part_name!r} did not appear within 30s") - time.sleep(0.5) - - dest_result = node.query(f"SELECT a, b FROM {s3_table} ORDER BY a, b") - expected = "1\t1\n1\t2\n" - assert dest_result == expected, f"Expected:\n{expected}\nGot:\n{dest_result}" + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index 844f79e85f7c..f4fa3f71da90 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -1336,15 +1336,15 @@ def test_export_partition_same_partition_key_different_column_order(cluster): node.query(f"INSERT INTO {mt_table} VALUES (1, 1), (1, 2)") - node.query( + error = node.query_and_get_error( f"ALTER TABLE {mt_table} EXPORT PARTITION ID '1' TO TABLE {iceberg_table}", settings={"allow_insert_into_iceberg": 1}, ) - wait_for_export_status(node, mt_table, iceberg_table, "1", "COMPLETED") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" - result = node.query(f"SELECT a, b FROM {iceberg_table} ORDER BY a, b").strip() - expected = "1\t1\n1\t2" - assert result == expected, f"Expected:\n{expected}\nGot:\n{result}" + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" def test_export_partition_with_castable_widening(cluster): diff --git a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py index 6a6e63e36bc1..7cb9b71202fa 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py @@ -1772,10 +1772,9 @@ def test_export_partition_same_partition_key_different_column_order(cluster): node.query(f"INSERT INTO {mt_table} VALUES (1, 1), (1, 2)") - node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '1' TO TABLE {s3_table}") - wait_for_export_status(node, mt_table, s3_table, "1", "COMPLETED") - - dest_result = node.query(f"SELECT * FROM {s3_table} ORDER BY a, b") + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '1' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" - expected = "1\t1\n2\t1\n" - assert dest_result == expected, f"Expected:\n{expected}\nGot:\n{dest_result}" + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" From 5322e48aae583cbaea6a6f427f0beb24c7248b53 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Fri, 31 Jul 2026 17:10:22 +0200 Subject: [PATCH 4/4] add tests Signed-off-by: Konstantin Morozov --- .../test.py | 183 +++++++++++++- .../test.py | 187 +++++++++++++- .../test.py | 144 ++++++++++- .../test.py | 236 +++++++++++++++++- 4 files changed, 744 insertions(+), 6 deletions(-) diff --git a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py index cefd25cc0796..5494846b4433 100644 --- a/tests/integration/test_export_merge_tree_part_to_iceberg/test.py +++ b/tests/integration/test_export_merge_tree_part_to_iceberg/test.py @@ -13,7 +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 – same partition key, different column order + 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 @@ -463,7 +468,7 @@ 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(cluster): +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}" @@ -491,6 +496,180 @@ def test_export_part_same_partition_key_different_column_order(cluster): 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) diff --git a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py index e9b43c0141bb..132d13ecc303 100644 --- a/tests/integration/test_export_merge_tree_part_to_object_storage/test.py +++ b/tests/integration/test_export_merge_tree_part_to_object_storage/test.py @@ -314,7 +314,7 @@ def test_pending_patch_parts_skip_before_export(cluster): node.query(f"DROP TABLE {mt_table}") -def test_export_part_same_partition_key_different_column_order(cluster): +def test_export_part_same_partition_key_different_column_order_single_column(cluster): skip_if_remote_database_disk_enabled(cluster) node = cluster.instances["node1"] @@ -349,3 +349,188 @@ def test_export_part_same_partition_key_different_column_order(cluster): count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_part_same_partition_key_different_column_order_multi_column(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_reordered_part_mt_table_{postfix}" + s3_table = f"multi_reordered_part_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = MergeTree() + PARTITION BY (a, b, c) + ORDER BY tuple() + SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1 + """) + + node.query(f""" + CREATE TABLE {s3_table} (c Int32, b Int32, a Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b, c) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1, 1, 'x'), (1, 1, 1, 'y')") + + part_name = node.query( + f"SELECT name FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_part_multi_column_partition_key_success(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_pkey_ok_mt_table_{postfix}" + s3_table = f"multi_pkey_ok_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = MergeTree() + PARTITION BY (a, b, c) + ORDER BY tuple() + SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1 + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b, c) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1, 1, 'x'), (1, 1, 1, 'y')") + + part_name = node.query( + f"SELECT name FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + node.query(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") + + time.sleep(5) + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 2, f"Expected 2 rows in destination after export, got {count}" + + result = node.query(f"SELECT a, b, c, val FROM {s3_table} ORDER BY val").strip() + assert result == "1\t1\t1\tx\n1\t1\t1\ty", f"Unexpected exported data:\n{result}" + + +def test_export_part_multi_column_partition_key_order_mismatch_is_rejected(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_order_mt_table_{postfix}" + s3_table = f"multi_order_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = MergeTree() + PARTITION BY (a, b, c) + ORDER BY tuple() + SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1 + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (c, b, a) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x')") + + part_name = node.query( + f"SELECT name FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_part_multi_column_partition_key_fewer_in_destination_is_rejected(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_fewer_mt_table_{postfix}" + s3_table = f"multi_fewer_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = MergeTree() + PARTITION BY (a, b, c) + ORDER BY tuple() + SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1 + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x')") + + part_name = node.query( + f"SELECT name FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_part_multi_column_partition_key_more_in_destination_is_rejected(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_more_mt_table_{postfix}" + s3_table = f"multi_more_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = MergeTree() + PARTITION BY (a, b) + ORDER BY tuple() + SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1 + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b, c) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x')") + + part_name = node.query( + f"SELECT name FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PART '{part_name}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index f4fa3f71da90..e3aa21baeb33 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -754,6 +754,7 @@ def check_accepted(mt, iceberg, description): f"ALTER TABLE {mt} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg}", settings={"allow_insert_into_iceberg": 1}, ) + return pid # 1. Compound identity: (year, region) cols = "id Int64, year Int32, region String" @@ -761,7 +762,12 @@ def check_accepted(mt, iceberg, description): make_rmt(node, t, cols, "(year, region)") node.query(f"INSERT INTO {t} VALUES (1, 2023, 'EU')") make_iceberg_s3(node, i, cols, "(year, region)") - check_accepted(t, i, "compound identity (year, region)") + pid = check_accepted(t, i, "compound identity (year, region)") + wait_for_export_status(node, t, i, pid, "COMPLETED") + count = int(node.query(f"SELECT count() FROM {i}").strip()) + assert count == 1, f"[compound identity (year, region)] Expected 1 row in Iceberg table, got {count}" + result = node.query(f"SELECT id, year, region FROM {i}").strip() + assert result == "1\t2023\tEU", f"[compound identity (year, region)] Unexpected exported data:\n{result}" # 2. Year transform cols = "id Int64, event_date Date" @@ -837,6 +843,8 @@ def assert_rejected(mt, iceberg, description): node.query(f"INSERT INTO {t} VALUES (1, 2020, 'EU')") make_iceberg_s3(node, i, cols, "(region, year)") assert_rejected(t, i, "compound field order reversed") + count = int(node.query(f"SELECT count() FROM {i}").strip()) + assert count == 0, f"[compound field order reversed] Expected 0 rows in destination, got {count}" # 2. Transform mismatch: MergeTree year-transform, Iceberg identity on same Date col cols = "id Int64, event_date Date" @@ -869,6 +877,8 @@ def assert_rejected(mt, iceberg, description): node.query(f"INSERT INTO {t} VALUES (1, 2020, 'EU')") make_iceberg_s3(node, i, cols, "year") assert_rejected(t, i, "2-field MergeTree vs 1-field Iceberg") + count = int(node.query(f"SELECT count() FROM {i}").strip()) + assert count == 0, f"[2-field MergeTree vs 1-field Iceberg] Expected 0 rows in destination, got {count}" # 6. Unsupported MergeTree expression: intDiv(year, 100) is not an Iceberg transform cols = "id Int64, year Int32" @@ -1324,7 +1334,7 @@ def test_export_partition_with_renamed_destination_column(cluster): ) -def test_export_partition_same_partition_key_different_column_order(cluster): +def test_export_partition_same_partition_key_different_column_order_single_column(cluster): node = cluster.instances["replica1"] uid = unique_suffix() @@ -1343,6 +1353,136 @@ def test_export_partition_same_partition_key_different_column_order(cluster): assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" + error_all = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_same_partition_key_different_column_order_multi_column(cluster): + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_multi_reordered_{uid}" + iceberg_table = f"iceberg_multi_reordered_{uid}" + + cols = "a Int32, b Int32, c Int32, val String" + make_rmt(node, mt_table, cols, "(a, b, c)", replica_name="replica1") + make_iceberg_s3(node, iceberg_table, "c Int32, b Int32, a Int32, val String", partition_by="(a, b, c)") + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1, 1, 'x'), (1, 1, 1, 'y')") + + pid = first_partition_id(node, mt_table) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" + + error_all = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_multi_column_partition_key_more_in_destination_is_rejected(cluster): + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_multi_more_{uid}" + iceberg_table = f"iceberg_multi_more_{uid}" + + cols = "a Int32, b Int32, c Int32, val String" + make_rmt(node, mt_table, cols, "(a, b)", replica_name="replica1") + make_iceberg_s3(node, iceberg_table, cols, partition_by="(a, b, c)") + + node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x')") + + pid = first_partition_id(node, mt_table) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + + error_all = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_multi_column_partition_key_success_all(cluster): + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_multi_pkey_ok_all_{uid}" + iceberg_table = f"iceberg_multi_pkey_ok_all_{uid}" + + cols = "a Int32, b Int32, c Int32, val String" + make_rmt(node, mt_table, cols, "(a, b, c)", replica_name="replica1") + make_iceberg_s3(node, iceberg_table, cols, partition_by="(a, b, c)") + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1, 1, 'x'), (2, 2, 2, 'y')") + + partition_ids = node.query( + f"SELECT DISTINCT partition_id FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY partition_id" + ).strip().split("\n") + + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + + for pid in partition_ids: + wait_for_export_status(node, mt_table, iceberg_table, pid, "COMPLETED") + + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) + assert count == 2, f"Expected 2 rows in destination after export, got {count}" + + result = node.query(f"SELECT a, b, c, val FROM {iceberg_table} ORDER BY val").strip() + assert result == "1\t1\t1\tx\n2\t2\t2\ty", f"Unexpected exported data:\n{result}" + + +def test_export_partition_transform_partition_key_different_column_order_is_rejected(cluster): + node = cluster.instances["replica1"] + + uid = unique_suffix() + mt_table = f"mt_transform_reordered_{uid}" + iceberg_table = f"iceberg_transform_reordered_{uid}" + + make_rmt(node, mt_table, "other_id Int64, user_id Int64", "icebergBucket(8, user_id)", replica_name="replica1") + make_iceberg_s3(node, iceberg_table, "user_id Int64, other_id Int64", partition_by="icebergBucket(8, user_id)") + + node.query(f"INSERT INTO {mt_table} VALUES (1, 42)") + + pid = first_partition_id(node, mt_table) + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{pid}' TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" + + error_all = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {iceberg_table}", + settings={"allow_insert_into_iceberg": 1}, + ) + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip()) assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" diff --git a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py index 7cb9b71202fa..cb73cc926f4e 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py @@ -1749,7 +1749,7 @@ def test_export_partition_all_failure_modes(cluster): ) -def test_export_partition_same_partition_key_different_column_order(cluster): +def test_export_partition_same_partition_key_different_column_order_single_column(cluster): skip_if_remote_database_disk_enabled(cluster) node = cluster.instances["replica1"] @@ -1776,5 +1776,239 @@ def test_export_partition_same_partition_key_different_column_order(cluster): assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" + error_all = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_same_partition_key_different_column_order_multi_column(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_reordered_mt_table_{postfix}" + s3_table = f"multi_reordered_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1') + PARTITION BY (a, b, c) + ORDER BY tuple() + """) + + node.query(f""" + CREATE TABLE {s3_table} (c Int32, b Int32, a Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b, c) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1, 1, 'x'), (1, 1, 1, 'y')") + + partition_id = node.query( + f"SELECT partition_id FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{partition_id}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + assert "partition key column" in error, f"Expected partition key column mismatch message, got: {error}" + + error_all = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_multi_column_partition_key_success(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_pkey_ok_mt_table_{postfix}" + s3_table = f"multi_pkey_ok_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1') + PARTITION BY (a, b, c) + ORDER BY tuple() + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b, c) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1, 1, 'x'), (1, 1, 1, 'y')") + + partition_id = node.query( + f"SELECT partition_id FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{partition_id}' TO TABLE {s3_table}") + wait_for_export_status(node, mt_table, s3_table, partition_id, "COMPLETED") + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 2, f"Expected 2 rows in destination after export, got {count}" + + result = node.query(f"SELECT a, b, c, val FROM {s3_table} ORDER BY val").strip() + assert result == "1\t1\t1\tx\n1\t1\t1\ty", f"Unexpected exported data:\n{result}" + + +def test_export_partition_multi_column_partition_key_order_mismatch_is_rejected(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_order_mt_table_{postfix}" + s3_table = f"multi_order_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1') + PARTITION BY (a, b, c) + ORDER BY tuple() + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (c, b, a) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x')") + + partition_id = node.query( + f"SELECT partition_id FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{partition_id}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + + error_all = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_multi_column_partition_key_fewer_in_destination_is_rejected(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_fewer_mt_table_{postfix}" + s3_table = f"multi_fewer_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1') + PARTITION BY (a, b, c) + ORDER BY tuple() + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x')") + + partition_id = node.query( + f"SELECT partition_id FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{partition_id}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + + error_all = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_multi_column_partition_key_more_in_destination_is_rejected(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_more_mt_table_{postfix}" + s3_table = f"multi_more_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1') + PARTITION BY (a, b) + ORDER BY tuple() + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b, c) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 2, 3, 'x')") + + partition_id = node.query( + f"SELECT partition_id FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY name LIMIT 1" + ).strip() + + error = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '{partition_id}' TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error, f"Expected BAD_ARGUMENTS, got: {error}" + + error_all = node.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}") + assert "BAD_ARGUMENTS" in error_all, f"Expected BAD_ARGUMENTS, got: {error_all}" + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) assert count == 0, f"Expected 0 rows in destination after rejected export, got {count}" + + +def test_export_partition_multi_column_partition_key_success_all(cluster): + skip_if_remote_database_disk_enabled(cluster) + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"multi_pkey_ok_all_mt_table_{postfix}" + s3_table = f"multi_pkey_ok_all_s3_table_{postfix}" + + node.query(f""" + CREATE TABLE {mt_table} (a Int32, b Int32, c Int32, val String) + ENGINE = ReplicatedMergeTree('/clickhouse/tables/{mt_table}', 'replica1') + PARTITION BY (a, b, c) + ORDER BY tuple() + """) + + node.query(f""" + CREATE TABLE {s3_table} (a Int32, b Int32, c Int32, val String) + ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') + PARTITION BY (a, b, c) + """) + + node.query(f"INSERT INTO {mt_table} VALUES (1, 1, 1, 'x'), (2, 2, 2, 'y')") + + partition_ids = node.query( + f"SELECT DISTINCT partition_id FROM system.parts WHERE database = currentDatabase() " + f"AND table = '{mt_table}' AND active ORDER BY partition_id" + ).strip().split("\n") + + node.query(f"ALTER TABLE {mt_table} EXPORT PARTITION ALL TO TABLE {s3_table}") + + for pid in partition_ids: + wait_for_export_status(node, mt_table, s3_table, pid, "COMPLETED") + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 2, f"Expected 2 rows in destination after export, got {count}" + + result = node.query(f"SELECT a, b, c, val FROM {s3_table} ORDER BY val").strip() + assert result == "1\t1\t1\tx\n2\t2\t2\ty", f"Unexpected exported data:\n{result}"