diff --git a/docs/en/antalya/part_export.md b/docs/en/antalya/part_export.md index ef7a7e25f5f7..1637bf7d6a8d 100644 --- a/docs/en/antalya/part_export.md +++ b/docs/en/antalya/part_export.md @@ -49,13 +49,13 @@ SETTINGS allow_experimental_export_merge_tree_part = 1 Source and destination tables must support positional schema conversion. The following differences between the two schemas are allowed: -- **Column names** may differ between source and destination for non-partition-key columns - columns are matched by position, similar to `INSERT INTO dest SELECT * FROM src`, not by name. +- **Column names** may differ between source and destination for non-partition-key columns when `export_merge_tree_part_schema_match_mode = 'match_by_position'` (the default) - columns are matched by position, similar to `INSERT INTO dest SELECT * FROM src`, not by name. Set `export_merge_tree_part_schema_match_mode = 'match_by_name'` to match columns by their exact, case-sensitive name instead, allowing destination columns to be declared in a different order than the source. - **Column types** may differ, as long as the source type is safely castable to the destination type. Set `export_merge_tree_part_allow_lossy_cast = 1` to also permit lossy casts. - **`Tuple` element names** may differ if either the source or destination declares the tuple without named elements: an unnamed `Tuple` (e.g. `Tuple(Int32, Int32)`) is matched against the destination by element position and type only, not by name. For example, exporting from `t Tuple(Int32, Int32)` to `t Tuple(x Int32, y Int32)` is allowed as long as element types match positionally. The following must match between source and destination: -1. **Column count** - source and destination must have the same number of columns by default. A mismatch in either direction throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode. +1. **Column count** - by default (`ignore_extra_source_columns = false`) every source column must have a corresponding destination column: with `export_merge_tree_part_schema_match_mode = 'match_by_position'` (the default) the source and destination must have the same number of columns; with `'match_by_name'` they must have the same set of column names. A mismatch throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. Set `ignore_extra_source_columns = 1` to allow a source table with columns that have no corresponding destination column; such columns are dropped and not exported. The destination having a column absent from the source is always rejected, regardless of this setting. 2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must be representable as an Iceberg partition spec and must match the destination partition fields and transforms. 3. **The position of every column backing the partition key** - it is not enough for the `PARTITION BY` expressions to be textually identical: every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. If such a column contains a named `Tuple`, its element names must also be declared in the same order (an unnamed `Tuple` on either side is exempt from this, per the allowance above). This comparison is recursive through nested tuples and through container types such as `Array` and `Map`. @@ -136,15 +136,32 @@ 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) +### `export_merge_tree_part_schema_match_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. +- **Type**: `MergeTreePartExportSchemaMatchMode` +- **Default**: `match_by_position` +- **Description**: Controls how `EXPORT PART`/`EXPORT PARTITION` matches source `MergeTree` columns to destination columns. Possible values: + - `match_by_position` (default) - columns are matched positionally, like `INSERT INTO dest SELECT * FROM src`. Column names are not otherwise considered. + - `match_by_name` - every destination column is matched to a source column with the same exact, case-sensitive name, so destination columns may be declared in a different order than the source. A destination column absent from the source, including when it was renamed, throws `THERE_IS_NO_COLUMN`; there is no positional fallback. - 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. + See `ignore_extra_source_columns` below for how a source column without a corresponding destination column is handled in each mode. + +### `ignore_extra_source_columns` (Optional) + +- **Type**: `Bool` +- **Default**: `false` +- **Description**: Controls whether `EXPORT PART`/`EXPORT PARTITION` tolerates source `MergeTree` columns that have no corresponding destination column. + - `false` (default) - such a source column is rejected: the source and destination must match exactly. With `export_merge_tree_part_schema_match_mode = 'match_by_position'` this means the same number of columns; with `'match_by_name'` this means the same set of column names, so a source table with columns absent from the destination is rejected even if the matched columns would otherwise be compatible. A mismatch throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. + - `true` - a source column without a corresponding destination column is dropped and not exported, instead of throwing. In `match_by_position` mode, this allows a source with extra trailing columns (the destination having more columns than the source is still always rejected). In `match_by_name` mode, this allows source columns whose name has no destination counterpart, which may occur in any position. + + Extra 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. Type conversion and `export_merge_tree_part_allow_lossy_cast` are applied after columns are matched. + + Error behavior: + + - With `ignore_extra_source_columns = false` (default), a source column without a corresponding destination column throws `NUMBER_OF_COLUMNS_DOESNT_MATCH` - in `match_by_position` mode this means any column-count mismatch, in `match_by_name` mode this means the source and destination column-name sets differ. + - In `match_by_position` mode, the destination having more columns than the source always throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`, regardless of `ignore_extra_source_columns`. + - In `match_by_name` mode, a destination column absent from the source (including a renamed one) always throws `THERE_IS_NO_COLUMN`, regardless of `ignore_extra_source_columns`; there is no positional fallback. + - After columns have been matched successfully, a cast rejected by the export type-safety check throws `INCOMPATIBLE_COLUMNS`. ## Examples diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index a3b9e3dfc4f2..17b42b91f66c 100644 --- a/docs/en/antalya/partition_export.md +++ b/docs/en/antalya/partition_export.md @@ -45,9 +45,9 @@ TO TABLE [destination_database.]destination_table ## Requirements -`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Column names may differ (columns are matched by position, not by name), and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following must match: +`EXPORT PARTITION` exports each part via the same mechanism as [`EXPORT PART`](/docs/en/antalya/part_export.md#requirements), so the source and destination tables must satisfy the same compatibility requirements. Columns are matched by position by default, or by their exact, case-sensitive name if `export_merge_tree_part_schema_match_mode = 'match_by_name'` is set, and column types may differ as long as they are safely castable (or `export_merge_tree_part_allow_lossy_cast = 1` is set). Beyond that, the following must match: -1. **Column count** - source and destination must have the same number of columns by default. Set `export_merge_tree_part_schema_mismatch_mode = 'ignore_extra_source_columns_by_position'` to allow a source table with extra trailing columns; the destination having more columns than the source is still rejected in this mode. +1. **Column count** - by default (`ignore_extra_source_columns = false`) every source column must have a corresponding destination column: with `export_merge_tree_part_schema_match_mode = 'match_by_position'` (the default) the source and destination must have the same number of columns; with `'match_by_name'` they must have the same set of column names. Set `ignore_extra_source_columns = 1` to allow a source table with columns that have no corresponding destination column; the destination having a column absent from the source is still always rejected. 2. **`PARTITION BY` expressions** - for destinations other than data lakes, the source and destination `PARTITION BY` expressions must be identical. For Apache Iceberg destinations, the source partition key must match the destination partition fields and transforms. 3. **Partition key column positions and layouts** - every top-level column that provides a column or subcolumn used by the source table's partition key must have the same name at the same position in the destination table's schema. Named `Tuple` elements within such a column must also be declared in the same order, including tuples nested inside `Array` or `Map`. This applies even if both tables' `PARTITION BY` expressions are textually identical. See [`EXPORT PART` requirements](/docs/en/antalya/part_export.md#requirements) for a worked example and the corresponding exception message. @@ -128,15 +128,32 @@ 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) +### `export_merge_tree_part_schema_match_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. +- **Type**: `MergeTreePartExportSchemaMatchMode` +- **Default**: `match_by_position` +- **Description**: Controls how `EXPORT PART`/`EXPORT PARTITION` matches source `MergeTree` columns to destination columns. Possible values: + - `match_by_position` (default) - columns are matched positionally, like `INSERT INTO dest SELECT * FROM src`. Column names are not otherwise considered. + - `match_by_name` - every destination column is matched to a source column with the same exact, case-sensitive name, so destination columns may be declared in a different order than the source. A destination column absent from the source, including when it was renamed, throws `THERE_IS_NO_COLUMN`; there is no positional fallback. - 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. + See `ignore_extra_source_columns` below for how a source column without a corresponding destination column is handled in each mode. + +### `ignore_extra_source_columns` (Optional) + +- **Type**: `Bool` +- **Default**: `false` +- **Description**: Controls whether `EXPORT PART`/`EXPORT PARTITION` tolerates source `MergeTree` columns that have no corresponding destination column. + - `false` (default) - such a source column is rejected: the source and destination must match exactly. With `export_merge_tree_part_schema_match_mode = 'match_by_position'` this means the same number of columns; with `'match_by_name'` this means the same set of column names, so a source table with columns absent from the destination is rejected even if the matched columns would otherwise be compatible. A mismatch throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. + - `true` - a source column without a corresponding destination column is dropped and not exported, instead of throwing. In `match_by_position` mode, this allows a source with extra trailing columns (the destination having more columns than the source is still always rejected). In `match_by_name` mode, this allows source columns whose name has no destination counterpart, which may occur in any position. + + Extra 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. Type conversion and `export_merge_tree_part_allow_lossy_cast` are applied after columns are matched. + + Error behavior: + + - With `ignore_extra_source_columns = false` (default), a source column without a corresponding destination column throws `NUMBER_OF_COLUMNS_DOESNT_MATCH` - in `match_by_position` mode this means any column-count mismatch, in `match_by_name` mode this means the source and destination column-name sets differ. + - In `match_by_position` mode, the destination having more columns than the source always throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`, regardless of `ignore_extra_source_columns`. + - In `match_by_name` mode, a destination column absent from the source (including a renamed one) always throws `THERE_IS_NO_COLUMN`, regardless of `ignore_extra_source_columns`; there is no positional fallback. + - After columns have been matched successfully, a cast rejected by the export type-safety check throws `INCOMPATIBLE_COLUMNS`. ## Examples diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 4109ce49ac12..45eb39385754 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7627,11 +7627,20 @@ Allow `EXPORT PART`/`EXPORT PARTITION` to apply lossy (non-value-preserving) cas 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`. + DECLARE(MergeTreePartExportSchemaMatchMode, export_merge_tree_part_schema_match_mode, MergeTreePartExportSchemaMatchMode::match_by_position, R"( +Controls how `EXPORT PART`/`EXPORT PARTITION` matches source `MergeTree` columns to destination columns. 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. +- `match_by_position` (default) - columns are matched positionally, like `INSERT INTO dest SELECT * FROM src`. Column names are not otherwise considered. +- `match_by_name` - every destination column is matched to a source column with the same exact, case-sensitive name, so destination columns may be declared in a different order than the source. A destination column absent from the source throws `THERE_IS_NO_COLUMN`, with no positional fallback. + +See also `ignore_extra_source_columns`, which controls whether a source column without a corresponding destination column is dropped or rejected. +)", 0) \ + DECLARE(Bool, ignore_extra_source_columns, false, R"( +Controls whether `EXPORT PART`/`EXPORT PARTITION` tolerates source `MergeTree` columns that have no corresponding destination column. +- `false` (default) - such a source column is rejected: the source and destination must match exactly (in `export_merge_tree_part_schema_match_mode = 'match_by_position'`, this means the same number of columns; in `'match_by_name'`, the same set of column names). A mismatch throws `NUMBER_OF_COLUMNS_DOESNT_MATCH`. +- `true` - a source column without a corresponding destination column is dropped and not exported, instead of throwing. In `match_by_position` mode, this allows a source with extra trailing columns (the destination having more columns than the source is still always rejected). In `match_by_name` mode, this allows source columns whose name has no destination counterpart. + +Extra 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. )", 0) \ \ /* ####################################################### */ \ diff --git a/src/Core/Settings.h b/src/Core/Settings.h index a5fef6160201..a166435fcd95 100644 --- a/src/Core/Settings.h +++ b/src/Core/Settings.h @@ -85,7 +85,7 @@ class WriteBuffer; M(CLASS_NAME, Map) \ M(CLASS_NAME, MaxThreads) \ M(CLASS_NAME, MergeTreePartExportFileAlreadyExistsPolicy) \ - M(CLASS_NAME, MergeTreePartExportSchemaMismatchMode) \ + M(CLASS_NAME, MergeTreePartExportSchemaMatchMode) \ 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 60b9114149a6..580d09fc8703 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -44,7 +44,8 @@ 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_part_schema_match_mode", "match_by_position", "match_by_position", "New setting to control how EXPORT PART/EXPORT PARTITION matches source columns to destination columns"}, + {"ignore_extra_source_columns", false, false, "New setting to allow EXPORT PART/EXPORT PARTITION when the source table has columns absent from 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 9aadee2db780..351f8f6be7a2 100644 --- a/src/Core/SettingsEnums.cpp +++ b/src/Core/SettingsEnums.cpp @@ -483,7 +483,7 @@ 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(MergeTreePartExportSchemaMatchMode, ErrorCodes::BAD_ARGUMENTS); IMPLEMENT_SETTING_AUTO_ENUM(ExportPartitionAllOnError, ErrorCodes::BAD_ARGUMENTS); diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h index 0c82d40345cd..a325835452d5 100644 --- a/src/Core/SettingsEnums.h +++ b/src/Core/SettingsEnums.h @@ -574,13 +574,13 @@ enum class MergeTreePartExportFileAlreadyExistsPolicy : uint8_t DECLARE_SETTING_ENUM(MergeTreePartExportFileAlreadyExistsPolicy) -enum class MergeTreePartExportSchemaMismatchMode : uint8_t +enum class MergeTreePartExportSchemaMatchMode : uint8_t { - strict, - ignore_extra_source_columns_by_position, + match_by_position, + match_by_name, }; -DECLARE_SETTING_ENUM(MergeTreePartExportSchemaMismatchMode) +DECLARE_SETTING_ENUM(MergeTreePartExportSchemaMatchMode) enum class ExportPartitionAllOnError : uint8_t { diff --git a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h index 753095900fa8..791691b849da 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h @@ -248,7 +248,8 @@ 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::optional schema_match_mode; + std::optional ignore_extra_source_columns; std::string toJsonString() const { @@ -291,8 +292,10 @@ 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))); + if (schema_match_mode) + json.set("schema_match_mode", String(magic_enum::enum_name(*schema_match_mode))); + if (ignore_extra_source_columns) + json.set("ignore_extra_source_columns", *ignore_extra_source_columns); std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM oss.exceptions(std::ios::failbit); Poco::JSON::Stringifier::stringify(json, oss); @@ -360,15 +363,20 @@ 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 + /// Left unset (nullopt) for tasks created before these fields existed - such tasks were + /// always scheduled under the old, strict column-matching 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")) + /// `match_by_position` with `ignore_extra_source_columns = false`. + if (json->has("schema_match_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; + const auto schema_match_mode = magic_enum::enum_cast(json->getValue("schema_match_mode")); + if (schema_match_mode) + manifest.schema_match_mode = schema_match_mode; + } + + if (json->has("ignore_extra_source_columns")) + { + manifest.ignore_extra_source_columns = json->getValue("ignore_extra_source_columns"); } if (json->has("parquet_compression_method")) diff --git a/src/Storages/MergeTree/ExportPartTask.cpp b/src/Storages/MergeTree/ExportPartTask.cpp index 5ef537bbef1c..2126601ce7c3 100644 --- a/src/Storages/MergeTree/ExportPartTask.cpp +++ b/src/Storages/MergeTree/ExportPartTask.cpp @@ -65,7 +65,8 @@ 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; + extern const SettingsMergeTreePartExportSchemaMatchMode export_merge_tree_part_schema_match_mode; + extern const SettingsBool ignore_extra_source_columns; } namespace @@ -113,15 +114,6 @@ namespace } } - /// Mirrors `InterpreterInsertQuery::addInsertToSelectPipeline`: positional match, - /// 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, @@ -131,13 +123,16 @@ namespace = 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; + const auto schema_match_mode = + local_context->getSettingsRef()[Setting::export_merge_tree_part_schema_match_mode].value; + const bool ignore_extra_source_columns = + local_context->getSettingsRef()[Setting::ignore_extra_source_columns]; + const bool match_by_name = schema_match_mode == MergeTreePartExportSchemaMatchMode::match_by_name; auto source_columns = plan_for_part.getCurrentHeader()->getColumnsWithTypeAndName(); + const bool src_has_extra_columns = source_columns.size() > destination_columns.size(); - if (ignore_extra_source_columns_by_position && source_columns.size() > destination_columns.size()) + if (!match_by_name && ignore_extra_source_columns && src_has_extra_columns) { LOG_DEBUG(getLogger("ExportPartTask"), "Source has {} columns while destination has {} columns, " @@ -169,7 +164,9 @@ namespace auto dag = ActionsDAG::makeConvertingActions( source_columns, destination_columns, - ActionsDAG::MatchColumnsMode::Position, + match_by_name + ? ActionsDAG::MatchColumnsMode::Name + : ActionsDAG::MatchColumnsMode::Position, local_context); auto expression_step = std::make_unique( @@ -353,8 +350,6 @@ bool ExportPartTask::executeStep() /// This is a hack that materializes the columns before the export so they can be exported to tables that have matching columns materializeSpecialColumns(plan_for_part.getCurrentHeader(), metadata_snapshot, local_context, plan_for_part); - /// Align the pipeline header with the destination's non-materialized sample block, - /// using the same `makeConvertingActions(Position)` call INSERT SELECT performs. addExportConvertingActions(plan_for_part, *destination_storage, local_context); QueryPlanOptimizationSettings optimization_settings(local_context); diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 0c69c24511d3..d19bb423c6c7 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -58,6 +58,7 @@ namespace ErrorCodes extern const int ILLEGAL_TYPE_OF_ARGUMENT; extern const int ILLEGAL_COLUMN; extern const int NUMBER_OF_COLUMNS_DOESNT_MATCH; + extern const int THERE_IS_NO_COLUMN; extern const int INCOMPATIBLE_COLUMNS; extern const int NO_SUCH_COLUMN_IN_TABLE; extern const int FILE_ALREADY_EXISTS; @@ -83,7 +84,8 @@ namespace ErrorCodes namespace Setting { extern const SettingsBool export_merge_tree_part_allow_lossy_cast; - extern const SettingsMergeTreePartExportSchemaMismatchMode export_merge_tree_part_schema_mismatch_mode; + extern const SettingsMergeTreePartExportSchemaMatchMode export_merge_tree_part_schema_match_mode; + extern const SettingsBool ignore_extra_source_columns; } namespace FailPoints @@ -112,6 +114,7 @@ namespace ExportPartitionUtils ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT, ErrorCodes::ILLEGAL_COLUMN, ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH, + ErrorCodes::THERE_IS_NO_COLUMN, ErrorCodes::INCOMPATIBLE_COLUMNS, ErrorCodes::NO_SUCH_COLUMN_IN_TABLE, ErrorCodes::NOT_IMPLEMENTED, @@ -173,12 +176,16 @@ 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). + /// Manifests written before these settings existed have no value here; such tasks were always + /// scheduled under the old, strict column-matching check, so an absent value must resolve to + /// `match_by_position` / `false` regardless of the ambient context's settings (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)))); + "export_merge_tree_part_schema_match_mode", + String(magic_enum::enum_name(manifest.schema_match_mode.value_or(MergeTreePartExportSchemaMatchMode::match_by_position)))); + context_copy->setSetting( + "ignore_extra_source_columns", + manifest.ignore_extra_source_columns.value_or(false)); 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))); @@ -708,19 +715,21 @@ namespace ExportPartitionUtils const ColumnWithTypeAndName & source_column, const ColumnWithTypeAndName & destination_column, size_t position, - const StorageID & destination_storage_id) + const StorageID & destination_storage_id, + bool match_by_name) { if (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.", + "PART/PARTITION {} so partition key columns must be declared {} in both tables.", destination_storage_id.getFullTableName(), source_column.name, position, - destination_column.name); + destination_column.name, + match_by_name ? "matches columns by name" : "matches columns by position", + match_by_name ? "with the same name" : "at the same position"); if (!haveSameTupleElementLayout(source_column.type, destination_column.type)) throw Exception( @@ -733,6 +742,24 @@ namespace ExportPartitionUtils source_column.type->getName(), destination_column.type->getName()); } + + void verifyExportColumnCastIsSafe( + const ColumnWithTypeAndName & source_column, + const ColumnWithTypeAndName & destination_column, + const StorageID & destination_storage_id) + { + if (canBeSafelyCast(source_column.type, destination_column.type)) + return; + + throw Exception(ErrorCodes::INCOMPATIBLE_COLUMNS, + "Cannot export to {}: column '{}' requires a lossy cast from {} to {}, " + "which may change values. Set `export_merge_tree_part_allow_lossy_cast = 1` " + "to allow lossy casts during export.", + destination_storage_id.getFullTableName(), + destination_column.name, + source_column.type->getName(), + destination_column.type->getName()); + } } void assertPartitionKeyASTAreEqual( @@ -748,6 +775,57 @@ namespace ExportPartitionUtils throw Exception(ErrorCodes::BAD_ARGUMENTS, "Tables have different partition key"); } + void verifyExportColumnCastsAreSafe( + const ColumnsWithTypeAndName & source_columns, + const ColumnsWithTypeAndName & destination_columns, + MergeTreePartExportSchemaMatchMode schema_match_mode, + bool ignore_extra_source_columns, + const StorageID & destination_storage_id) + { + if (schema_match_mode == MergeTreePartExportSchemaMatchMode::match_by_name) + { + /// `ignore_extra_source_columns` only excuses the source having MORE columns than the + /// destination; a source with fewer columns is always a mismatch. + const bool src_has_extra_columns = source_columns.size() > destination_columns.size(); + if (source_columns.size() != destination_columns.size() + && !(ignore_extra_source_columns && src_has_extra_columns)) + throw Exception( + ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH, + "Number of columns doesn't match (source: {} and result: {})", + source_columns.size(), + destination_columns.size()); + + std::unordered_map source_columns_by_name; + source_columns_by_name.reserve(source_columns.size()); + for (const auto & source_column : source_columns) + source_columns_by_name.emplace(source_column.name, &source_column); + + for (const auto & destination_column : destination_columns) + { + const auto source_it = source_columns_by_name.find(destination_column.name); + if (source_it == source_columns_by_name.end()) + throw Exception( + ErrorCodes::THERE_IS_NO_COLUMN, + "Cannot find column `{}` in source stream", + destination_column.name); + + verifyExportColumnCastIsSafe(*source_it->second, destination_column, destination_storage_id); + } + return; + } + + if (source_columns.size() < destination_columns.size() + || (!ignore_extra_source_columns && source_columns.size() != destination_columns.size())) + throw Exception( + ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH, + "Number of columns doesn't match (source: {} and result: {})", + source_columns.size(), + destination_columns.size()); + + for (size_t i = 0; i < destination_columns.size(); ++i) + verifyExportColumnCastIsSafe(source_columns[i], destination_columns[i], destination_storage_id); + } + void verifyExportSchemaCastable( const StorageMetadataPtr & source_metadata, const StorageMetadataPtr & destination_metadata, @@ -765,16 +843,14 @@ namespace ExportPartitionUtils 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; + const auto schema_match_mode = + context->getSettingsRef()[Setting::export_merge_tree_part_schema_match_mode].value; + const bool ignore_extra_source_columns = + context->getSettingsRef()[Setting::ignore_extra_source_columns]; + const bool match_by_name = schema_match_mode == MergeTreePartExportSchemaMatchMode::match_by_name; + const bool src_has_extra_columns = source_columns.size() > destination_columns.size(); - if (ignore_extra_source_columns_by_position && source_columns.size() > destination_columns.size()) + if (!match_by_name && ignore_extra_source_columns && src_has_extra_columns) { LOG_DEBUG(getLogger("ExportPartitionUtils"), "Source has {} columns while destination has {} columns, " @@ -785,10 +861,23 @@ namespace ExportPartitionUtils source_columns.resize(destination_columns.size()); } + /// `makeConvertingActions` in `Name` mode silently ignores an unreferenced source column, so it must be rejected here explicitly. + /// `ignore_extra_source_columns` only excuses the source having MORE columns than the destination; + /// a source with fewer columns is always a mismatch. + if (match_by_name && source_columns.size() != destination_columns.size() + && !(ignore_extra_source_columns && src_has_extra_columns)) + throw Exception( + ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH, + "Number of columns doesn't match (source: {} and result: {})", + source_columns.size(), + destination_columns.size()); + (void) ActionsDAG::makeConvertingActions( source_columns, destination_columns, - ActionsDAG::MatchColumnsMode::Position, + match_by_name + ? ActionsDAG::MatchColumnsMode::Name + : ActionsDAG::MatchColumnsMode::Position, context); const auto & source_columns_description = source_metadata->getColumns(); @@ -805,29 +894,46 @@ namespace ExportPartitionUtils 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) + if (match_by_name) { - const auto & source_column = source_columns[i]; - const auto & destination_column = destination_columns[i]; + std::unordered_map source_positions_by_name; + source_positions_by_name.reserve(source_columns.size()); + for (size_t i = 0; i < source_columns.size(); ++i) + source_positions_by_name.emplace(source_columns[i].name, i); - if (partition_key_owner_columns.contains(source_column.name)) - verifyPartitionKeyColumn(source_column, destination_column, i, destination_storage_id); + for (const auto & destination_column : destination_columns) + { + if (!partition_key_owner_columns.contains(destination_column.name)) + continue; - /// Lossy casts may silently change values, so reject them unless the user opts in. - if (allow_lossy_cast) - continue; + const auto source_it = source_positions_by_name.find(destination_column.name); + if (source_it == source_positions_by_name.end()) + continue; - if (!canBeSafelyCast(source_column.type, destination_column.type)) - throw Exception(ErrorCodes::INCOMPATIBLE_COLUMNS, - "Cannot export to {}: column '{}' requires a lossy cast from {} to {}, " - "which may change values. Set `export_merge_tree_part_allow_lossy_cast = 1` " - "to allow lossy casts during export.", - destination_storage_id.getFullTableName(), - destination_column.name, - source_column.type->getName(), - destination_column.type->getName()); + verifyPartitionKeyColumn( + source_columns[source_it->second], destination_column, source_it->second, destination_storage_id, + /*match_by_name=*/ true); + } } + else + { + const size_t num_columns = std::min(source_columns.size(), destination_columns.size()); + for (size_t i = 0; i < num_columns; ++i) + if (partition_key_owner_columns.contains(source_columns[i].name)) + verifyPartitionKeyColumn(source_columns[i], destination_columns[i], i, destination_storage_id, + /*match_by_name=*/ false); + } + + /// Lossy casts may silently change values, so reject them unless the user opts in. + if (allow_lossy_cast) + return; + + verifyExportColumnCastsAreSafe( + source_columns, + destination_columns, + schema_match_mode, + ignore_extra_source_columns, + destination_storage_id); } } diff --git a/src/Storages/MergeTree/ExportPartitionUtils.h b/src/Storages/MergeTree/ExportPartitionUtils.h index 652ec9223394..a56377728b79 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.h +++ b/src/Storages/MergeTree/ExportPartitionUtils.h @@ -4,6 +4,8 @@ #include #include #include +#include +#include #include #include #include "Storages/IStorage.h" @@ -93,17 +95,17 @@ namespace ExportPartitionUtils 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. - /// - /// 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 verifyExportColumnCastsAreSafe( + const ColumnsWithTypeAndName & source_columns, + const ColumnsWithTypeAndName & destination_columns, + MergeTreePartExportSchemaMatchMode schema_match_mode, + bool ignore_extra_source_columns, + const StorageID & destination_storage_id); + + /// Validates that source columns can be exported into the destination with the configured + /// positional or name-based CAST matching (`export_merge_tree_part_schema_match_mode`) and + /// unmatched-column policy (`ignore_extra_source_columns`). Lossy casts are rejected unless + /// `export_merge_tree_part_allow_lossy_cast` is set. Throws BAD_ARGUMENTS on any violation. void verifyExportSchemaCastable( const StorageMetadataPtr & source_metadata, const StorageMetadataPtr & destination_metadata, diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index 0519730a5ee8..dfc49f1bf7bb 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6782,7 +6782,6 @@ void MergeTreeData::exportPartToTable( #endif } - /// Positional CAST matching, like `INSERT INTO dest SELECT * FROM src`. ExportPartitionUtils::verifyExportSchemaCastable( source_metadata_ptr, destination_metadata_ptr, dest_storage->getStorageID(), query_context); diff --git a/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp b/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp index df20755ba590..71206642da1a 100644 --- a/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp +++ b/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp @@ -3,7 +3,10 @@ #include #include #include +#include #include +#include +#include #include #include #include @@ -11,13 +14,42 @@ namespace DB { +namespace ErrorCodes +{ + extern const int INCOMPATIBLE_COLUMNS; + extern const int THERE_IS_NO_COLUMN; + extern const int NUMBER_OF_COLUMNS_DOESNT_MATCH; +} + namespace Setting { - extern const SettingsMergeTreePartExportSchemaMismatchMode export_merge_tree_part_schema_mismatch_mode; + extern const SettingsMergeTreePartExportSchemaMatchMode export_merge_tree_part_schema_match_mode; + extern const SettingsBool ignore_extra_source_columns; } namespace { + template + ColumnWithTypeAndName makeColumn(const String & name) + { + auto type = std::make_shared(); + return {type->createColumn(), type, name}; + } + + template + void expectExceptionCode(Function && function, int expected_code) + { + try + { + function(); + FAIL() << "Expected exception code " << expected_code; + } + catch (const Exception & exception) + { + EXPECT_EQ(exception.code(), expected_code) << exception.message(); + } + } + ExportReplicatedMergeTreePartitionManifest makeValidManifest() { ExportReplicatedMergeTreePartitionManifest manifest; @@ -114,61 +146,276 @@ TEST_F(ExportPartitionOrderingTest, IterationOrderMatchesCreateTime) } -TEST_F(ExportPartitionManifestBackCompatTest, MissingSchemaMismatchModeParsesAsNullopt) +TEST_F(ExportPartitionManifestBackCompatTest, MissingSchemaMatchModeParsesAsNullopt) +{ + auto manifest = makeValidManifest(); + manifest.schema_match_mode = MergeTreePartExportSchemaMatchMode::match_by_name; + + Poco::JSON::Parser parser; + auto json = parser.parse(manifest.toJsonString()).extract(); + json->remove("schema_match_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_match_mode.has_value()); +} + +TEST_F(ExportPartitionManifestBackCompatTest, SchemaMatchModeRoundTripsForEveryValue) +{ + for (const auto value : magic_enum::enum_values()) + { + auto manifest = makeValidManifest(); + manifest.schema_match_mode = value; + + auto parsed = ExportReplicatedMergeTreePartitionManifest::fromJsonString(manifest.toJsonString()); + + ASSERT_TRUE(parsed.schema_match_mode.has_value()) << "value=" << magic_enum::enum_name(value); + EXPECT_EQ(*parsed.schema_match_mode, value) << "value=" << magic_enum::enum_name(value); + } +} + +TEST_F(ExportPartitionManifestBackCompatTest, MissingIgnoreExtraSourceColumnsParsesAsNullopt) { auto manifest = makeValidManifest(); - manifest.schema_mismatch_mode = MergeTreePartExportSchemaMismatchMode::ignore_extra_source_columns_by_position; + manifest.ignore_extra_source_columns = true; Poco::JSON::Parser parser; auto json = parser.parse(manifest.toJsonString()).extract(); - json->remove("schema_mismatch_mode"); + json->remove("ignore_extra_source_columns"); 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()); + EXPECT_FALSE(parsed.ignore_extra_source_columns.has_value()); } -TEST_F(ExportPartitionManifestBackCompatTest, SchemaMismatchModeRoundTripsForEveryValue) +TEST_F(ExportPartitionManifestBackCompatTest, IgnoreExtraSourceColumnsRoundTripsForEveryValue) { - for (const auto value : magic_enum::enum_values()) + for (const bool value : {false, true}) { auto manifest = makeValidManifest(); - manifest.schema_mismatch_mode = value; + manifest.ignore_extra_source_columns = 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); + ASSERT_TRUE(parsed.ignore_extra_source_columns.has_value()) << "value=" << value; + EXPECT_EQ(*parsed.ignore_extra_source_columns, value) << "value=" << value; } } -TEST_F(ExportPartitionManifestBackCompatTest, MissingSchemaMismatchModeFallsBackToStrictInWorkerContext) +TEST_F(ExportPartitionManifestBackCompatTest, MissingSchemaMatchSettingsFallBackToDefaultsInWorkerContext) { auto manifest = makeValidManifest(); - ASSERT_FALSE(manifest.schema_mismatch_mode.has_value()); + ASSERT_FALSE(manifest.schema_match_mode.has_value()); + ASSERT_FALSE(manifest.ignore_extra_source_columns.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); + worker_context->getSettingsRef()[Setting::export_merge_tree_part_schema_match_mode].value, + MergeTreePartExportSchemaMatchMode::match_by_position); + EXPECT_EQ( + worker_context->getSettingsRef()[Setting::ignore_extra_source_columns].value, + false); } -TEST_F(ExportPartitionManifestBackCompatTest, SchemaMismatchModeAppliedToWorkerContextForEveryValue) +TEST_F(ExportPartitionManifestBackCompatTest, SchemaMatchModeAppliedToWorkerContextForEveryValue) { - for (const auto value : magic_enum::enum_values()) + for (const auto value : magic_enum::enum_values()) { auto manifest = makeValidManifest(); - manifest.schema_mismatch_mode = value; + manifest.schema_match_mode = value; auto worker_context = ExportPartitionUtils::getContextCopyWithTaskSettings(getContext().context, manifest); EXPECT_EQ( - worker_context->getSettingsRef()[Setting::export_merge_tree_part_schema_mismatch_mode].value, + worker_context->getSettingsRef()[Setting::export_merge_tree_part_schema_match_mode].value, value) << "value=" << magic_enum::enum_name(value); } } +TEST_F(ExportPartitionManifestBackCompatTest, IgnoreExtraSourceColumnsAppliedToWorkerContextForEveryValue) +{ + for (const bool value : {false, true}) + { + auto manifest = makeValidManifest(); + manifest.ignore_extra_source_columns = value; + + auto worker_context = ExportPartitionUtils::getContextCopyWithTaskSettings(getContext().context, manifest); + + EXPECT_EQ( + worker_context->getSettingsRef()[Setting::ignore_extra_source_columns].value, + value) << "value=" << value; + } +} + +TEST(ExportColumnCastsTest, UsesSelectedMatchingMode) +{ + const ColumnsWithTypeAndName source_columns = { + makeColumn("id"), + makeColumn("year"), + makeColumn("payload"), + makeColumn("extra"), + }; + const ColumnsWithTypeAndName destination_columns = { + makeColumn("payload"), + makeColumn("year"), + makeColumn("id"), + }; + const StorageID destination_storage_id{"test", "destination"}; + + expectExceptionCode( + [&] + { + ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_position, + /*ignore_extra_source_columns=*/ true, + destination_storage_id); + }, + ErrorCodes::INCOMPATIBLE_COLUMNS); + + EXPECT_NO_THROW(ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_name, + /*ignore_extra_source_columns=*/ true, + destination_storage_id)); +} + +TEST(ExportColumnCastsTest, MatchByNameAcceptsReorderedColumnsWithEqualColumnCount) +{ + const ColumnsWithTypeAndName source_columns = { + makeColumn("id"), + makeColumn("year"), + makeColumn("payload"), + }; + const ColumnsWithTypeAndName reordered_destination_columns = { + makeColumn("payload"), + makeColumn("year"), + makeColumn("id"), + }; + const StorageID destination_storage_id{"test", "destination"}; + + for (const bool ignore_extra_source_columns : {false, true}) + { + expectExceptionCode( + [&] + { + ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + reordered_destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_position, + ignore_extra_source_columns, + destination_storage_id); + }, + ErrorCodes::INCOMPATIBLE_COLUMNS); + + EXPECT_NO_THROW(ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + reordered_destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_name, + ignore_extra_source_columns, + destination_storage_id)); + } + + const ColumnsWithTypeAndName same_order_destination_columns = { + makeColumn("id"), + makeColumn("year"), + makeColumn("payload"), + }; + + for (const auto mode : {MergeTreePartExportSchemaMatchMode::match_by_position, MergeTreePartExportSchemaMatchMode::match_by_name}) + for (const bool ignore_extra_source_columns : {false, true}) + EXPECT_NO_THROW(ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, same_order_destination_columns, mode, ignore_extra_source_columns, destination_storage_id)); +} + +TEST(ExportColumnCastsTest, MatchByNameRejectsUnmatchedSourceColumnUnlessIgnored) +{ + const ColumnsWithTypeAndName source_columns = { + makeColumn("id"), + makeColumn("year"), + makeColumn("extra"), + }; + const ColumnsWithTypeAndName destination_columns = { + makeColumn("id"), + makeColumn("year"), + }; + const StorageID destination_storage_id{"test", "destination"}; + + expectExceptionCode( + [&] + { + ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_name, + /*ignore_extra_source_columns=*/ false, + destination_storage_id); + }, + ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH); + + EXPECT_NO_THROW(ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_name, + /*ignore_extra_source_columns=*/ true, + destination_storage_id)); +} + +TEST(ExportColumnCastsTest, RejectsLossyCastAfterMatchingByName) +{ + const ColumnsWithTypeAndName source_columns = { + makeColumn("id"), + makeColumn("year"), + makeColumn("extra"), + }; + const ColumnsWithTypeAndName destination_columns = { + makeColumn("id"), + makeColumn("year"), + }; + + expectExceptionCode( + [&] + { + ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_name, + /*ignore_extra_source_columns=*/ true, + StorageID{"test", "destination"}); + }, + ErrorCodes::INCOMPATIBLE_COLUMNS); +} + +TEST(ExportColumnCastsTest, RejectsMissingDestinationColumnAfterMatchingByName) +{ + const ColumnsWithTypeAndName source_columns = { + makeColumn("id"), + makeColumn("year"), + makeColumn("extra"), + }; + const ColumnsWithTypeAndName destination_columns = { + makeColumn("renamed_id"), + makeColumn("year"), + }; + + expectExceptionCode( + [&] + { + ExportPartitionUtils::verifyExportColumnCastsAreSafe( + source_columns, + destination_columns, + MergeTreePartExportSchemaMatchMode::match_by_name, + /*ignore_extra_source_columns=*/ true, + StorageID{"test", "destination"}); + }, + ErrorCodes::THERE_IS_NO_COLUMN); +} + } diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index 9d40c5b35df1..b3bd2e8db53f 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -229,7 +229,8 @@ 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 SettingsMergeTreePartExportSchemaMatchMode export_merge_tree_part_schema_match_mode; + extern const SettingsBool ignore_extra_source_columns; 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; @@ -8412,7 +8413,6 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & auto src_snapshot = getInMemoryMetadataPtr(); auto destination_snapshot = dest_storage->getInMemoryMetadataPtr(); - /// Positional CAST matching, like `INSERT INTO dest SELECT * FROM src`. ExportPartitionUtils::verifyExportSchemaCastable( src_snapshot, destination_snapshot, dest_storage->getStorageID(), query_context); @@ -8536,7 +8536,8 @@ 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; + manifest.schema_match_mode = query_context->getSettingsRef()[Setting::export_merge_tree_part_schema_match_mode].value; + manifest.ignore_extra_source_columns = query_context->getSettingsRef()[Setting::ignore_extra_source_columns].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 34590115d7de..578592f6ee0f 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 @@ -37,6 +37,12 @@ ) +EXTRA_SOURCE_COLUMN_MODES = [ + pytest.param("match_by_position", id="by-position"), + pytest.param("match_by_name", id="by-name"), +] + + # --------------------------------------------------------------------------- # Cluster fixture # --------------------------------------------------------------------------- @@ -757,13 +763,8 @@ 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. - """ +@pytest.mark.parametrize("schema_match_mode", EXTRA_SOURCE_COLUMN_MODES) +def test_export_part_source_more_columns_allowed_with_ignore_extra_setting(cluster, schema_match_mode): node = cluster.instances["node1"] sfx = unique_suffix() mt = f"mt_ignore_extra_{sfx}" @@ -777,7 +778,10 @@ def test_export_part_source_more_columns_allowed_with_ignore_extra_setting(clust 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'", + extra_settings=( + f"export_merge_tree_part_schema_match_mode = '{schema_match_mode}', " + f"ignore_extra_source_columns = 1" + ), ) wait_for_export_part(node=node, table=mt, part=part_2020) @@ -793,13 +797,16 @@ def test_export_part_source_more_columns_allowed_with_ignore_extra_setting(clust 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. - """ +@pytest.mark.parametrize( + "schema_match_mode,expected_error", + [ + pytest.param("match_by_position", "NUMBER_OF_COLUMNS_DOESNT_MATCH", id="by-position"), + pytest.param("match_by_name", "NUMBER_OF_COLUMNS_DOESNT_MATCH", id="by-name"), + ], +) +def test_export_part_column_count_mismatch_source_fewer_still_rejected_with_ignore_extra_setting( + cluster, schema_match_mode, expected_error +): node = cluster.instances["node1"] sfx = unique_suffix() mt = f"mt_ignore_extra_fewer_{sfx}" @@ -815,12 +822,10 @@ def test_export_part_column_count_mismatch_source_fewer_still_rejected_with_igno 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