From 267e79f1d24243496762d6a1d82ef42cccd2916b Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Thu, 23 Jul 2026 16:27:13 +0200 Subject: [PATCH 1/6] init Signed-off-by: Konstantin Morozov --- src/Core/Settings.cpp | 6 ++ src/Core/Settings.h | 1 + src/Core/SettingsChangesHistory.cpp | 1 + src/Core/SettingsEnums.cpp | 2 + src/Core/SettingsEnums.h | 8 ++ ...portReplicatedMergeTreePartitionManifest.h | 11 +++ src/Storages/MergeTree/ExportPartTask.cpp | 40 ++++++++- .../MergeTree/ExportPartitionUtils.cpp | 20 ++++- src/Storages/MergeTree/ExportPartitionUtils.h | 7 ++ src/Storages/StorageReplicatedMergeTree.cpp | 2 + .../test.py | 69 +++++++++++++++ .../test.py | 85 +++++++++++++++++++ 12 files changed, 249 insertions(+), 3 deletions(-) diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 0485797420ff..7b671722a5b3 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7611,6 +7611,12 @@ Pattern for the filename of the exported merge tree part. The `part_name` and `c Allow `EXPORT PART`/`EXPORT PARTITION` to apply lossy (non-value-preserving) casts when the source and destination column types differ. When disabled, an export that would require a lossy cast throws instead. When exporting to Apache Iceberg, the partition value written to the metadata is derived from the source partition columns by casting them to the destination partition-field types and applying the destination partition transform — the same computation the exported data files use, so the metadata stays consistent with the data. A lossy cast on a partition column remains semantically truncating: both the data files and the metadata contain the truncated value, and such casts require this setting to be enabled. +)", 0) \ + DECLARE(MergeTreePartExportSchemaMismatchMode, export_merge_tree_part_schema_mismatch_mode, MergeTreePartExportSchemaMismatchMode::strict, R"( +Controls whether `EXPORT PART`/`EXPORT PARTITION` allows a column-count mismatch between the source `MergeTree` table and the destination table. Columns are matched positionally, like `INSERT INTO dest SELECT * FROM src`. +Possible values: +- `strict` (default) - the source and destination must have the same number of columns. A mismatch in either direction throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. +- `ignore_extra_source_columns_by_position` - the source may have more columns than the destination. The extra trailing source columns (by position) are dropped and not exported. The destination having more columns than the source is still rejected in this mode. )", 0) \ \ /* ####################################################### */ \ diff --git a/src/Core/Settings.h b/src/Core/Settings.h index 312d7a1f706f..a5fef6160201 100644 --- a/src/Core/Settings.h +++ b/src/Core/Settings.h @@ -85,6 +85,7 @@ class WriteBuffer; M(CLASS_NAME, Map) \ M(CLASS_NAME, MaxThreads) \ M(CLASS_NAME, MergeTreePartExportFileAlreadyExistsPolicy) \ + M(CLASS_NAME, MergeTreePartExportSchemaMismatchMode) \ M(CLASS_NAME, Milliseconds) \ M(CLASS_NAME, MsgPackUUIDRepresentation) \ M(CLASS_NAME, MySQLDataTypesSupport) \ diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index ff4dea11f189..a7825676c628 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -44,6 +44,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() {"object_storage_cluster_join_mode", "allow", "allow", "New setting"}, {"export_merge_tree_partition_task_timeout_seconds", "3600", "86400", "Increase default value to make it more realistic"}, {"export_merge_tree_part_allow_lossy_cast", false, false, "New setting to gate lossy casts in EXPORT PART/PARTITION behind explicit acknowledgment"}, + {"export_merge_tree_part_schema_mismatch_mode", "strict", "strict", "New setting to allow EXPORT PART/EXPORT PARTITION when the source table has more columns than the destination"}, {"export_merge_tree_partition_retry_initial_backoff_seconds", 5, 5, "New setting for exponential back-off between failed part export retries in an export partition task"}, {"export_merge_tree_partition_retry_max_backoff_seconds", 300, 300, "New setting capping the exponential back-off between failed part export retries in an export partition task"}, {"export_merge_tree_partition_max_retries", 3, 3, "Obsolete and ignored: export partition tasks now retry retryable failures until the task timeout and fail immediately on non-retryable errors, instead of using a fixed retry budget"}, diff --git a/src/Core/SettingsEnums.cpp b/src/Core/SettingsEnums.cpp index bb5a0ea360dd..9aadee2db780 100644 --- a/src/Core/SettingsEnums.cpp +++ b/src/Core/SettingsEnums.cpp @@ -483,6 +483,8 @@ IMPLEMENT_SETTING_ENUM(JemallocProfileFormat, ErrorCodes::BAD_ARGUMENTS, IMPLEMENT_SETTING_AUTO_ENUM(MergeTreePartExportFileAlreadyExistsPolicy, ErrorCodes::BAD_ARGUMENTS); +IMPLEMENT_SETTING_AUTO_ENUM(MergeTreePartExportSchemaMismatchMode, ErrorCodes::BAD_ARGUMENTS); + IMPLEMENT_SETTING_AUTO_ENUM(ExportPartitionAllOnError, ErrorCodes::BAD_ARGUMENTS); } diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index 1c73eee4c97c..0c82d40345cd 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -574,6 +574,14 @@ enum class MergeTreePartExportFileAlreadyExistsPolicy : uint8_t DECLARE_SETTING_ENUM(MergeTreePartExportFileAlreadyExistsPolicy) +enum class MergeTreePartExportSchemaMismatchMode : uint8_t +{ + strict, + ignore_extra_source_columns_by_position, +}; + +DECLARE_SETTING_ENUM(MergeTreePartExportSchemaMismatchMode) + enum class ExportPartitionAllOnError : uint8_t { throw_first, diff --git a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h index d58e9d534949..7f2d5cf7b302 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h @@ -241,6 +241,7 @@ struct ExportReplicatedMergeTreePartitionManifest String filename_pattern; bool write_full_path_in_iceberg_metadata = false; bool allow_lossy_cast = false; + MergeTreePartExportSchemaMismatchMode schema_mismatch_mode = MergeTreePartExportSchemaMismatchMode::strict; String iceberg_metadata_json; /// Optional because of backwards compatibility @@ -282,6 +283,7 @@ struct ExportReplicatedMergeTreePartitionManifest json.set("task_timeout_seconds", task_timeout_seconds); json.set("write_full_path_in_iceberg_metadata", write_full_path_in_iceberg_metadata); json.set("allow_lossy_cast", allow_lossy_cast); + json.set("schema_mismatch_mode", String(magic_enum::enum_name(schema_mismatch_mode))); if (parquet_compression_method) json.set("parquet_compression_method", *parquet_compression_method); if (output_format_compression_level) @@ -357,6 +359,15 @@ struct ExportReplicatedMergeTreePartitionManifest /// on upgrade. New tasks always persist the initiator's actual choice. manifest.allow_lossy_cast = json->has("allow_lossy_cast") ? json->getValue("allow_lossy_cast") : true; + /// Tasks created before this field existed were always scheduled under the old, + /// strict column-count check (a mismatch could never reach scheduling in the first + /// place), so the struct's default of `strict` is the correct fallback here. + if (json->has("schema_mismatch_mode")) + { + const auto schema_mismatch_mode = magic_enum::enum_cast(json->getValue("schema_mismatch_mode")); + if (schema_mismatch_mode) + manifest.schema_mismatch_mode = schema_mismatch_mode.value(); + } if (json->has("parquet_compression_method")) { diff --git a/src/Storages/MergeTree/ExportPartTask.cpp b/src/Storages/MergeTree/ExportPartTask.cpp index 2305757c895c..84d7bc34c72d 100644 --- a/src/Storages/MergeTree/ExportPartTask.cpp +++ b/src/Storages/MergeTree/ExportPartTask.cpp @@ -65,6 +65,7 @@ namespace Setting extern const SettingsUInt64 export_merge_tree_part_max_rows_per_file; extern const SettingsBool allow_experimental_analyzer; extern const SettingsString export_merge_tree_part_filename_pattern; + extern const SettingsMergeTreePartExportSchemaMismatchMode export_merge_tree_part_schema_mismatch_mode; } namespace @@ -116,6 +117,11 @@ namespace /// destination header = `getSampleBlockNonMaterialized()`, all type bridging is done /// by the CAST inside `makeConvertingActions`. No pre-validation, no per-column /// lossy/non-lossy classification — restrictions are exactly what INSERT SELECT enforces. + /// + /// Exception: when `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` + /// and the source has more columns than the destination, the extra trailing source + /// columns (by position) are dropped by a preliminary projection step before the + /// positional convert, so `makeConvertingActions` always sees equal-sized inputs. void addExportConvertingActions( QueryPlan & plan_for_part, const IStorage & destination_storage, @@ -123,10 +129,40 @@ namespace { const auto destination_header = destination_storage.getInMemoryMetadataPtr()->getSampleBlockNonMaterialized(); + const auto destination_columns = destination_header.getColumnsWithTypeAndName(); + + const bool ignore_extra_source_columns_by_position = + local_context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode] + == MergeTreePartExportSchemaMismatchMode::ignore_extra_source_columns_by_position; + + auto source_columns = plan_for_part.getCurrentHeader()->getColumnsWithTypeAndName(); + + if (ignore_extra_source_columns_by_position && source_columns.size() > destination_columns.size()) + { + Names kept_names; + kept_names.reserve(destination_columns.size()); + for (size_t i = 0; i < destination_columns.size(); ++i) + kept_names.push_back(source_columns[i].name); + + /// `allow_remove_inputs = false` keeps the dropped columns registered as DAG + /// inputs (just not as outputs), so `ExpressionActions::execute` still + /// recognizes and consumes them from the block instead of passing them through + /// unchanged. See the `defaults_dag` merge above for the same pattern. + ActionsDAG trim_dag(source_columns); + trim_dag.removeUnusedActions(kept_names, false); + + auto trim_step = std::make_unique( + plan_for_part.getCurrentHeader(), + std::move(trim_dag)); + trim_step->setStepDescription("Drop source columns beyond destination schema for export"); + plan_for_part.addStep(std::move(trim_step)); + + source_columns = plan_for_part.getCurrentHeader()->getColumnsWithTypeAndName(); + } auto dag = ActionsDAG::makeConvertingActions( - plan_for_part.getCurrentHeader()->getColumnsWithTypeAndName(), - destination_header.getColumnsWithTypeAndName(), + source_columns, + destination_columns, ActionsDAG::MatchColumnsMode::Position, local_context); diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 3f85ebc0e1fb..4da6554239c2 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -75,6 +75,7 @@ namespace ErrorCodes namespace Setting { extern const SettingsBool export_merge_tree_part_allow_lossy_cast; + extern const SettingsMergeTreePartExportSchemaMismatchMode export_merge_tree_part_schema_mismatch_mode; } namespace FailPoints @@ -191,6 +192,11 @@ namespace ExportPartitionUtils /// schema drifts to a lossy target between scheduling and execution. context_copy->setSetting("export_merge_tree_part_allow_lossy_cast", manifest.allow_lossy_cast); + /// Same reasoning as above, for the schema column-count mismatch mode: the worker + /// re-runs `verifyExportSchemaCastable` per part (via `exportPartToTable`) and must + /// honor the initiator's choice regardless of which replica picks up the task. + context_copy->setSetting("export_merge_tree_part_schema_mismatch_mode", String(magic_enum::enum_name(manifest.schema_mismatch_mode))); + return context_copy; } @@ -648,9 +654,21 @@ namespace ExportPartitionUtils const auto destination_sample_block = destination_metadata->getSampleBlockNonMaterialized(); - const auto source_columns = source_sample_block.getColumnsWithTypeAndName(); + auto source_columns = source_sample_block.getColumnsWithTypeAndName(); const auto destination_columns = destination_sample_block.getColumnsWithTypeAndName(); + /// In `ignore_extra_source_columns_by_position` mode a source with more columns than the destination + /// is allowed: the extra trailing source columns (by position) are dropped, mirroring + /// the trimming `ExportPartTask::addExportConvertingActions` applies to the real data. + /// The reverse (destination has more columns than source) is always rejected below by + /// `makeConvertingActions`, in both modes. + const bool ignore_extra_source_columns_by_position = + context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode] + == MergeTreePartExportSchemaMismatchMode::ignore_extra_source_columns_by_position; + + if (ignore_extra_source_columns_by_position && source_columns.size() > destination_columns.size()) + source_columns.resize(destination_columns.size()); + (void) ActionsDAG::makeConvertingActions( source_columns, destination_columns, diff --git a/src/Storages/MergeTree/ExportPartitionUtils.h b/src/Storages/MergeTree/ExportPartitionUtils.h index 0bb8acb9bda4..6988621b093a 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.h +++ b/src/Storages/MergeTree/ExportPartitionUtils.h @@ -92,6 +92,13 @@ namespace ExportPartitionUtils /// 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. + /// + /// By default the source and destination must have the same number of columns. + /// If `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'`, a + /// source with more columns than the destination is allowed: the extra trailing + /// source columns (by position) are excluded from the comparison here, matching + /// what `ExportPartTask::addExportConvertingActions` drops from the actual data. + /// /// Throws BAD_ARGUMENTS on any violation. void verifyExportSchemaCastable( const StorageMetadataPtr & source_metadata, diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index 1623611fec1b..387c646aa926 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -229,6 +229,7 @@ namespace Setting extern const SettingsBool export_merge_tree_part_throw_on_pending_mutations; extern const SettingsBool export_merge_tree_part_throw_on_pending_patch_parts; extern const SettingsBool export_merge_tree_part_allow_lossy_cast; + extern const SettingsMergeTreePartExportSchemaMismatchMode export_merge_tree_part_schema_mismatch_mode; extern const SettingsExportPartitionAllOnError export_merge_tree_partition_all_on_error; extern const SettingsString export_merge_tree_part_filename_pattern; extern const SettingsBool write_full_path_in_iceberg_metadata; @@ -8530,6 +8531,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.schema_mismatch_mode = query_context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode].value; if (dest_storage->isDataLake()) { 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..3bcb6f4c738b 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 @@ -614,6 +614,75 @@ def test_export_part_column_count_mismatch_source_fewer_is_rejected(cluster): node.query(f"DROP TABLE IF EXISTS {iceberg}") +def test_export_part_source_more_columns_allowed_with_ignore_extra_setting(cluster): + """ + Source has 3 columns (id, year, extra), destination has 2 (id, year). + With `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'`, + the export must succeed: the trailing `extra` source column is dropped + (matched positionally) and only `id`/`year` land in the destination. + """ + node = cluster.instances["node1"] + sfx = unique_suffix() + mt = f"mt_ignore_extra_{sfx}" + iceberg = f"iceberg_ignore_extra_{sfx}" + + make_mt(node, mt, "id Int32, year Int32, extra String", "year") + make_iceberg_s3(node, iceberg, "id Int32, year Int32", "year") + + node.query(f"INSERT INTO {mt} VALUES (1, 2020, 'foo'), (2, 2020, 'bar'), (3, 2020, 'baz')") + part_2020 = get_part(node, mt, "2020") + + export_part( + node, mt, part_2020, iceberg, + extra_settings="export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'", + ) + wait_for_export_part(node, mt, part_2020) + + count = int(node.query(f"SELECT count() FROM {iceberg}").strip()) + assert count == 3, f"Expected 3 rows in Iceberg table after export, got {count}" + + result = node.query(f"SELECT id, year FROM {iceberg} ORDER BY id").strip() + assert result == "1\t2020\n2\t2020\n3\t2020", f"Unexpected data:\n{result}" + + assert_part_log(node, mt, part_2020) + + node.query(f"DROP TABLE IF EXISTS {mt} SYNC") + node.query(f"DROP TABLE IF EXISTS {iceberg}") + + +def test_export_part_column_count_mismatch_source_fewer_still_rejected_with_ignore_extra_setting(cluster): + """ + `ignore_extra_source_columns_by_position` only relaxes the source-has-more-columns + direction. Source has 2 columns (id, year), destination has 3 (id, year, extra): + the destination cannot be filled from the source, so this must still be + rejected synchronously even with the relaxed setting. + """ + node = cluster.instances["node1"] + sfx = unique_suffix() + mt = f"mt_ignore_extra_fewer_{sfx}" + iceberg = f"iceberg_ignore_extra_fewer_{sfx}" + + make_mt(node, mt, "id Int32, year Int32", "year") + make_iceberg_s3(node, iceberg, "id Int32, year Int32, extra String", "year") + + node.query(f"INSERT INTO {mt} VALUES (1, 2020), (2, 2020)") + part_2020 = get_part(node, mt, "2020") + + error = node.query_and_get_error( + f"ALTER TABLE {mt} EXPORT PART '{part_2020}' TO TABLE {iceberg} " + f"SETTINGS allow_experimental_export_merge_tree_part = 1, " + f"allow_experimental_insert_into_iceberg = 1, " + f"export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'" + ) + assert "NUMBER_OF_COLUMNS_DOESNT_MATCH" in error, ( + f"Expected NUMBER_OF_COLUMNS_DOESNT_MATCH for source Date: Fri, 24 Jul 2026 12:34:06 +0200 Subject: [PATCH 2/6] update export Signed-off-by: Konstantin Morozov --- ...ExportReplicatedMergeTreePartitionManifest.h | 14 ++++++++------ src/Storages/MergeTree/ExportPartTask.cpp | 8 +++++++- src/Storages/MergeTree/ExportPartitionUtils.cpp | 17 +++++++++++------ 3 files changed, 26 insertions(+), 13 deletions(-) diff --git a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h index 7f2d5cf7b302..753095900fa8 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h @@ -241,7 +241,6 @@ struct ExportReplicatedMergeTreePartitionManifest String filename_pattern; bool write_full_path_in_iceberg_metadata = false; bool allow_lossy_cast = false; - MergeTreePartExportSchemaMismatchMode schema_mismatch_mode = MergeTreePartExportSchemaMismatchMode::strict; String iceberg_metadata_json; /// Optional because of backwards compatibility @@ -249,6 +248,7 @@ struct ExportReplicatedMergeTreePartitionManifest std::optional output_format_compression_level; std::optional parquet_row_group_size; std::optional parquet_row_group_size_bytes; + std::optional schema_mismatch_mode; std::string toJsonString() const { @@ -283,7 +283,6 @@ struct ExportReplicatedMergeTreePartitionManifest json.set("task_timeout_seconds", task_timeout_seconds); json.set("write_full_path_in_iceberg_metadata", write_full_path_in_iceberg_metadata); json.set("allow_lossy_cast", allow_lossy_cast); - json.set("schema_mismatch_mode", String(magic_enum::enum_name(schema_mismatch_mode))); if (parquet_compression_method) json.set("parquet_compression_method", *parquet_compression_method); if (output_format_compression_level) @@ -292,6 +291,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 (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 oss.exceptions(std::ios::failbit); Poco::JSON::Stringifier::stringify(json, oss); @@ -359,14 +360,15 @@ struct ExportReplicatedMergeTreePartitionManifest /// on upgrade. New tasks always persist the initiator's actual choice. manifest.allow_lossy_cast = json->has("allow_lossy_cast") ? json->getValue("allow_lossy_cast") : true; - /// Tasks created before this field existed were always scheduled under the old, - /// strict column-count check (a mismatch could never reach scheduling in the first - /// place), so the struct's default of `strict` is the correct fallback here. + /// Left unset (nullopt) for tasks created before this field existed - such tasks were + /// always scheduled under the old, strict column-count check (a mismatch could never + /// reach scheduling in the first place), so callers should treat an absent value as + /// `strict`. if (json->has("schema_mismatch_mode")) { const auto schema_mismatch_mode = magic_enum::enum_cast(json->getValue("schema_mismatch_mode")); if (schema_mismatch_mode) - manifest.schema_mismatch_mode = schema_mismatch_mode.value(); + manifest.schema_mismatch_mode = schema_mismatch_mode; } if (json->has("parquet_compression_method")) diff --git a/src/Storages/MergeTree/ExportPartTask.cpp b/src/Storages/MergeTree/ExportPartTask.cpp index 84d7bc34c72d..5ef537bbef1c 100644 --- a/src/Storages/MergeTree/ExportPartTask.cpp +++ b/src/Storages/MergeTree/ExportPartTask.cpp @@ -129,7 +129,7 @@ namespace { const auto destination_header = destination_storage.getInMemoryMetadataPtr()->getSampleBlockNonMaterialized(); - const auto destination_columns = destination_header.getColumnsWithTypeAndName(); + const auto & destination_columns = destination_header.getColumnsWithTypeAndName(); const bool ignore_extra_source_columns_by_position = local_context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode] @@ -139,6 +139,12 @@ namespace if (ignore_extra_source_columns_by_position && source_columns.size() > destination_columns.size()) { + LOG_DEBUG(getLogger("ExportPartTask"), + "Source has {} columns while destination has {} columns, " + "the {} extra trailing source column(s) will be ignored", + source_columns.size(), destination_columns.size(), + source_columns.size() - destination_columns.size()); + Names kept_names; kept_names.reserve(destination_columns.size()); for (size_t i = 0; i < destination_columns.size(); ++i) diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 4da6554239c2..07da0eebb60e 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -165,6 +165,8 @@ namespace ExportPartitionUtils context_copy->setSetting("output_format_parquet_row_group_size", *manifest.parquet_row_group_size); if (manifest.parquet_row_group_size_bytes) context_copy->setSetting("output_format_parquet_row_group_size_bytes", *manifest.parquet_row_group_size_bytes); + if (manifest.schema_mismatch_mode) + context_copy->setSetting("export_merge_tree_part_schema_mismatch_mode", String(magic_enum::enum_name(*manifest.schema_mismatch_mode))); context_copy->setSetting("max_threads", manifest.max_threads); context_copy->setSetting("export_merge_tree_part_file_already_exists_policy", String(magic_enum::enum_name(manifest.file_already_exists_policy))); @@ -192,11 +194,6 @@ namespace ExportPartitionUtils /// schema drifts to a lossy target between scheduling and execution. context_copy->setSetting("export_merge_tree_part_allow_lossy_cast", manifest.allow_lossy_cast); - /// Same reasoning as above, for the schema column-count mismatch mode: the worker - /// re-runs `verifyExportSchemaCastable` per part (via `exportPartToTable`) and must - /// honor the initiator's choice regardless of which replica picks up the task. - context_copy->setSetting("export_merge_tree_part_schema_mismatch_mode", String(magic_enum::enum_name(manifest.schema_mismatch_mode))); - return context_copy; } @@ -655,7 +652,7 @@ namespace ExportPartitionUtils const auto destination_sample_block = destination_metadata->getSampleBlockNonMaterialized(); auto source_columns = source_sample_block.getColumnsWithTypeAndName(); - const auto destination_columns = destination_sample_block.getColumnsWithTypeAndName(); + const auto & destination_columns = destination_sample_block.getColumnsWithTypeAndName(); /// In `ignore_extra_source_columns_by_position` mode a source with more columns than the destination /// is allowed: the extra trailing source columns (by position) are dropped, mirroring @@ -667,7 +664,15 @@ namespace ExportPartitionUtils == MergeTreePartExportSchemaMismatchMode::ignore_extra_source_columns_by_position; if (ignore_extra_source_columns_by_position && source_columns.size() > destination_columns.size()) + { + LOG_DEBUG(getLogger("ExportPartitionUtils"), + "Source has {} columns while destination has {} columns, " + "the {} extra trailing source column(s) will be ignored", + source_columns.size(), destination_columns.size(), + source_columns.size() - destination_columns.size()); + source_columns.resize(destination_columns.size()); + } (void) ActionsDAG::makeConvertingActions( source_columns, From 39869c26eb85e980b390bd5e213e47257e38dd66 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Fri, 24 Jul 2026 12:34:42 +0200 Subject: [PATCH 3/6] add unit tests for backward cap Signed-off-by: Konstantin Morozov --- .../tests/gtest_export_partition_ordering.cpp | 99 +++++++++++++++++++ 1 file changed, 99 insertions(+) diff --git a/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp b/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp index b3c750129988..df20755ba590 100644 --- a/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp +++ b/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp @@ -1,9 +1,46 @@ #include +#include #include +#include +#include +#include +#include +#include +#include namespace DB { +namespace Setting +{ + extern const SettingsMergeTreePartExportSchemaMismatchMode export_merge_tree_part_schema_mismatch_mode; +} + +namespace +{ + ExportReplicatedMergeTreePartitionManifest makeValidManifest() + { + ExportReplicatedMergeTreePartitionManifest manifest; + manifest.transaction_id = "tx1"; + manifest.query_id = "query1"; + manifest.partition_id = "2020"; + manifest.destination_database = "db1"; + manifest.destination_table = "table1"; + manifest.source_replica = "r1"; + manifest.number_of_parts = 1; + manifest.create_time = 1000; + manifest.task_timeout_seconds = 60; + manifest.max_threads = 1; + manifest.parallel_formatting = true; + manifest.parquet_parallel_encoding = true; + manifest.max_bytes_per_file = 1000000; + manifest.max_rows_per_file = 1000; + manifest.file_already_exists_policy = MergeTreePartExportManifest::FileAlreadyExistsPolicy::error; + manifest.filename_pattern = "{part_name}"; + return manifest; + } +} + class ExportPartitionOrderingTest : public ::testing::Test { protected: @@ -18,6 +55,10 @@ class ExportPartitionOrderingTest : public ::testing::Test } }; +class ExportPartitionManifestBackCompatTest : public ::testing::Test +{ +}; + TEST_F(ExportPartitionOrderingTest, IterationOrderMatchesCreateTime) { time_t base_time = 1000; @@ -72,4 +113,62 @@ TEST_F(ExportPartitionOrderingTest, IterationOrderMatchesCreateTime) EXPECT_EQ(it, by_create_time.end()); } + +TEST_F(ExportPartitionManifestBackCompatTest, MissingSchemaMismatchModeParsesAsNullopt) +{ + auto manifest = makeValidManifest(); + manifest.schema_mismatch_mode = MergeTreePartExportSchemaMismatchMode::ignore_extra_source_columns_by_position; + + Poco::JSON::Parser parser; + auto json = parser.parse(manifest.toJsonString()).extract(); + json->remove("schema_mismatch_mode"); + std::ostringstream oss; + oss.exceptions(std::ios::failbit); + Poco::JSON::Stringifier::stringify(json, oss); + + auto parsed = ExportReplicatedMergeTreePartitionManifest::fromJsonString(oss.str()); + EXPECT_FALSE(parsed.schema_mismatch_mode.has_value()); +} + +TEST_F(ExportPartitionManifestBackCompatTest, SchemaMismatchModeRoundTripsForEveryValue) +{ + for (const auto value : magic_enum::enum_values()) + { + auto manifest = makeValidManifest(); + manifest.schema_mismatch_mode = value; + + auto parsed = ExportReplicatedMergeTreePartitionManifest::fromJsonString(manifest.toJsonString()); + + ASSERT_TRUE(parsed.schema_mismatch_mode.has_value()) << "value=" << magic_enum::enum_name(value); + EXPECT_EQ(*parsed.schema_mismatch_mode, value) << "value=" << magic_enum::enum_name(value); + } +} + +TEST_F(ExportPartitionManifestBackCompatTest, MissingSchemaMismatchModeFallsBackToStrictInWorkerContext) +{ + auto manifest = makeValidManifest(); + ASSERT_FALSE(manifest.schema_mismatch_mode.has_value()); + + auto worker_context = ExportPartitionUtils::getContextCopyWithTaskSettings(getContext().context, manifest); + + EXPECT_EQ( + worker_context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode].value, + MergeTreePartExportSchemaMismatchMode::strict); +} + +TEST_F(ExportPartitionManifestBackCompatTest, SchemaMismatchModeAppliedToWorkerContextForEveryValue) +{ + for (const auto value : magic_enum::enum_values()) + { + auto manifest = makeValidManifest(); + manifest.schema_mismatch_mode = value; + + auto worker_context = ExportPartitionUtils::getContextCopyWithTaskSettings(getContext().context, manifest); + + EXPECT_EQ( + worker_context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode].value, + value) << "value=" << magic_enum::enum_name(value); + } +} + } From d4696464216dbfa99e82f9ded5e370ec7374ae2f Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Fri, 24 Jul 2026 12:34:55 +0200 Subject: [PATCH 4/6] add doc Signed-off-by: Konstantin Morozov --- docs/en/antalya/part_export.md | 12 +++++++++++- docs/en/antalya/partition_export.md | 10 ++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/docs/en/antalya/part_export.md b/docs/en/antalya/part_export.md index 73d467c5d9b1..45a81df6dcbd 100644 --- a/docs/en/antalya/part_export.md +++ b/docs/en/antalya/part_export.md @@ -49,7 +49,7 @@ SETTINGS allow_experimental_export_merge_tree_part = 1 Source and destination tables must be 100% compatible: -1. **Identical schemas** - same columns, types, and order +1. **Identical schemas** - same columns, types, and order, matched positionally like `INSERT INTO dest SELECT * FROM src`. By default the source and destination must also have the same number of columns; a source with extra trailing columns can be allowed via `export_merge_tree_part_schema_mismatch_mode`. 2. **Matching partition keys** - partition expressions must be identical In case a table function is used as the destination, the schema can be omitted and it will be inferred from the source table. @@ -117,6 +117,16 @@ In case a table function is used as the destination, the schema can be omitted a **Warning:** A lossy cast on a partition column remains semantically truncating. For example, if a table is partitioned by an `Int64` column and some partition values do not fit into a destination `Int32` partition column, both the data files and the Iceberg metadata will contain the truncated `Int32` value (they agree with each other, but the original `Int64` value is lost). Such casts require `export_merge_tree_part_allow_lossy_cast = 1`. +### `export_merge_tree_part_schema_mismatch_mode` (Optional) + +- **Type**: `MergeTreePartExportSchemaMismatchMode` +- **Default**: `strict` +- **Description**: Controls whether `EXPORT PART`/`EXPORT PARTITION` allows a column-count mismatch between the source `MergeTree` table and the destination table. Columns are matched positionally, like `INSERT INTO dest SELECT * FROM src`. Possible values: + - `strict` - the source and destination must have the same number of columns. A mismatch in either direction throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. + - `ignore_extra_source_columns_by_position` - the source may have more columns than the destination. The extra trailing source columns (by position) are dropped and not exported. The destination having more columns than the source is still rejected in this mode. + + The extra trailing source columns are still read and evaluated (including `MATERIALIZED`/`ALIAS` columns, and any column another kept column's `ALIAS`/`MATERIALIZED` expression depends on) before being dropped, so this setting only changes which columns end up in the destination, not what is computed while reading the part. + ## Examples diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index 687029b9adc6..36b532db7e16 100644 --- a/docs/en/antalya/partition_export.md +++ b/docs/en/antalya/partition_export.md @@ -120,6 +120,16 @@ Notes: **Warning:** A lossy cast on a partition column remains semantically truncating. For example, if a table is partitioned by an `Int64` column and some partition values do not fit into a destination `Int32` partition column, both the data files and the Iceberg metadata will contain the truncated `Int32` value (they agree with each other, but the original `Int64` value is lost). Such casts require `export_merge_tree_part_allow_lossy_cast = 1`. +### `export_merge_tree_part_schema_mismatch_mode` (Optional) + +- **Type**: `MergeTreePartExportSchemaMismatchMode` +- **Default**: `strict` +- **Description**: Controls whether `EXPORT PART`/`EXPORT PARTITION` allows a column-count mismatch between the source `MergeTree` table and the destination table. Columns are matched positionally, like `INSERT INTO dest SELECT * FROM src`. Possible values: + - `strict` - the source and destination must have the same number of columns. A mismatch in either direction throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. + - `ignore_extra_source_columns_by_position` - the source may have more columns than the destination. The extra trailing source columns (by position) are dropped and not exported. The destination having more columns than the source is still rejected in this mode. + + The extra trailing source columns are still read and evaluated (including `MATERIALIZED`/`ALIAS` columns, and any column another kept column's `ALIAS`/`MATERIALIZED` expression depends on) before being dropped, so this setting only changes which columns end up in the destination, not what is computed while reading the part. + ## Examples ### Basic Export to S3 From 52ec1e4f137f44f76a2f954ace0e9831cee6f0a9 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Fri, 24 Jul 2026 15:55:31 +0200 Subject: [PATCH 5/6] add integration tests Signed-off-by: Konstantin Morozov --- .../test.py | 124 +++++- .../test.py | 116 ++++++ .../test.py | 361 +++++++++++++++++- .../test.py | 38 ++ 4 files changed, 624 insertions(+), 15 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 3bcb6f4c738b..d60912ddbf88 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 @@ -626,17 +626,17 @@ def test_export_part_source_more_columns_allowed_with_ignore_extra_setting(clust mt = f"mt_ignore_extra_{sfx}" iceberg = f"iceberg_ignore_extra_{sfx}" - make_mt(node, mt, "id Int32, year Int32, extra String", "year") - make_iceberg_s3(node, iceberg, "id Int32, year Int32", "year") + make_mt(node=node, name=mt, columns="id Int32, year Int32, extra String", partition_by="year") + make_iceberg_s3(node=node, name=iceberg, columns="id Int32, year Int32", partition_by="year") node.query(f"INSERT INTO {mt} VALUES (1, 2020, 'foo'), (2, 2020, 'bar'), (3, 2020, 'baz')") - part_2020 = get_part(node, mt, "2020") + part_2020 = get_part(node=node, table=mt, partition_id="2020") export_part( - node, mt, part_2020, iceberg, + node=node, table=mt, part=part_2020, dest=iceberg, extra_settings="export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'", ) - wait_for_export_part(node, mt, part_2020) + wait_for_export_part(node=node, table=mt, part=part_2020) count = int(node.query(f"SELECT count() FROM {iceberg}").strip()) assert count == 3, f"Expected 3 rows in Iceberg table after export, got {count}" @@ -644,7 +644,7 @@ def test_export_part_source_more_columns_allowed_with_ignore_extra_setting(clust result = node.query(f"SELECT id, year FROM {iceberg} ORDER BY id").strip() assert result == "1\t2020\n2\t2020\n3\t2020", f"Unexpected data:\n{result}" - assert_part_log(node, mt, part_2020) + assert_part_log(node=node, table=mt, part=part_2020) node.query(f"DROP TABLE IF EXISTS {mt} SYNC") node.query(f"DROP TABLE IF EXISTS {iceberg}") @@ -662,11 +662,11 @@ def test_export_part_column_count_mismatch_source_fewer_still_rejected_with_igno mt = f"mt_ignore_extra_fewer_{sfx}" iceberg = f"iceberg_ignore_extra_fewer_{sfx}" - make_mt(node, mt, "id Int32, year Int32", "year") - make_iceberg_s3(node, iceberg, "id Int32, year Int32, extra String", "year") + make_mt(node=node, name=mt, columns="id Int32, year Int32", partition_by="year") + make_iceberg_s3(node=node, name=iceberg, columns="id Int32, year Int32, extra String", partition_by="year") node.query(f"INSERT INTO {mt} VALUES (1, 2020), (2, 2020)") - part_2020 = get_part(node, mt, "2020") + part_2020 = get_part(node=node, table=mt, partition_id="2020") error = node.query_and_get_error( f"ALTER TABLE {mt} EXPORT PART '{part_2020}' TO TABLE {iceberg} " @@ -683,6 +683,112 @@ def test_export_part_column_count_mismatch_source_fewer_still_rejected_with_igno node.query(f"DROP TABLE IF EXISTS {iceberg}") +def test_export_part_with_materialized_column(cluster): + node = cluster.instances["node1"] + sfx = unique_suffix() + mt = f"mt_materialized_{sfx}" + iceberg = f"iceberg_materialized_{sfx}" + + make_mt(node=node, name=mt, columns="id Int32, year Int32, doubled Int32 MATERIALIZED id * 2", partition_by="year") + make_iceberg_s3(node=node, name=iceberg, columns="id Int32, year Int32, doubled Int32", partition_by="year") + + node.query(f"INSERT INTO {mt} (id, year) VALUES (1, 2020), (2, 2020)") + part_2020 = get_part(node=node, table=mt, partition_id="2020") + + export_part(node=node, table=mt, part=part_2020, dest=iceberg) + wait_for_export_part(node=node, table=mt, part=part_2020) + + result = node.query(f"SELECT id, year, doubled FROM {iceberg} ORDER BY id").strip() + assert result == "1\t2020\t2\n2\t2020\t4", f"Unexpected data:\n{result}" + + assert_part_log(node=node, table=mt, part=part_2020) + + node.query(f"DROP TABLE IF EXISTS {mt} SYNC") + node.query(f"DROP TABLE IF EXISTS {iceberg}") + + +def test_export_part_with_alias_column(cluster): + node = cluster.instances["node1"] + sfx = unique_suffix() + mt = f"mt_alias_{sfx}" + iceberg = f"iceberg_alias_{sfx}" + + make_mt(node=node, name=mt, columns="id Int32, year Int32, id_alias Int32 ALIAS id", partition_by="year") + make_iceberg_s3(node=node, name=iceberg, columns="id Int32, year Int32, id_alias Int32", partition_by="year") + + node.query(f"INSERT INTO {mt} (id, year) VALUES (1, 2020), (2, 2020)") + part_2020 = get_part(node=node, table=mt, partition_id="2020") + + export_part(node=node, table=mt, part=part_2020, dest=iceberg) + wait_for_export_part(node=node, table=mt, part=part_2020) + + result = node.query(f"SELECT id, year, id_alias FROM {iceberg} ORDER BY id").strip() + assert result == "1\t2020\t1\n2\t2020\t2", f"Unexpected data:\n{result}" + + assert_part_log(node=node, table=mt, part=part_2020) + + node.query(f"DROP TABLE IF EXISTS {mt} SYNC") + node.query(f"DROP TABLE IF EXISTS {iceberg}") + + +def test_export_part_ignore_extra_setting_drops_trailing_alias_column(cluster): + node = cluster.instances["node1"] + sfx = unique_suffix() + mt = f"mt_ignore_extra_alias_{sfx}" + iceberg = f"iceberg_ignore_extra_alias_{sfx}" + + make_mt(node=node, name=mt, columns="id Int32, year Int32, extra_alias String ALIAS toString(id)", partition_by="year") + make_iceberg_s3(node=node, name=iceberg, columns="id Int32, year Int32", partition_by="year") + + node.query(f"INSERT INTO {mt} (id, year) VALUES (1, 2020), (2, 2020)") + part_2020 = get_part(node=node, table=mt, partition_id="2020") + + export_part( + node=node, table=mt, part=part_2020, dest=iceberg, + extra_settings="export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'", + ) + wait_for_export_part(node=node, table=mt, part=part_2020) + + result = node.query(f"SELECT id, year FROM {iceberg} ORDER BY id").strip() + assert result == "1\t2020\n2\t2020", f"Unexpected data:\n{result}" + + assert_part_log(node=node, table=mt, part=part_2020) + + node.query(f"DROP TABLE IF EXISTS {mt} SYNC") + node.query(f"DROP TABLE IF EXISTS {iceberg}") + + +def test_export_part_ignore_extra_setting_kept_alias_depends_on_dropped_column(cluster): + node = cluster.instances["node1"] + sfx = unique_suffix() + mt = f"mt_ignore_extra_dep_{sfx}" + iceberg = f"iceberg_ignore_extra_dep_{sfx}" + + make_mt( + node=node, name=mt, + columns="id Int32, year Int32, computed_alias Int32 ALIAS extra * 2, extra Int32", + partition_by="year", + ) + make_iceberg_s3(node=node, name=iceberg, columns="id Int32, year Int32, computed_alias Int32", partition_by="year") + + node.query(f"INSERT INTO {mt} (id, year, extra) VALUES (1, 2020, 5), (2, 2020, 7)") + part_2020 = get_part(node=node, table=mt, partition_id="2020") + + export_part( + node=node, table=mt, part=part_2020, dest=iceberg, + extra_settings="export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'", + ) + wait_for_export_part(node=node, table=mt, part=part_2020) + + result = node.query(f"SELECT id, year, computed_alias FROM {iceberg} ORDER BY id").strip() + assert result == "1\t2020\t10\n2\t2020\t14", f"Unexpected data:\n{result}" + + assert_part_log(node=node, table=mt, part=part_2020) + + node.query(f"DROP TABLE IF EXISTS {mt} SYNC") + node.query(f"DROP TABLE IF EXISTS {iceberg}") + + def test_export_part_with_renamed_destination_column(cluster): """ Source has column `id`, destination has the same shape but the column is 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..d412a900f462 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 @@ -39,6 +39,24 @@ def create_s3_table(node, s3_table): node.query(f"CREATE TABLE {s3_table} (id UInt64, year UInt16) ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') PARTITION BY year") +def wait_for_export_part(node, table, part, timeout=60, poll_interval=0.5): + deadline = time.time() + timeout + while time.time() < deadline: + node.query("SYSTEM FLUSH LOGS") + count = node.query( + f"SELECT count() FROM system.part_log " + f"WHERE event_type = 'ExportPart' " + f"AND table = '{table}' " + f"AND part_name = '{part}'" + ).strip() + if count != "0": + return + time.sleep(poll_interval) + raise TimeoutError( + f"ExportPart event for part {part!r} in table {table!r} did not appear within {timeout}s" + ) + + def create_tables_and_insert_data(node, mt_table, s3_table): # enable_block_number_column and enable_block_offset_column are needed for patch parts support node.query(f"CREATE TABLE {mt_table} (id UInt64, year UInt16) ENGINE = MergeTree() PARTITION BY year ORDER BY tuple() SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1") @@ -312,3 +330,101 @@ 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_column_count_mismatch_source_more_is_rejected(cluster): + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"count_more_mt_table_{postfix}" + s3_table = f"count_more_s3_table_{postfix}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, year UInt16, extra String) " + f"ENGINE = MergeTree() PARTITION BY year ORDER BY tuple() " + f"SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, 2020, 'foo'), (2, 2020, 'bar')") + + create_s3_table(node=node, s3_table=s3_table) + + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PART '2020_1_1_0' TO TABLE {s3_table}" + ) + assert "NUMBER_OF_COLUMNS_DOESNT_MATCH" in error, ( + f"Expected NUMBER_OF_COLUMNS_DOESNT_MATCH for source>dest column count, got: {error}" + ) + + count = int(node.query(f"SELECT count() FROM {s3_table}").strip()) + assert count == 0, f"Expected 0 rows in destination table after rejected export, got {count}" + + node.query(f"DROP TABLE {mt_table}") + node.query(f"DROP TABLE {s3_table}") + + +def test_export_part_column_count_mismatch_source_fewer_is_rejected(cluster): + node = cluster.instances["node1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"count_fewer_mt_table_{postfix}" + s3_table = f"count_fewer_s3_table_{postfix}" + + node.query( + f"CREATE TABLE {mt_table} (id UInt64, year UInt16) " + f"ENGINE = MergeTree() PARTITION BY year ORDER BY tuple() " + f"SETTINGS enable_block_number_column = 1, enable_block_offset_column = 1" + ) + node.query(f"INSERT INTO {mt_table} VALUES (1, 2020), (2, 2020)") + + node.query( + f"CREATE TABLE {s3_table} (id UInt64, year UInt16, extra String) " + f"ENGINE = S3(s3_conn, filename='{s3_table}', format=Parquet, partition_strategy='hive') " + f"PARTITION BY year" + ) + + error = node.query_and_get_error( + f"ALTER TABLE {mt_table} EXPORT PART '2020_1_1_0' TO TABLE {s3_table}" + ) + assert "NUMBER_OF_COLUMNS_DOESNT_MATCH" in error, ( + f"Expected NUMBER_OF_COLUMNS_DOESNT_MATCH for source Date: Fri, 24 Jul 2026 16:34:31 +0200 Subject: [PATCH 6/6] set to strict in context Signed-off-by: Konstantin Morozov --- src/Storages/MergeTree/ExportPartitionUtils.cpp | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 07da0eebb60e..806912407279 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -165,8 +165,12 @@ namespace ExportPartitionUtils context_copy->setSetting("output_format_parquet_row_group_size", *manifest.parquet_row_group_size); if (manifest.parquet_row_group_size_bytes) context_copy->setSetting("output_format_parquet_row_group_size_bytes", *manifest.parquet_row_group_size_bytes); - if (manifest.schema_mismatch_mode) - context_copy->setSetting("export_merge_tree_part_schema_mismatch_mode", String(magic_enum::enum_name(*manifest.schema_mismatch_mode))); + /// Manifests written before this setting existed have no value here; such tasks were always + /// scheduled under the old, strict column-count check, so an absent value must resolve to + /// `strict` regardless of the ambient context's setting (which may have since been changed). + context_copy->setSetting( + "export_merge_tree_part_schema_mismatch_mode", + String(magic_enum::enum_name(manifest.schema_mismatch_mode.value_or(MergeTreePartExportSchemaMismatchMode::strict)))); context_copy->setSetting("max_threads", manifest.max_threads); context_copy->setSetting("export_merge_tree_part_file_already_exists_policy", String(magic_enum::enum_name(manifest.file_already_exists_policy)));