EXPORT PARTITION with position matching + extra columns in source

docs/en/antalya/part_export.md

@@ -55,7 +55,7 @@ Source and destination tables must support positional schema conversion. The fol
 
 The following must match between source and destination:
 
-1. **Column count** - source and destination must have the same number of columns.
+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.
 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,6 +136,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
 

docs/en/antalya/partition_export.md

@@ -47,7 +47,7 @@ TO TABLE [destination_database.]destination_table
 
 `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:
 
-1. **Column count** - source and destination must have the same number of columns.
+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.
 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,6 +128,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

src/Core/Settings.cpp

@@ -7626,6 +7626,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) \
     \
     /* ####################################################### */ \

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) \

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"},

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);
 
 }

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,

src/Storages/ExportReplicatedMergeTreePartitionManifest.h

@@ -248,6 +248,7 @@ struct ExportReplicatedMergeTreePartitionManifest
     std::optional<UInt64> output_format_compression_level;
     std::optional<UInt64> parquet_row_group_size;
     std::optional<UInt64> parquet_row_group_size_bytes;
+    std::optional<MergeTreePartExportSchemaMismatchMode> 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<bool>("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<MergeTreePartExportSchemaMismatchMode>(json->getValue<String>("schema_mismatch_mode"));
+            if (schema_mismatch_mode)
+                manifest.schema_mismatch_mode = schema_mismatch_mode;
+        }
 
         if (json->has("parquet_compression_method"))
         {

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,17 +117,58 @@ 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,
         const ContextPtr & local_context)
     {
         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<ExpressionStep>(
+                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);
 

src/Storages/MergeTree/ExportPartitionUtils.cpp

@@ -83,6 +83,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
@@ -172,6 +173,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)));
@@ -755,8 +762,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,