Add commit info to partition exports table

docs/en/antalya/partition_export.md

@@ -6,7 +6,7 @@ The `ALTER TABLE EXPORT PARTITION` command exports entire partitions from Replic
 
 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.
 
-The partition export tasks can be observed through `system.replicated_partition_exports`. Querying this table results in a query to ZooKeeper, so it must be used with care. Individual part export progress can be observed as usual through `system.exports`.
+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 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.
 
@@ -105,7 +105,7 @@ TO TABLE [destination_database.]destination_table
 - **Type**: `UInt64`
 - **Default**: `3600`
 - **Description**: The timeout is measured from the manifest's create_time. Set to 0 to disable the timeout.
-When the timeout is exceeded the task transitions to KILLED (same terminal state as `KILL QUERY ... EXPORT PARTITION`), and `last_exception` is populated with a timeout reason.
+When the timeout is exceeded the task transitions to KILLED (same terminal state as `KILL QUERY ... EXPORT PARTITION`), and `last_exception_per_replica` is populated with a timeout reason for the replica that enforced the timeout.
 
 Notes:
 - Enforcement is best-effort: actual kill latency is bounded by one manifest-updater poll cycle (~30s) plus ZooKeeper watch propagation.
@@ -167,41 +167,49 @@ Query id: 9efc271a-a501-44d1-834f-bc4d20156164
 
 Row 1:
 ──────
-source_database:      default
-source_table:         replicated_source
-destination_database: default
-destination_table:    replicated_destination
-create_time:          2025-11-21 18:21:51
-partition_id:         2022
-transaction_id:       7397746091717128192
-source_replica:       r1
-parts:                ['2022_0_0_0','2022_1_1_0','2022_2_2_0']
-parts_count:          3
-parts_to_do:          0
-status:               COMPLETED
-exception_replica:    
-last_exception:       
-exception_part:       
-exception_count:      0
+source_database:            default
+source_table:               replicated_source
+destination_database:       default
+destination_table:          s3_destination
+create_time:                2025-11-21 18:21:51
+partition_id:               2022
+transaction_id:             9b2c1e5a-3f47-4c8e-8a1d-6f0b2d4e7c31
+query_id:                   3fa3c8d3-7d6b-4f8b-9aa2-2c1f1ad0a111
+source_replica:             r1
+parts:                      ['2022_0_0_0','2022_1_1_0','2022_2_2_0']
+parts_count:                3
+parts_to_do:                0
+status:                     COMPLETED
+last_exception_per_replica: []
+exception_count:            0
+destination_file_paths:     {'2022_0_0_0':['data/year=2022/2022_0_0_0_<hash>.parquet'],'2022_1_1_0':['data/year=2022/2022_1_1_0_<hash>.parquet'],'2022_2_2_0':['data/year=2022/2022_2_2_0_<hash>.parquet']}
+committed_metadata_file:
+committed_manifest_list:
+committed_manifest_file:
+committed_marker_file:      data/commit_2022_9b2c1e5a-3f47-4c8e-8a1d-6f0b2d4e7c31
 
 Row 2:
 ──────
-source_database:      default
-source_table:         replicated_source
-destination_database: default
-destination_table:    replicated_destination
-create_time:          2025-11-21 18:20:35
-partition_id:         2021
-transaction_id:       7397745772618674176
-source_replica:       r1
-parts:                ['2021_0_0_0']
-parts_count:          1
-parts_to_do:          0
-status:               COMPLETED
-exception_replica:    
-last_exception:       
-exception_part:       
-exception_count:      0
+source_database:            default
+source_table:               replicated_source
+destination_database:       default
+destination_table:          iceberg_destination
+create_time:                2025-11-21 18:20:35
+partition_id:               2021
+transaction_id:             d0e4f7a2-8c19-4b6d-9e3a-1f5c7b2e9d40
+query_id:                   1c8e0fd0-6a3a-4d6e-9bd6-bdf64adfe118
+source_replica:             r2
+parts:                      ['2021_0_0_0']
+parts_count:                1
+parts_to_do:                0
+status:                     COMPLETED
+last_exception_per_replica: [('r1','Code: 999. Coordination::Exception: Session expired','2021_0_0_0','2025-11-21 18:20:42',1)]
+exception_count:            1
+destination_file_paths:     {'2021_0_0_0':['data/year=2021/2021_0_0_0_<hash>.parquet']}
+committed_metadata_file:    data/metadata/v3.metadata.json
+committed_manifest_list:    data/metadata/snap-4029103741930112856-1-<uuid>.avro
+committed_manifest_file:    data/metadata/<uuid>-m0.avro
+committed_marker_file:
 
 2 rows in set. Elapsed: 0.019 sec. 
 
@@ -214,6 +222,33 @@ Status values include:
 - `FAILED` - Export failed
 - `KILLED` - Export was cancelled
 
+### 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.
+- `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.
+
+### 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:
+
+- `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.
+
+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
+WHERE source_table = 'rmt_table' AND destination_table = 's3_table';
+```
+
 ## Related Features
 
 - [ALTER TABLE EXPORT PART](/docs/en/engines/table-engines/mergetree-family/part_export.md) - Export individual parts (non-replicated)

src/Common/FailPoint.cpp

@@ -135,6 +135,7 @@ static struct InitFiu
     ONCE(iceberg_export_after_commit_before_zk_completed) \
     REGULAR(export_partition_commit_always_throw) \
     ONCE(export_partition_status_change_throw) \
+    REGULAR(export_partition_processed_paths_sync_fail) \
     REGULAR(export_part_non_retryable_throw) \
     REGULAR(export_part_retryable_throw) \
     ONCE(backup_add_empty_memory_table) \

src/Storages/ExportReplicatedMergeTreePartitionManifest.h

@@ -7,6 +7,7 @@
 #include <Poco/JSON/Array.h>
 #include <Poco/JSON/Parser.h>
 #include <Storages/MergeTree/MergeTreePartExportManifest.h>
+#include <optional>
 
 namespace DB
 {
@@ -149,6 +150,74 @@ struct ExportReplicatedMergeTreePartitionProcessedPartEntry
     }
 };
 
+/// Per-task "commit info" record persisted at <export-entry>/commit_info.
+///
+/// Written exactly once, atomically with the status -> COMPLETED transition
+/// (see ExportPartitionUtils::commit). Captures the metadata-layer file paths
+/// produced by the destination storage during commit so they can be surfaced in
+/// system.replicated_partition_exports for debugging.
+///
+/// All Iceberg fields are empty for non-Iceberg destinations. They may also be
+/// empty for an Iceberg destination if the committing replica crashed between
+/// writing the object-storage files and writing this znode; in that case the
+/// task still transitions to COMPLETED via the recovery path but commit_info
+/// remains absent. This is best-effort observability and acceptable.
+struct ExportReplicatedMergeTreePartitionCommitInfoEntry
+{
+    /// Iceberg: path (in destination object storage) of the new vN.metadata.json
+    /// written by the commit.
+    String iceberg_metadata_file;
+
+    /// Iceberg: path of the snap-<id>-<format_version>-<uuid>.avro manifest list
+    /// referenced by the new snapshot.
+    String iceberg_manifest_list;
+
+    /// Iceberg: path of the manifest entry file (*.avro) referenced by the
+    /// manifest list.
+    String iceberg_manifest_file;
+
+    /// Plain object storage: path of the commit marker file written by
+    /// StorageObjectStorage::commitExportPartitionTransaction. Empty for Iceberg.
+    String commit_marker_file;
+
+    std::string toJsonString() const
+    {
+        Poco::JSON::Object json;
+        json.set("iceberg_metadata_file", iceberg_metadata_file);
+        json.set("iceberg_manifest_list", iceberg_manifest_list);
+        json.set("iceberg_manifest_file", iceberg_manifest_file);
+        json.set("commit_marker_file", commit_marker_file);
+
+        std::ostringstream oss;     // STYLE_CHECK_ALLOW_STD_STRING_STREAM
+        oss.exceptions(std::ios::failbit);
+        Poco::JSON::Stringifier::stringify(json, oss);
+        return oss.str();
+    }
+
+    static ExportReplicatedMergeTreePartitionCommitInfoEntry fromJsonString(const std::string & json_string)
+    {
+        ExportReplicatedMergeTreePartitionCommitInfoEntry entry;
+        if (json_string.empty())
+            return entry;
+
+        Poco::JSON::Parser parser;
+        auto json = parser.parse(json_string).extract<Poco::JSON::Object::Ptr>();
+
+        if (json->has("iceberg_metadata_file"))
+            entry.iceberg_metadata_file = json->getValue<String>("iceberg_metadata_file");
+        if (json->has("iceberg_manifest_list"))
+            entry.iceberg_manifest_list = json->getValue<String>("iceberg_manifest_list");
+
+        if (json->has("iceberg_manifest_file"))
+            entry.iceberg_manifest_file = json->getValue<String>("iceberg_manifest_file");
+
+        if (json->has("commit_marker_file"))
+            entry.commit_marker_file = json->getValue<String>("commit_marker_file");
+
+        return entry;
+    }
+};
+
 struct ExportReplicatedMergeTreePartitionManifest
 {
     String transaction_id;
@@ -173,10 +242,12 @@ struct ExportReplicatedMergeTreePartitionManifest
     bool write_full_path_in_iceberg_metadata = false;
     bool allow_lossy_cast = false;
     String iceberg_metadata_json;
-    String parquet_compression_method;
-    UInt64 output_format_compression_level;
-    UInt64 parquet_row_group_size;
-    UInt64 parquet_row_group_size_bytes;
+
+    /// Optional because of backwards compatibility
+    std::optional<String> parquet_compression_method;
+    std::optional<UInt64> output_format_compression_level;
+    std::optional<UInt64> parquet_row_group_size;
+    std::optional<UInt64> parquet_row_group_size_bytes;
 
     std::string toJsonString() const
     {
@@ -211,10 +282,14 @@ struct ExportReplicatedMergeTreePartitionManifest
         json.set("task_timeout_seconds", task_timeout_seconds);
         json.set("write_full_path_in_iceberg_metadata", write_full_path_in_iceberg_metadata);
         json.set("allow_lossy_cast", allow_lossy_cast);
-        json.set("parquet_compression_method", parquet_compression_method);
-        json.set("output_format_compression_level", output_format_compression_level);
-        json.set("parquet_row_group_size", parquet_row_group_size);
-        json.set("parquet_row_group_size_bytes", parquet_row_group_size_bytes);
+        if (parquet_compression_method)
+            json.set("parquet_compression_method", *parquet_compression_method);
+        if (output_format_compression_level)
+            json.set("output_format_compression_level", *output_format_compression_level);
+        if (parquet_row_group_size)
+            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);
         std::ostringstream oss;     // STYLE_CHECK_ALLOW_STD_STRING_STREAM
         oss.exceptions(std::ios::failbit);
         Poco::JSON::Stringifier::stringify(json, oss);
@@ -282,10 +357,26 @@ 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;
 
-        manifest.parquet_compression_method = json->getValue<String>("parquet_compression_method");
-        manifest.output_format_compression_level = json->getValue<UInt64>("output_format_compression_level");
-        manifest.parquet_row_group_size = json->getValue<UInt64>("parquet_row_group_size");
-        manifest.parquet_row_group_size_bytes = json->getValue<UInt64>("parquet_row_group_size_bytes");
+
+        if (json->has("parquet_compression_method"))
+        {
+            manifest.parquet_compression_method = json->getValue<String>("parquet_compression_method");
+        }
+
+        if (json->has("output_format_compression_level"))
+        {
+            manifest.output_format_compression_level = json->getValue<UInt64>("output_format_compression_level");
+        }
+
+        if (json->has("parquet_row_group_size"))
+        {
+            manifest.parquet_row_group_size = json->getValue<UInt64>("parquet_row_group_size");
+        }
+
+        if (json->has("parquet_row_group_size_bytes"))
+        {
+            manifest.parquet_row_group_size_bytes = json->getValue<UInt64>("parquet_row_group_size_bytes");
+        }
 
         return manifest;
     }

src/Storages/ExportReplicatedMergeTreePartitionTaskEntry.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <map>
+#include <optional>
 #include <Storages/ExportReplicatedMergeTreePartitionManifest.h>
 #include <Storages/MergeTree/IMergeTreeDataPart.h>
 #include "Core/QualifiedTableName.h"
@@ -40,6 +41,23 @@ struct ExportReplicatedMergeTreePartitionTaskEntry
     /// An empty map means no replica has recorded an exception yet for this task.
     mutable std::map<String, LastExceptionEntry> last_exception_per_replica;
 
+    /// In-memory mirror of <export-entry>/processed/<part> leaves in ZK, keyed by
+    /// part name. Each value is the list of destination file paths produced by the
+    /// per-part export (typically Parquet object-storage keys). Refreshed on every
+    /// poll() cycle and on status-change handler invocations; served verbatim to
+    /// system.replicated_partition_exports without any extra ZK read at query time.
+    /// An empty map means no part has finished exporting yet for this task.
+    /// Incomplete Keeper refreshes (or unreadable processed leaves) publish
+    /// "<failed to read from zk>" as a whole-map key, or as the sole path value
+    /// for the affected part leaf.
+    mutable std::map<String, std::vector<String>> destination_file_paths_per_part;
+
+    /// In-memory mirror of the <export-entry>/commit_info znode (written atomically
+    /// with the COMPLETED status transition; see ExportPartitionUtils::commit).
+    /// nullopt until commit_info is observed in ZK. Empty fields inside the struct
+    /// for non-Iceberg destinations.
+    mutable std::optional<ExportReplicatedMergeTreePartitionCommitInfoEntry> commit_info;
+
     std::string getCompositeKey() const
     {
         const auto qualified_table_name = QualifiedTableName {manifest.destination_database, manifest.destination_table};

src/Storages/IStorage.h

@@ -511,7 +511,21 @@ It is currently only implemented in StorageObjectStorage.
       Block partition_source_block;
     };
 
-    virtual void commitExportPartitionTransaction(
+    /// Paths produced by the destination storage during commit. Surfaced via
+    /// system.replicated_partition_exports for debugging
+    struct ExportPartitionCommitInfo
+    {
+      /// Iceberg destinations only.
+      String iceberg_metadata_file;
+      String iceberg_manifest_list;
+      String iceberg_manifest_file;
+
+      /// Plain object storage destinations only: path of the commit marker file
+      /// written/observed by StorageObjectStorage::commitExportPartitionTransaction.
+      String commit_marker_file;
+    };
+
+    virtual ExportPartitionCommitInfo commitExportPartitionTransaction(
       const String & /* transaction_id */,
       const String & /* partition_id */,
       const Strings & /* exported_paths */,