EXPORT PARTITION with position matching + extra columns in source

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
 

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

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

@@ -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,

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,