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 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..753095900fa8 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h @@ -248,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 { @@ -290,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); @@ -357,6 +360,16 @@ 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; + /// 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; + } if (json->has("parquet_compression_method")) { diff --git a/src/Storages/MergeTree/ExportPartTask.cpp b/src/Storages/MergeTree/ExportPartTask.cpp index 2305757c895c..5ef537bbef1c 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,46 @@ 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()) + { + 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) + 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..806912407279 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 @@ -164,6 +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); + /// 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))); @@ -648,8 +655,28 @@ namespace ExportPartitionUtils const auto destination_sample_block = destination_metadata->getSampleBlockNonMaterialized(); - const auto source_columns = source_sample_block.getColumnsWithTypeAndName(); - const auto destination_columns = destination_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()) + { + 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, 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/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); + } +} + } 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..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 @@ -614,6 +614,181 @@ 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=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=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) + + 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=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_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=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=node, table=mt, partition_id="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 sourcedest 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