Plain merge tree partition export

docs/en/antalya/partition_export.md

@@ -2,11 +2,14 @@
 
 ## 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`. Querying this table reads an in-memory mirror that the manifest-updater refreshes from ZooKeeper on its poll cycle (roughly every 30s); it does not issue a fresh ZooKeeper read per query. 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 can be observed through `system.replicated_partition_exports` (for `Replicated*MergeTree`) or `system.partition_exports` (for plain `MergeTree`). Querying `system.replicated_partition_exports` reads an in-memory mirror that the manifest-updater refreshes from ZooKeeper on its poll cycle (roughly every 30s); it does not issue a fresh ZooKeeper read per query. `system.partition_exports` is served from a local in-memory mirror and does not touch ZooKeeper or disk. Individual part export progress can be observed as usual through `system.exports`.
 
 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.
 
@@ -28,6 +31,18 @@ The Iceberg manifest files contain statistics about the data. Exporting a merge
 
 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
+
+`EXPORT PARTITION` is supported on plain `MergeTree` tables in addition to `Replicated*MergeTree`. Because a single node owns the whole export there is no cross-replica coordination and no ZooKeeper is involved:
+
+- The task descriptor (the list of parts, their per-part progress, the export settings and the task status) is persisted as a small JSON file on the table's disk under `<table_data_path>/partition_exports/<transaction_id>.json`, instead of in ZooKeeper.
+- A local background task drives the export: it schedules the part exports, records their completion, and performs the final commit (a commit file for plain object storage, or the Iceberg snapshot commit for Iceberg destinations).
+- The task is persistent — an in-flight export is reloaded from disk and resumed after a server restart.
+- Progress is observed through `system.partition_exports` (not `system.replicated_partition_exports`). It exposes the same core columns: `source_database`, `source_table`, `destination_database`, `destination_table`, `create_time`, `partition_id`, `transaction_id`, `query_id`, `parts`, `parts_count`, `parts_to_do`, `status`, `last_exception`, `last_exception_part`, `last_exception_time`, and `exception_count`. Replica-specific columns are omitted.
+- Duplicate protection, `export_merge_tree_partition_force_export`, `EXPORT PARTITION ALL`, and `KILL EXPORT PARTITION` all behave the same way as for `Replicated*MergeTree`. A single `KILL EXPORT PARTITION ... WHERE ...` matches tasks from both engines.
+
+All the object-storage and Iceberg output layouts, atomicity guarantees, and settings described in this document apply equally to plain `MergeTree`.
+
 ## Syntax
 
 ```sql
@@ -51,7 +66,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
 
@@ -148,7 +163,7 @@ 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.replicated_partition_exports` table (for `Replicated*MergeTree`) and the `system.partition_exports` table (for plain `MergeTree`); a single `KILL EXPORT PARTITION` consults both. You can use any columns common to those tables in the filter (for example `partition_id`, `source_table`, `destination_table`).
 
 ## Monitoring
 

src/Interpreters/InterpreterKillQueryQuery.cpp

@@ -265,65 +265,119 @@ BlockIO InterpreterKillQueryQuery::execute()
                 "Exporting merge tree partition is experimental. Set the server setting `allow_experimental_export_merge_tree_partition` to enable it");
         }
 
-        Block exports_block = getSelectResult(
-            "source_database, source_table, transaction_id, destination_database, destination_table, partition_id",
-            "system.replicated_partition_exports");
-        if (exports_block.empty())
-            return res_io;
+        const String export_columns = "source_database, source_table, transaction_id, destination_database, destination_table, partition_id";
+
+        /// Partition exports live in two system tables: `replicated_partition_exports` for
+        /// ReplicatedMergeTree and `partition_exports` for plain MergeTree. Query both so a single
+        /// KILL EXPORT PARTITION targets either engine. Each query applies the user WHERE to its own
+        /// table (which exposes all of its columns); a WHERE that references columns specific to the
+        /// other engine will fail against the table that lacks them. We tolerate that per-table so a
+        /// predicate on engine-specific columns still reaches the matching engine — but if BOTH reads
+        /// fail (e.g. a genuinely bad predicate / unknown column), we rethrow so the error surfaces.
+        Block replicated_exports_block;
+        Block plain_exports_block;
+        std::exception_ptr replicated_error;
+        std::exception_ptr plain_error;
+
+        try
+        {
+            replicated_exports_block = getSelectResult(export_columns, "system.replicated_partition_exports");
+        }
+        catch (...)
+        {
+            replicated_error = std::current_exception();
+            tryLogCurrentException(getLogger("InterpreterKillQueryQuery"),
+                "KILL EXPORT PARTITION: could not read system.replicated_partition_exports (the WHERE may "
+                "reference columns that only exist for plain MergeTree); ignoring ReplicatedMergeTree tables");
+        }
+
+        try
+        {
+            plain_exports_block = getSelectResult(export_columns, "system.partition_exports");
+        }
+        catch (...)
+        {
+            plain_error = std::current_exception();
+            tryLogCurrentException(getLogger("InterpreterKillQueryQuery"),
+                "KILL EXPORT PARTITION: could not read system.partition_exports (the WHERE may reference "
+                "columns that only exist for ReplicatedMergeTree); ignoring plain MergeTree tables");
+        }
 
-        const ColumnString & src_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_database").column);
-        const ColumnString & src_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_table").column);
-        const ColumnString & dst_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_database").column);
-        const ColumnString & dst_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_table").column);
-        const ColumnString & tx_col = typeid_cast<const ColumnString &>(*exports_block.getByName("transaction_id").column);
+        /// Neither table could be read: the predicate is invalid for both, so surface the error
+        /// instead of silently matching nothing.
+        if (replicated_error && plain_error)
+            std::rethrow_exception(plain_error);
+
+        if (replicated_exports_block.empty() && plain_exports_block.empty())
+            return res_io;
 
-        auto header = exports_block.cloneEmpty();
+        /// Build the result header explicitly from the fixed projection so it does not depend on
+        /// whether either source block ended up with rows.
+        Block header;
+        for (const auto & column_name : {"source_database", "source_table", "transaction_id",
+                                         "destination_database", "destination_table", "partition_id"})
+            header.insert({ColumnString::create(), std::make_shared<DataTypeString>(), column_name});
         header.insert(0, {ColumnString::create(), std::make_shared<DataTypeString>(), "kill_status"});
 
         MutableColumns res_columns = header.cloneEmptyColumns();
         AccessRightsElements required_access_rights;
         auto access = getContext()->getAccess();
         bool access_denied = false;
 
-        for (size_t i = 0; i < exports_block.rows(); ++i)
+        auto process_block = [&](const Block & exports_block)
         {
-            const auto src_database = src_db_col.getDataAt(i);
-            const auto src_table = src_table_col.getDataAt(i);
-            const auto dst_database = dst_db_col.getDataAt(i);
-            const auto dst_table = dst_table_col.getDataAt(i);
+            if (exports_block.empty())
+                return;
 
-            const auto table_id = StorageID{std::string{src_database}, std::string{src_table}};
-            const auto transaction_id = tx_col.getDataAt(i);
+            const ColumnString & src_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_database").column);
+            const ColumnString & src_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("source_table").column);
+            const ColumnString & dst_db_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_database").column);
+            const ColumnString & dst_table_col = typeid_cast<const ColumnString &>(*exports_block.getByName("destination_table").column);
+            const ColumnString & tx_col = typeid_cast<const ColumnString &>(*exports_block.getByName("transaction_id").column);
 
-            CancellationCode code = CancellationCode::Unknown;
-            if (!query.test)
+            for (size_t i = 0; i < exports_block.rows(); ++i)
             {
-                auto storage = DatabaseCatalog::instance().tryGetTable(table_id, getContext());
-                if (!storage)
-                    code = CancellationCode::NotFound;
-                else
-                {
-                    ASTAlterCommand alter_command{};
-                    alter_command.type = ASTAlterCommand::EXPORT_PARTITION;
-                    alter_command.move_destination_type = DataDestinationType::TABLE;
-                    alter_command.from_database = src_database;
-                    alter_command.from_table = src_table;
-                    alter_command.to_database = dst_database;
-                    alter_command.to_table = dst_table;
+                const auto src_database = src_db_col.getDataAt(i);
+                const auto src_table = src_table_col.getDataAt(i);
+                const auto dst_database = dst_db_col.getDataAt(i);
+                const auto dst_table = dst_table_col.getDataAt(i);
 
-                    required_access_rights = InterpreterAlterQuery::getRequiredAccessForCommand(
-                        alter_command, table_id.database_name, table_id.table_name);
-                    if (!access->isGranted(required_access_rights))
+                const auto table_id = StorageID{std::string{src_database}, std::string{src_table}};
+                const auto transaction_id = tx_col.getDataAt(i);
+
+                CancellationCode code = CancellationCode::Unknown;
+                if (!query.test)
+                {
+                    auto storage = DatabaseCatalog::instance().tryGetTable(table_id, getContext());
+                    if (!storage)
+                        code = CancellationCode::NotFound;
+                    else
                     {
-                        access_denied = true;
-                        continue;
+                        ASTAlterCommand alter_command{};
+                        alter_command.type = ASTAlterCommand::EXPORT_PARTITION;
+                        alter_command.move_destination_type = DataDestinationType::TABLE;
+                        alter_command.from_database = src_database;
+                        alter_command.from_table = src_table;
+                        alter_command.to_database = dst_database;
+                        alter_command.to_table = dst_table;
+
+                        required_access_rights = InterpreterAlterQuery::getRequiredAccessForCommand(
+                            alter_command, table_id.database_name, table_id.table_name);
+                        if (!access->isGranted(required_access_rights))
+                        {
+                            access_denied = true;
+                            continue;
+                        }
+                        code = storage->killExportPartition(std::string{transaction_id});
                     }
-                    code = storage->killExportPartition(std::string{transaction_id});
                 }
+
+                insertResultRow(i, code, exports_block, header, res_columns);
             }
+        };
 
-            insertResultRow(i, code, exports_block, header, res_columns);
-        }
+        process_block(replicated_exports_block);
+        process_block(plain_exports_block);
 
         if (res_columns[0]->empty() && access_denied)
             throw Exception(ErrorCodes::ACCESS_DENIED, "Not allowed to kill export partition. "