Plain merge tree partition exports

antalya/docs/design/alter-table-export-part-partition.md

@@ -171,10 +171,11 @@ The design should address the following topics in the near future.
 - `tests/queries/0_stateless/03572_export_merge_tree_part_special_columns.sh`
 - `tests/queries/0_stateless/03572_export_replicated_merge_tree_part_to_object_storage.sh`
 - `tests/queries/0_stateless/03572_export_replicated_merge_tree_part_to_object_storage_simple.sql`
-- `tests/queries/0_stateless/03604_export_merge_tree_partition.sh`
+- `tests/queries/0_stateless/05027_export_partition_merge_tree.sh`
+- `tests/queries/0_stateless/05028_export_partition_replicated_merge_tree.sh`
 - `tests/queries/0_stateless/03608_export_merge_tree_part_filename_pattern.sh`
 - `tests/integration/test_export_merge_tree_part_to_object_storage/test.py`
-- `tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py`
+- `tests/integration/test_export_partition_to_object_storage/test.py`
 
 ---
 
@@ -619,28 +620,35 @@ path — a REST / Glue fixture. The commit-file test above only exercises plain
 object-storage atomicity.
 
 Do not add `no-parallel` to any new test unless explicitly required by shared S3 bucket paths;
-`03604` currently has the tag and should be re-examined to see whether unique per-run paths
-remove the need.
+the replicated wrappers (`05028`, `05030`) carry the tag inherited from the test they replaced,
+and should be re-examined: the shared bodies now derive every table and S3 filename from
+`$CLICKHOUSE_DATABASE`, so the per-run paths are already unique.
 
 ### Integration tests — `tests/integration`
 
 **Keep (modified in this PR):**
 
 - `test_export_merge_tree_part_to_object_storage/` — part export in a multi-node setup.
   PR 1618 makes minor adjustments.
-- `test_export_replicated_mt_partition_to_object_storage/` — partition export across
-  replicas, including `wait_for_export_status`, retry counting, and replica failure
+- `test_export_partition_to_object_storage/` — partition export to a plain object-storage
+  destination, including `wait_for_export_status`, retry counting, and replica failure
   scenarios. PR 1618 removes the `s3_retries.xml` config and reshapes several test cases
-  against the new shared helpers.
+  against the new shared helpers. The scenarios that do not depend on cross-replica
+  coordination request the `source_engine` fixture and run once per MergeTree flavour; the
+  rest stay replicated-only.
 
 **New in PR 1618:**
 
 - `test_export_merge_tree_part_to_iceberg/` — per-part export to an Iceberg destination,
   covering golden path, sidecar emission, manifest shape, and error paths.
-- `test_export_replicated_mt_partition_to_iceberg/` — distributed partition export to
-  Iceberg across replicas, including `test_export_task_timeout_kills_stuck_pending_task`
-  (uses the `export_partition_commit_always_throw` failpoint to exhaust the commit path,
-  then asserts the timeout transitions the task to `KILLED`).
+- `test_export_partition_to_iceberg/` — partition export to Iceberg, including
+  `test_export_task_timeout_kills_stuck_pending_task` (uses the
+  `export_partition_commit_always_throw` failpoint to exhaust the commit path, then asserts
+  the timeout transitions the task to `KILLED`). Parametrized over the source engine on the
+  same terms as the object-storage suite.
+- `test_export_mt_partition_to_object_storage/` — the plain-`MergeTree` behavior the unified
+  suites cannot express: a Keeper-free cluster (proving no ensemble is needed) and
+  restart-resume from the on-disk task descriptor.
 - `test_storage_iceberg_with_spark/test_export_partition_iceberg.py` — catalog-less
   Iceberg round-trip; Spark reads ClickHouse-written data and verifies schema, partition
   layout, and snapshot atomicity.
@@ -653,6 +661,28 @@ remove the need.
   - `tests/integration/helpers/iceberg_export_stats.py` — sidecar decoders and stats
     assertion helpers.
 
+**Known divergence between the engines:**
+
+- Dispatch-time destination failures do not fail a `Replicated*MergeTree` task. When the
+  destination is dropped or recreated with an incompatible schema after the export was
+  scheduled but before the part task is dispatched, the plain scheduler fails the task, while
+  the replicated path leaves it `PENDING` until
+  `export_merge_tree_partition_task_timeout_seconds` (a day by default). For the dropped
+  destination the cause is that `UNKNOWN_TABLE` is missing from
+  `ExportPartitionUtils::isNonRetryableExportError`, so the failure counts as retryable. The
+  schema-mismatch case is not explained by that, since `INCOMPATIBLE_COLUMNS` is in the
+  non-retryable set — the replicated dispatch appears not to reach the classification.
+  `test_export_partition_to_object_storage/test_failures.py::test_dispatch_fails_when_destination_dropped`
+  and `::test_dispatch_fails_when_destination_schema_incompatible` cover the plain path only;
+  parametrize them over `source_engine` once the replicated path fails the task too.
+- The commit failpoint `export_partition_commit_always_throw` only exists in
+  `ExportPartitionUtils::commit`, the ZooKeeper-coordinated commit routine. A plain `MergeTree`
+  commits through `MergeTreePartitionExportScheduler::tryCommit`, so its commit-failure handling
+  (retry, `FAILED` transition, timeout kill) has no failpoint coverage.
+- An `IN PARTITION`-scoped mutation blocks the export of every other partition on a plain
+  `MergeTree`, because its mutations snapshot is not partition-scoped. See "Pending mutations"
+  under Plain (non-replicated) MergeTree in `docs/en/antalya/partition_export.md`.
+
 **Remaining gaps to add:**
 
 - Initiating-replica dies mid-commit (post-data-file-write, pre-catalog-CAS) — asserts a
@@ -668,7 +698,7 @@ remove the need.
   `EXPORT PARTITION` or the new Keeper manifest fields.
 
 Invocation:
-`python -m ci.praktika run "integration" --test test_export_merge_tree_part_to_object_storage,test_export_replicated_mt_partition_to_object_storage,test_export_merge_tree_part_to_iceberg,test_export_replicated_mt_partition_to_iceberg,test_storage_iceberg_with_spark`.
+`python -m ci.praktika run "integration" --test test_export_merge_tree_part_to_object_storage test_export_partition_to_object_storage test_export_merge_tree_part_to_iceberg test_export_partition_to_iceberg test_export_mt_partition_to_object_storage test_storage_iceberg_with_spark`.
 
 ### Failpoints
 

ci/jobs/scripts/integration_tests_configs.py

@@ -58,7 +58,19 @@ class TC:
         True,
         "pins azurite to fixed host port 10000 (Spark emulator mode); concurrent --dist=each workers collide on bind",
     ),
-    TC("test_export_replicated_mt_partition_to_object_storage/", True, "ZooKeeper can't handle too many parallel requests"),
+    # The rest of test_export_partition_to_object_storage/ (validation, lifecycle) and all of
+    # test_export_partition_to_iceberg/ are parallel-safe: each xdist worker gets its own cluster,
+    # and those modules only create tables and assert on synchronous rejects.
+    TC(
+        "test_export_partition_to_object_storage/test_failures.py",
+        True,
+        "paces itself against retry back-off and scheduler ticks while object storage is cut off; host load makes the timings flaky",
+    ),
+    TC(
+        "test_export_partition_to_object_storage/test_replication.py",
+        True,
+        "ZooKeeper can't handle too many parallel requests",
+    ),
 ]
 
 IMAGES_ENV = {

docs/en/antalya/partition_export.md

@@ -2,15 +2,20 @@
 
 ## Overview
 
-The `ALTER TABLE EXPORT PARTITION` command exports entire partitions from Replicated*MergeTree tables to object storage (S3, Azure Blob Storage, etc.) or data lakes like Apache Iceberg tables (with and without catalogs), typically in Parquet format. This feature coordinates export part operations across all replicas using ZooKeeper.
+The `ALTER TABLE EXPORT PARTITION` command exports entire partitions from `MergeTree`-family tables to object storage (S3, Azure Blob Storage, etc.) or data lakes like Apache Iceberg tables (with and without catalogs), typically in Parquet format.
 
-The set of parts that are exported is based on the list of parts the replica that received the export command sees. The other replicas will assist in the export process if they have those parts locally. Otherwise they will ignore it.
+- On `Replicated*MergeTree` tables the export is coordinated across all replicas using ZooKeeper.
+- On plain (non-replicated) `MergeTree` tables the export runs entirely on the single node that received the command. No ZooKeeper / `clickhouse-keeper` ensemble is required. See [Plain (non-replicated) MergeTree](#plain-non-replicated-mergetree) below.
 
-The partition export tasks can be observed through `system.replicated_partition_exports`. The table is served from each replica's in-memory mirror, so queries do not contact ZooKeeper and are cheap to run. The mirror is refreshed on the manifest-updater poll cycle and on every status change, so a freshly written exception or terminal state may take up to one poll interval to appear. Individual part export progress can be observed as usual through `system.exports`.
+The set of parts that are exported is based on the list of parts the replica that received the export command sees. On `Replicated*MergeTree`, the other replicas will assist in the export process if they have those parts locally. Otherwise they will ignore it.
+
+The partition export tasks of both engines can be observed through `system.partition_exports`. The table is served from an in-memory mirror, so queries do not contact ZooKeeper or disk and are cheap to run. For a `Replicated*MergeTree` source the mirror is refreshed on the manifest-updater poll cycle and on every status change, so a freshly written exception or terminal state may take up to one poll interval to appear; for a plain `MergeTree` source it is updated synchronously with every state change. Individual part export progress can be observed as usual through `system.exports`.
+
+`system.replicated_partition_exports` is kept as an alias of `system.partition_exports` for backwards compatibility. It returns exactly the same rows, including exports of plain `MergeTree` tables. Filter on `source_table` (or join against `system.tables`) if you need only one engine.
 
 The same partition can not be exported to the same destination more than once. There are two ways to override this behavior: either by setting the `export_merge_tree_partition_force_export` setting or waiting for the task to expire.
 
-The export task can be killed by issuing the kill command: `KILL EXPORT PARTITION <where predicate for system.replicated_partition_exports>`.
+The export task can be killed by issuing the kill command: `KILL EXPORT PARTITION <where predicate for system.partition_exports>`.
 
 The task is persistent - it should be resumed after crashes, failures and etc.
 
@@ -35,6 +40,19 @@ The source partition must not be split in the destination. This is validated at
 
 Each MergeTree part will become a separate file with the following name convention: `<table_directory>/<partitioning>/<data_part_name>_<merge_tree_part_checksum>.<format>`. To ensure atomicity, a commit file containing the relative paths of all exported parts is also shipped. A data file should only be considered part of the dataset if a commit file references it. The commit file will be named using the following convention: `<table_directory>/commit_<partition_id>_<transaction_id>`.
 
+## Plain (non-replicated) MergeTree {#plain-non-replicated-mergetree}
+
+The command, its settings, the partition-key compatibility rules and the destination file layout are the same for both engines. Only the coordination differs:
+
+- There is no ZooKeeper coordination and no `clickhouse-keeper` ensemble is needed. The node that received the command persists the task descriptor on the source table's disk and a local background scheduler drives it to completion.
+- The task is still persistent: a node that is restarted (or killed) mid-export resumes the export from the on-disk descriptor.
+- There are no other replicas to assist, so a single node exports every part of the partition. `source_replica` is empty in `system.partition_exports` and `last_exception_per_replica` holds at most one entry, whose `replica` is empty.
+- The commit-path columns (`committed_metadata_file`, `committed_manifest_list`, `committed_manifest_file`, `committed_marker_file`) are always empty, because a plain `MergeTree` does not persist the commit paths. See [Columns that depend on the source engine](#columns-that-depend-on-the-source-engine).
+
+### Pending mutations {#plain-merge-tree-pending-mutations}
+
+The pending-mutation gate is more conservative than on a `Replicated*MergeTree`. A plain `MergeTree` does not scope its mutation snapshot by partition, so a mutation restricted with `IN PARTITION` still marks the parts of every other partition as having pending mutations, and exporting an unaffected partition is refused with `PENDING_MUTATIONS_NOT_ALLOWED`. The gate fails closed - it never exports data that a pending mutation would have changed - so the effect is that you may have to wait for an unrelated mutation to finish, or set `export_merge_tree_part_throw_on_pending_mutations` to `false`.
+
 ## Syntax
 
 ```sql
@@ -66,7 +84,7 @@ TO TABLE [destination_database.]destination_table
 
 - **Type**: `Bool`
 - **Default**: `false`
-- **Description**: Enable export replicated merge tree partition feature. It is experimental and not yet ready for production use.
+- **Description**: Enable the `EXPORT PARTITION` feature for both `Replicated*MergeTree` and plain `MergeTree` tables. It is experimental and not yet ready for production use.
 
 ### Query Settings
 
@@ -183,19 +201,19 @@ WHERE partition_id = '2020'
   AND destination_table = 's3_table'
 ```
 
-The `WHERE` clause filters exports from the `system.replicated_partition_exports` table. You can use any columns from that table in the filter.
+The `WHERE` clause filters exports from the `system.partition_exports` table, which covers both `Replicated*MergeTree` and plain `MergeTree` sources, so a single `KILL EXPORT PARTITION` targets either engine. You can use any column of that table in the filter (for example `partition_id`, `source_table`, `destination_table`).
 
 ## Monitoring
 
 ### Active and Completed Exports
 
-Monitor partition exports using the `system.replicated_partition_exports` table:
+Monitor partition exports using the `system.partition_exports` table:
 
 ```sql
-arthur :) select * from system.replicated_partition_exports Format Vertical;
+arthur :) select * from system.partition_exports Format Vertical;
 
 SELECT *
-FROM system.replicated_partition_exports
+FROM system.partition_exports
 FORMAT Vertical
 
 Query id: 9efc271a-a501-44d1-834f-bc4d20156164
@@ -259,28 +277,38 @@ Status values include:
 
 ### Exception columns
 
-- `last_exception_per_replica` is an `Array(Tuple(replica String, message String, part String, time DateTime, count UInt64))`. Each tuple is the most recent exception observed by a single replica plus a best-effort within-replica `count`. Replicas that have never reported an exception are omitted.
+- `last_exception_per_replica` is an `Array(Tuple(replica String, message String, part String, time DateTime, count UInt64))`. Each tuple is the most recent exception observed by a single replica plus a best-effort within-replica `count`. Replicas that have never reported an exception are omitted. A plain `MergeTree` export runs on a single node, so it contributes at most one tuple and its `replica` is empty.
 - `exception_count` is the sum of every `count` in `last_exception_per_replica`. Each replica owns its own counter, so cross-replica updates do not race; the sum is exact w.r.t. the snapshot returned. Within a single replica concurrent failing writers may under-count by one.
 
 ### Per-part destination file paths
 
-- `destination_file_paths` is a `Map(String, Array(String))` keyed by source part name. Each value is the list of file paths written to the destination object storage when that part was exported (a single part can produce multiple files depending on `max_bytes` / `max_rows`). If a refresh cannot read a processed entry from ZooKeeper, the affected key holds the sentinel `<failed to read from zk>` instead of silently under-counting.
+- `destination_file_paths` is a `Map(String, Array(String))` keyed by source part name. Each value is the list of file paths written to the destination object storage when that part was exported (a single part can produce multiple files depending on `max_bytes` / `max_rows`). On a `Replicated*MergeTree` source, if a refresh cannot read a processed entry from ZooKeeper, the affected key holds the sentinel `<failed to read from zk>` instead of silently under-counting.
 
 ### Commit info columns
 
-These columns surface paths produced by the destination storage during commit, so it is possible to inspect what was written without consulting the destination directly:
+These columns surface paths produced by the destination storage during commit, so it is possible to inspect what was written without consulting the destination directly. They are populated for `Replicated*MergeTree` sources only; a plain `MergeTree` does not persist the commit paths, so they stay empty there even after a successful commit.
 
 - `committed_metadata_file` — for Iceberg destinations: path of the new `vN.metadata.json` written by the commit. Empty for non-Iceberg destinations and before the commit lands. If the commit was already finished by a previous run (detected via the transaction id stored in the snapshot summary), this column carries a human-readable sentinel string instead of a path because the original committer's paths are not recoverable from inside the impl.
 - `committed_manifest_list` — for Iceberg destinations: path of the manifest list file (`snap-*.avro`) referenced by the new snapshot. Empty under the same conditions as `committed_metadata_file`.
 - `committed_manifest_file` — for Iceberg destinations: path of the manifest file referenced by `committed_manifest_list`. Empty under the same conditions as `committed_metadata_file`.
 - `committed_marker_file` — for plain object storage destinations: path of the per-transaction commit marker file written by the destination. Empty for Iceberg destinations and for tasks that have not committed yet.
 
+### Columns that depend on the source engine {#columns-that-depend-on-the-source-engine}
+
+Rows for plain `MergeTree` sources share the schema with replicated ones, and leave the columns that only make sense with cross-replica coordination empty:
+
+- `source_replica` — empty, since there is a single node.
+- `last_exception_per_replica` — at most one tuple, with an empty `replica`.
+- `committed_metadata_file`, `committed_manifest_list`, `committed_manifest_file`, `committed_marker_file` — always empty.
+
+`local_backoff_per_part` is local to the node answering the query for both engines.
+
 To pick the latest exception across replicas:
 
 ```sql
 SELECT
     arraySort(x -> -x.time, last_exception_per_replica)[1] AS latest_exception
-FROM system.replicated_partition_exports
+FROM system.partition_exports
 WHERE source_table = 'rmt_table' AND destination_table = 's3_table';
 ```
 

src/Interpreters/InterpreterKillQueryQuery.cpp

@@ -267,7 +267,8 @@ BlockIO InterpreterKillQueryQuery::execute()
 
         Block exports_block = getSelectResult(
             "source_database, source_table, transaction_id, destination_database, destination_table, partition_id",
-            "system.replicated_partition_exports");
+            "system.partition_exports");
+
         if (exports_block.empty())
             return res_io;