diff --git a/docs/en/antalya/partition_export.md b/docs/en/antalya/partition_export.md index 9a964af11581..93eb559c8bc4 100644 --- a/docs/en/antalya/partition_export.md +++ b/docs/en/antalya/partition_export.md @@ -167,37 +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 +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 +exception_count: 0 +destination_file_paths: {'2022_0_0_0':['data/year=2022/2022_0_0_0_.parquet'],'2022_1_1_0':['data/year=2022/2022_1_1_0_.parquet'],'2022_2_2_0':['data/year=2022/2022_2_2_0_.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 -last_exception_per_replica: [] -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_.parquet']} +committed_metadata_file: data/metadata/v3.metadata.json +committed_manifest_list: data/metadata/snap-4029103741930112856-1-.avro +committed_manifest_file: data/metadata/-m0.avro +committed_marker_file: 2 rows in set. Elapsed: 0.019 sec. @@ -215,6 +227,19 @@ Status values include: - `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 `` 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 diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index 60b516e377e0..7693560c8e44 100644 --- a/src/Common/FailPoint.cpp +++ b/src/Common/FailPoint.cpp @@ -170,6 +170,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) \ diff --git a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h index 49da8eff4895..d58e9d534949 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionManifest.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionManifest.h @@ -7,6 +7,7 @@ #include #include #include +#include namespace DB { @@ -149,6 +150,74 @@ struct ExportReplicatedMergeTreePartitionProcessedPartEntry } }; +/// Per-task "commit info" record persisted at /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---.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(); + + if (json->has("iceberg_metadata_file")) + entry.iceberg_metadata_file = json->getValue("iceberg_metadata_file"); + if (json->has("iceberg_manifest_list")) + entry.iceberg_manifest_list = json->getValue("iceberg_manifest_list"); + + if (json->has("iceberg_manifest_file")) + entry.iceberg_manifest_file = json->getValue("iceberg_manifest_file"); + + if (json->has("commit_marker_file")) + entry.commit_marker_file = json->getValue("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 parquet_compression_method; + std::optional output_format_compression_level; + std::optional parquet_row_group_size; + std::optional 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("allow_lossy_cast") : true; - manifest.parquet_compression_method = json->getValue("parquet_compression_method"); - manifest.output_format_compression_level = json->getValue("output_format_compression_level"); - manifest.parquet_row_group_size = json->getValue("parquet_row_group_size"); - manifest.parquet_row_group_size_bytes = json->getValue("parquet_row_group_size_bytes"); + + if (json->has("parquet_compression_method")) + { + manifest.parquet_compression_method = json->getValue("parquet_compression_method"); + } + + if (json->has("output_format_compression_level")) + { + manifest.output_format_compression_level = json->getValue("output_format_compression_level"); + } + + if (json->has("parquet_row_group_size")) + { + manifest.parquet_row_group_size = json->getValue("parquet_row_group_size"); + } + + if (json->has("parquet_row_group_size_bytes")) + { + manifest.parquet_row_group_size_bytes = json->getValue("parquet_row_group_size_bytes"); + } return manifest; } diff --git a/src/Storages/ExportReplicatedMergeTreePartitionTaskEntry.h b/src/Storages/ExportReplicatedMergeTreePartitionTaskEntry.h index 8af873e0b89c..36c0ef303fbf 100644 --- a/src/Storages/ExportReplicatedMergeTreePartitionTaskEntry.h +++ b/src/Storages/ExportReplicatedMergeTreePartitionTaskEntry.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #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 last_exception_per_replica; + /// In-memory mirror of /processed/ 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 + /// "" as a whole-map key, or as the sole path value + /// for the affected part leaf. + mutable std::map> destination_file_paths_per_part; + + /// In-memory mirror of the /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 commit_info; + std::string getCompositeKey() const { const auto qualified_table_name = QualifiedTableName {manifest.destination_database, manifest.destination_table}; diff --git a/src/Storages/IStorage.h b/src/Storages/IStorage.h index 6ece044156a2..a77f425b9f75 100644 --- a/src/Storages/IStorage.h +++ b/src/Storages/IStorage.h @@ -479,7 +479,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 */, diff --git a/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.cpp b/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.cpp index f020f6da6386..077f06349c5d 100644 --- a/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.cpp +++ b/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.cpp @@ -5,11 +5,13 @@ #include "Common/logger_useful.h" #include #include +#include #include #include #include #include #include +#include namespace ProfileEvents { @@ -33,10 +35,16 @@ namespace ErrorCodes namespace FailPoints { extern const char export_partition_status_change_throw[]; + extern const char export_partition_processed_paths_sync_fail[]; } namespace { + /// Value published into destination_file_paths when a processed/ Keeper refresh + /// is incomplete (or a leaf is unreadable), so system.replicated_partition_exports + /// can show that the in-memory mirror failed to sync instead of silently under-counting. + constexpr std::string_view zk_sync_failed_marker = ""; + /// Describes pending commits struct CommitRecoveryWork { @@ -48,7 +56,7 @@ namespace /// Fetch all per-replica last_exception leaves under /last_exception and build /// a fresh map keyed by replica name. - std::map readLastExceptionPerReplica( + std::optional> readLastExceptionPerReplica( const zkutil::ZooKeeperPtr & zk, const std::filesystem::path & entry_path, const std::string & log_key, @@ -64,7 +72,7 @@ namespace if (Coordination::Error::ZOK != zk->tryGetChildren(container_path, children)) { LOG_WARNING(log, "ExportPartition Manifest Updating Task: failed to list last_exception leaves for {}, leaving in-memory copy untouched", log_key); - return out; + return std::nullopt; } if (children.empty()) @@ -115,6 +123,139 @@ namespace return out; } + std::map> readDestinationFilePathsPerPart( + const zkutil::ZooKeeperPtr & zk, + const std::filesystem::path & entry_path, + const std::string & log_key, + const LoggerPtr & log) + { + std::map> out; + + const auto container_path = entry_path / "processed"; + + Strings children; + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests); + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGetChildren); + if (Coordination::Error::ZOK != zk->tryGetChildren(container_path, children)) + { + LOG_INFO(log, "ExportPartition Manifest Updating Task: failed to list processed leaves for {}, publishing sync-failed marker", log_key); + out.emplace(String(zk_sync_failed_marker), std::vector{String(zk_sync_failed_marker)}); + return out; + } + + if (children.empty()) + return out; + + std::vector paths; + paths.reserve(children.size()); + for (const auto & child : children) + paths.emplace_back(container_path / child); + + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests); + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGet, paths.size()); + auto responses = zk->tryGet(paths); + responses.waitForResponses(); + + for (size_t i = 0; i < paths.size(); ++i) + { + Coordination::GetResponse response; + try + { + /// Simulate a non-ZNONODE multi-get failure so the catch path below + /// publishes the sync-failed marker (same shape as operator[] rethrow). + fiu_do_on(FailPoints::export_partition_processed_paths_sync_fail, + { + throw zkutil::KeeperException(Coordination::Error::ZCONNECTIONLOSS); + }); + response = responses[i]; + } + catch (...) + { + LOG_WARNING(log, "ExportPartition Manifest Updating Task: ZK error fetching processed leaf {} for {}, publishing sync-failed marker", children[i], log_key); + out.emplace(children[i], std::vector{String(zk_sync_failed_marker)}); + continue; + } + + if (response.error != Coordination::Error::ZOK) + { + LOG_WARNING(log, "ExportPartition Manifest Updating Task: could not read processed leaf {} for {} (error {}), publishing sync-failed marker", children[i], log_key, response.error); + out.emplace(children[i], std::vector{String(zk_sync_failed_marker)}); + continue; + } + + try + { + auto entry = ExportReplicatedMergeTreePartitionProcessedPartEntry::fromJsonString(response.data); + out.emplace(std::move(entry.part_name), std::move(entry.paths_in_destination)); + } + catch (...) + { + LOG_WARNING(log, "ExportPartition Manifest Updating Task: malformed processed JSON for {} (leaf {}), publishing sync-failed marker", log_key, children[i]); + out.emplace(children[i], std::vector{String(zk_sync_failed_marker)}); + } + } + + return out; + } + + /// True when the cached `/processed` mirror carries a `zk_sync_failed_marker` sentinel, + /// published whenever a listing or a leaf read/parse failed. Such a mirror is incomplete + /// and must be refreshed again on the next poll. + bool destinationFilePathsMirrorHasSyncFailure(const std::map> & cached_paths) + { + for (const auto & [part_name, destination_paths] : cached_paths) + { + if (part_name == zk_sync_failed_marker) + return true; + for (const auto & destination_path : destination_paths) + if (destination_path == zk_sync_failed_marker) + return true; + } + return false; + } + + bool skipReadingDestinationFilePaths( + ExportReplicatedMergeTreePartitionTaskEntry::Status status, + const std::map> & cached_paths, + size_t number_of_parts) + { + if (status == ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING) + return false; + if (destinationFilePathsMirrorHasSyncFailure(cached_paths)) + return false; + if (status == ExportReplicatedMergeTreePartitionTaskEntry::Status::COMPLETED) + return cached_paths.size() == number_of_parts; + return true; + } + + /// Read the optional /commit_info znode and return the parsed entry. + /// Returns nullopt when the znode is absent (task has not committed yet, peer + /// crashed before writing it, or transient ZK error). Callers should treat + /// nullopt as "leave the in-memory copy untouched". + std::optional readCommitInfo( + const zkutil::ZooKeeperPtr & zk, + const std::filesystem::path & entry_path, + const std::string & log_key, + const LoggerPtr & log) + { + const auto commit_info_path = entry_path / "commit_info"; + + std::string data; + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests); + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGet); + if (!zk->tryGet(commit_info_path, data)) + return std::nullopt; + + try + { + return ExportReplicatedMergeTreePartitionCommitInfoEntry::fromJsonString(data); + } + catch (...) + { + LOG_WARNING(log, "ExportPartition Manifest Updating Task: malformed commit_info JSON for {}, ignoring", log_key); + return std::nullopt; + } + } /// collects pending commits and kills tasks that have timed out void tryCleanup( @@ -287,6 +428,16 @@ std::vector ExportPartitionManifestUpdatingTask:: } info.exception_count = total_exception_count; + info.destination_file_paths_per_part = entry.destination_file_paths_per_part; + + if (entry.commit_info) + { + info.committed_metadata_file = entry.commit_info->iceberg_metadata_file; + info.committed_manifest_list = entry.commit_info->iceberg_manifest_list; + info.committed_manifest_file = entry.commit_info->iceberg_manifest_file; + info.committed_marker_file = entry.commit_info->commit_marker_file; + } + if (const auto it = backoff.find(entry.getTransactionId()); it != backoff.end()) { info.backoff_per_part.reserve(it->second.size()); @@ -309,6 +460,7 @@ void ExportPartitionManifestUpdatingTask::poll() std::vector deferred_commits; auto zk = storage.getZooKeeper(); + const auto log = storage.log.load(); const std::string exports_path = fs::path(storage.zookeeper_path) / "exports"; const std::string cleanup_lock_path = fs::path(storage.zookeeper_path) / "exports_cleanup_lock"; @@ -321,7 +473,7 @@ void ExportPartitionManifestUpdatingTask::poll() auto cleanup_lock = zkutil::EphemeralNodeHolder::tryCreate(cleanup_lock_path, *zk, storage.replica_name); if (cleanup_lock) { - LOG_DEBUG(storage.log, "ExportPartition Manifest Updating Task: Cleanup lock acquired, will remove stale entries"); + LOG_DEBUG(log, "ExportPartition Manifest Updating Task: Cleanup lock acquired, will remove stale entries"); } { @@ -338,7 +490,7 @@ void ExportPartitionManifestUpdatingTask::poll() auto & entries_by_key = working_model->get(); - LOG_DEBUG(storage.log, "ExportPartition Manifest Updating Task: Polling for new entries for table {}. Current number of entries: {}", storage.getStorageID().getNameForLogs(), entries_by_key.size()); + LOG_DEBUG(log, "ExportPartition Manifest Updating Task: Polling for new entries for table {}. Current number of entries: {}", storage.getStorageID().getNameForLogs(), entries_by_key.size()); ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests); ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGetChildrenWatch); @@ -361,7 +513,7 @@ void ExportPartitionManifestUpdatingTask::poll() std::string metadata_json; if (!zk->tryGet(fs::path(entry_path) / "metadata.json", metadata_json)) { - LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: missing metadata.json", key); + LOG_WARNING(log, "ExportPartition Manifest Updating Task: Skipping {}: missing metadata.json", key); continue; } @@ -375,13 +527,13 @@ void ExportPartitionManifestUpdatingTask::poll() /// A single unparseable metadata.json (e.g. genuinely corrupt, or written by a /// future incompatible format) must not abort the whole poll and stall discovery, /// cleanup and status convergence for every other task. Skip just this entry. - tryLogCurrentException(storage.log, __PRETTY_FUNCTION__); - LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: could not parse metadata.json", key); + tryLogCurrentException(log, __PRETTY_FUNCTION__); + LOG_WARNING(log, "ExportPartition Manifest Updating Task: Skipping {}: could not parse metadata.json", key); continue; } auto last_exception_per_replica = readLastExceptionPerReplica( - zk, fs::path(entry_path), key, storage.log.load()); + zk, fs::path(entry_path), key, log); /// If the zk entry has been replaced with export_merge_tree_partition_force_export, checking only for the export key is not enough /// we need to make sure it is the same transaction id. If it is not, it needs to be replaced. @@ -424,17 +576,26 @@ void ExportPartitionManifestUpdatingTask::poll() if (status_string.empty()) { - LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: missing status", key); + LOG_WARNING(log, "ExportPartition Manifest Updating Task: Skipping {}: missing status", key); continue; } const auto status = magic_enum::enum_cast(status_string); if (!status) { - LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", status_string, key); + LOG_WARNING(log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", status_string, key); continue; } + const bool skip_processed_refresh = + has_local_entry + && skipReadingDestinationFilePaths(*status, local_entry->destination_file_paths_per_part, metadata.number_of_parts); + + std::optional>> destination_file_paths_per_part; + if (!skip_processed_refresh) + destination_file_paths_per_part = readDestinationFilePathsPerPart( + zk, fs::path(entry_path), key, log); + /// If we hold the cleanup lock, enforce the task timeout and recover uncommitted exports. /// Entries are never removed here, so we always fall through to refresh / addTask below. if (cleanup_lock) @@ -442,7 +603,7 @@ void ExportPartitionManifestUpdatingTask::poll() tryCleanup( zk, entry_path, - storage.log.load(), + log, storage.getContext(), storage, metadata, @@ -453,34 +614,47 @@ void ExportPartitionManifestUpdatingTask::poll() if (!has_local_entry) { - addTask(metadata, *status, std::move(last_exception_per_replica), key, entries_by_key); - LOG_INFO(storage.log, "ExportPartition Manifest Updating Task: Added new entry for task {}", key); + addTask( + metadata, + *status, + last_exception_per_replica ? std::move(*last_exception_per_replica) : std::map{}, + destination_file_paths_per_part ? std::move(*destination_file_paths_per_part) : std::map>{}, + readCommitInfo(zk, fs::path(entry_path), key, log), + key, + entries_by_key); + LOG_INFO(log, "ExportPartition Manifest Updating Task: Added new entry for task {}", key); continue; } - /// If we already have the local entry, we need to update it if the status has changed or if there are new last exceptions. + if (!local_entry->commit_info && *status == ExportReplicatedMergeTreePartitionTaskEntry::Status::COMPLETED) + { + local_entry->commit_info = readCommitInfo(zk, fs::path(entry_path), key, log); + } + + /// If we already have the local entry, we need to update it + if (last_exception_per_replica) + local_entry->last_exception_per_replica = std::move(*last_exception_per_replica); + if (destination_file_paths_per_part) + local_entry->destination_file_paths_per_part = std::move(*destination_file_paths_per_part); + const bool status_changed = local_entry->status != *status; - if (!last_exception_per_replica.empty() || status_changed) + if (status_changed) { - if (!last_exception_per_replica.empty()) - local_entry->last_exception_per_replica = std::move(last_exception_per_replica); - if (status_changed) + local_entry->status = *status; + if (local_entry->status != ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING) { - local_entry->status = *status; - if (local_entry->status != ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING) + /// terminal now - we no longer need to keep the data parts alive + local_entry->part_references.clear(); + + /// looks like we missed a status change event, we should kill local operations. + if (local_entry->status == ExportReplicatedMergeTreePartitionTaskEntry::Status::KILLED) { - /// terminal now - we no longer need to keep the data parts alive - local_entry->part_references.clear(); - - /// looks like we missed a status change event, we should kill local operations. - if (local_entry->status == ExportReplicatedMergeTreePartitionTaskEntry::Status::KILLED) - { - storage.killExportPart(local_entry->manifest.transaction_id); - } + storage.killExportPart(local_entry->manifest.transaction_id); } } } - LOG_DEBUG(storage.log, "ExportPartition Manifest Updating Task: Skipping {}: already exists", key); + + LOG_DEBUG(log, "ExportPartition Manifest Updating Task: Skipping {}: already exists", key); } @@ -492,22 +666,20 @@ void ExportPartitionManifestUpdatingTask::poll() /// `entries_by_key` (a reference into it) must not be used afterwards. storage.export_partition_manifests.set(std::move(working_model)); - LOG_DEBUG(storage.log, "ExportPartition Manifest Updating task: finished polling for new entries. Number of entries: {}", entries_count); + LOG_DEBUG(log, "ExportPartition Manifest Updating task: finished polling for new entries. Number of entries: {}", entries_count); } - const auto log_ptr = storage.log.load(); - /// Execute pending commits for (const auto & work : deferred_commits) { /// A replica exported the last part but the commit never landed. Try to fix it. try { - ExportPartitionUtils::commit(work.metadata, work.destination_storage, zk, log_ptr, work.entry_path, work.context, storage, storage.getReplicaName()); + ExportPartitionUtils::commit(work.metadata, work.destination_storage, zk, log, work.entry_path, work.context, storage, storage.getReplicaName()); } catch (const Exception & e) { - LOG_WARNING(log_ptr, + LOG_WARNING(log, "ExportPartition Manifest Updating Task: " "Caught exception while committing export for {}: {}", work.entry_path, e.message()); @@ -518,11 +690,11 @@ void ExportPartitionManifestUpdatingTask::poll() e.code(), storage.getReplicaName(), e.message(), - log_ptr); + log); if (became_failed) { - LOG_WARNING(log_ptr, + LOG_WARNING(log, "ExportPartition Manifest Updating Task: " "Commit for {} transitioned to FAILED due to non-retryable error (code {})", work.entry_path, e.code()); @@ -537,6 +709,8 @@ void ExportPartitionManifestUpdatingTask::addTask( const ExportReplicatedMergeTreePartitionManifest & metadata, ExportReplicatedMergeTreePartitionTaskEntry::Status status, std::map last_exception_per_replica, + std::map> destination_file_paths_per_part, + std::optional commit_info, const std::string & key, auto & entries_by_key ) @@ -559,12 +733,32 @@ void ExportPartitionManifestUpdatingTask::addTask( } /// Called from poll() under M_task (sole mutator), so no extra locking is required. - ExportReplicatedMergeTreePartitionTaskEntry entry {metadata, status, std::move(part_references), std::move(last_exception_per_replica)}; + ExportReplicatedMergeTreePartitionTaskEntry entry { + metadata, + status, + std::move(part_references), + std::move(last_exception_per_replica), + std::move(destination_file_paths_per_part), + std::move(commit_info)}; + auto it = entries_by_key.find(key); if (it != entries_by_key.end()) - entries_by_key.replace(it, entry); - else - entries_by_key.insert(entry); + { + if (!entries_by_key.replace(it, entry)) + LOG_ERROR(storage.log, + "ExportPartition Manifest Updating Task: failed to replace in-memory entry for {} (transaction_id {}). " + "This most likely means another export already holds the same transaction_id (id collision); " + "this export will be missing from system.replicated_partition_exports.", + key, entry.getTransactionId()); + } + else if (!entries_by_key.insert(entry).second) + { + LOG_ERROR(storage.log, + "ExportPartition Manifest Updating Task: failed to insert in-memory entry for {} (transaction_id {}). " + "Another entry already holds this transaction_id (id collision); " + "this export will be invisible in system.replicated_partition_exports.", + key, entry.getTransactionId()); + } } void ExportPartitionManifestUpdatingTask::removeStaleEntries( @@ -613,6 +807,7 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges() /// Take a snapshot of all status changes. If an exception is thrown, we will requeue the whole batch. const std::queue batch = local_status_changes; + const auto log = storage.log.load(); try { @@ -625,7 +820,7 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges() const bool had_changes = !local_status_changes.empty(); - LOG_DEBUG(storage.log, "ExportPartition Manifest Updating task: handling status changes. Number of status changes: {}", local_status_changes.size()); + LOG_DEBUG(log, "ExportPartition Manifest Updating task: handling status changes. Number of status changes: {}", local_status_changes.size()); const auto current_model = storage.export_partition_manifests.get(); auto working_model = current_model @@ -636,7 +831,7 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges() while (!local_status_changes.empty()) { const auto & key = local_status_changes.front(); - LOG_INFO(storage.log, "ExportPartition Manifest Updating task: handling status change for task {}", key); + LOG_INFO(log, "ExportPartition Manifest Updating task: handling status change for task {}", key); fiu_do_on(FailPoints::export_partition_status_change_throw, { @@ -651,13 +846,15 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges() continue; } + const auto export_path = fs::path(storage.zookeeper_path) / "exports" / key; + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests); ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperGet); /// get new status from zk std::string new_status_string; - if (!zk->tryGet(fs::path(storage.zookeeper_path) / "exports" / key / "status", new_status_string)) + if (!zk->tryGet(export_path / "status", new_status_string)) { - LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Failed to get new status for task {}, skipping", key); + LOG_WARNING(log, "ExportPartition Manifest Updating Task: Failed to get new status for task {}, skipping", key); local_status_changes.pop(); continue; } @@ -665,33 +862,46 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges() const auto new_status = magic_enum::enum_cast(new_status_string); if (!new_status) { - LOG_WARNING(storage.log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", new_status_string, key); + LOG_WARNING(log, "ExportPartition Manifest Updating Task: Invalid status {} for task {}, skipping", new_status_string, key); local_status_changes.pop(); continue; } - LOG_INFO(storage.log, "ExportPartition Manifest Updating task: status changed for task {}. New status: {}", key, magic_enum::enum_name(*new_status).data()); + LOG_INFO(log, "ExportPartition Manifest Updating task: status changed for task {}. New status: {}", key, magic_enum::enum_name(*new_status).data()); auto fetched = readLastExceptionPerReplica( - zk, fs::path(storage.zookeeper_path) / "exports" / key, key, storage.log.load()); + zk, export_path, key, log); + + if (!skipReadingDestinationFilePaths(*new_status, it->destination_file_paths_per_part, it->manifest.number_of_parts)) + { + auto destination_file_paths_per_part = readDestinationFilePathsPerPart( + zk, export_path, key, log); + it->destination_file_paths_per_part = std::move(destination_file_paths_per_part); + } + + if (*new_status == ExportReplicatedMergeTreePartitionTaskEntry::Status::COMPLETED) + { + if (auto fetched_commit_info = readCommitInfo(zk, export_path, key, log)) + it->commit_info = std::move(fetched_commit_info); + } - /// If status changed to KILLED, cancel local export operations. + /// If status changed to KILLED, cancel local export operations if (*new_status == ExportReplicatedMergeTreePartitionTaskEntry::Status::KILLED) { try { - LOG_INFO(storage.log, "ExportPartition Manifest Updating task: killing export partition for task {}", key); + LOG_INFO(log, "ExportPartition Manifest Updating task: killing export partition for task {}", key); storage.killExportPart(it->manifest.transaction_id); } catch (...) { - tryLogCurrentException(storage.log, __PRETTY_FUNCTION__); + tryLogCurrentException(log, __PRETTY_FUNCTION__); } } /// Apply the in-memory updates directly (poll() cannot run concurrently under M_task). - if (!fetched.empty()) - it->last_exception_per_replica = std::move(fetched); + if (fetched) + it->last_exception_per_replica = std::move(*fetched); it->status = *new_status; @@ -711,9 +921,9 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges() } catch (...) { - tryLogCurrentException(storage.log, __PRETTY_FUNCTION__); + tryLogCurrentException(log, __PRETTY_FUNCTION__); - LOG_WARNING(storage.log, "ExportPartition Manifest Updating task: exception thrown while handling status changes; nothing was published, requeuing the whole batch. Batch size: {}", batch.size()); + LOG_WARNING(log, "ExportPartition Manifest Updating task: exception thrown while handling status changes; nothing was published, requeuing the whole batch. Batch size: {}", batch.size()); std::lock_guard lock(status_changes_mutex); @@ -730,7 +940,7 @@ void ExportPartitionManifestUpdatingTask::handleStatusChanges() std::swap(status_changes, requeued); } - LOG_DEBUG(storage.log, "ExportPartition Manifest Updating task: pending status changes after requeue: {}", status_changes.size()); + LOG_DEBUG(log, "ExportPartition Manifest Updating task: pending status changes after requeue: {}", status_changes.size()); throw; } diff --git a/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.h b/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.h index 3bb4e7ac92ab..129629f6ed9b 100644 --- a/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.h +++ b/src/Storages/MergeTree/ExportPartitionManifestUpdatingTask.h @@ -34,6 +34,8 @@ class ExportPartitionManifestUpdatingTask const ExportReplicatedMergeTreePartitionManifest & metadata, ExportReplicatedMergeTreePartitionTaskEntry::Status status, std::map last_exception_per_replica, + std::map> destination_file_paths_per_part, + std::optional commit_info, const std::string & key, auto & entries_by_key ); diff --git a/src/Storages/MergeTree/ExportPartitionUtils.cpp b/src/Storages/MergeTree/ExportPartitionUtils.cpp index 5d064373aaf7..942b2e454f01 100644 --- a/src/Storages/MergeTree/ExportPartitionUtils.cpp +++ b/src/Storages/MergeTree/ExportPartitionUtils.cpp @@ -154,10 +154,17 @@ namespace ExportPartitionUtils context_copy->setCurrentQueryId(manifest.query_id); context_copy->setSetting("output_format_parallel_formatting", manifest.parallel_formatting); context_copy->setSetting("output_format_parquet_parallel_encoding", manifest.parquet_parallel_encoding); - context_copy->setSetting("output_format_parquet_compression_method", manifest.parquet_compression_method); - context_copy->setSetting("output_format_compression_level", manifest.output_format_compression_level); - context_copy->setSetting("output_format_parquet_row_group_size", manifest.parquet_row_group_size); - context_copy->setSetting("output_format_parquet_row_group_size_bytes", manifest.parquet_row_group_size_bytes); + + /// Backwards compatibility + if (manifest.parquet_compression_method) + context_copy->setSetting("output_format_parquet_compression_method", *manifest.parquet_compression_method); + if (manifest.output_format_compression_level) + context_copy->setSetting("output_format_compression_level", *manifest.output_format_compression_level); + if (manifest.parquet_row_group_size) + 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); + 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))); context_copy->setSetting("export_merge_tree_part_max_bytes_per_file", manifest.max_bytes_per_file); @@ -313,7 +320,8 @@ namespace ExportPartitionUtils getPartitionSourceBlockForIcebergCommit(source_storage, manifest.partition_id); } - destination_storage->commitExportPartitionTransaction(manifest.transaction_id, manifest.partition_id, exported_paths, iceberg_args, context); + const auto destination_commit_info = destination_storage->commitExportPartitionTransaction( + manifest.transaction_id, manifest.partition_id, exported_paths, iceberg_args, context); /// Failpoint to simulate a crash after the Iceberg commit succeeds but before /// ZooKeeper is updated to COMPLETED. Used by idempotency integration tests. @@ -326,16 +334,41 @@ namespace ExportPartitionUtils }); LOG_INFO(log, "ExportPartition: Committed export, mark as completed"); + + const std::string status_path = fs::path(entry_path) / "status"; + const std::string completed_name = String(magic_enum::enum_name(ExportReplicatedMergeTreePartitionTaskEntry::Status::COMPLETED)).data(); + + Coordination::Requests ops; + ops.emplace_back(zkutil::makeSetRequest(status_path, completed_name, -1)); + + ExportReplicatedMergeTreePartitionCommitInfoEntry commit_info_entry { + destination_commit_info.iceberg_metadata_file, + destination_commit_info.iceberg_manifest_list, + destination_commit_info.iceberg_manifest_file, + destination_commit_info.commit_marker_file}; + + const std::string commit_info_path = fs::path(entry_path) / "commit_info"; + ops.emplace_back(zkutil::makeCreateRequest(commit_info_path, commit_info_entry.toJsonString(), zkutil::CreateMode::Persistent)); + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperRequests); - ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperSet); - if (Coordination::Error::ZOK == zk->trySet(fs::path(entry_path) / "status", String(magic_enum::enum_name(ExportReplicatedMergeTreePartitionTaskEntry::Status::COMPLETED)).data(), -1)) + ProfileEvents::increment(ProfileEvents::ExportPartitionZooKeeperMulti); + + Coordination::Responses responses; + const auto rc = zk->tryMulti(ops, responses); + + if (rc == Coordination::Error::ZOK) { - LOG_INFO(log, "ExportPartition: Marked export as completed"); + LOG_INFO(log, "ExportPartition: Marked export as completed and persisted commit_info"); + return; } - else + + if (rc == Coordination::Error::ZNODEEXISTS) { - throw Exception(ErrorCodes::NETWORK_ERROR, "ExportPartition: Failed to mark export as completed, will not try to fix it"); + LOG_INFO(log, "ExportPartition: commit_info already present (peer wrote it first); task already COMPLETED"); + return; } + + throw Exception(ErrorCodes::NETWORK_ERROR, "ExportPartition: Failed to mark export as completed (rc={}), will not try to fix it", rc); } bool handleCommitFailure( diff --git a/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp b/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp index c4669a9bf55c..b3c750129988 100644 --- a/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp +++ b/src/Storages/MergeTree/tests/gtest_export_partition_ordering.cpp @@ -43,9 +43,9 @@ TEST_F(ExportPartitionOrderingTest, IterationOrderMatchesCreateTime) manifest3.transaction_id = "tx3"; manifest3.create_time = base_time; // Oldest - ExportReplicatedMergeTreePartitionTaskEntry entry1{manifest1, ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING, {}, {}}; - ExportReplicatedMergeTreePartitionTaskEntry entry2{manifest2, ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING, {}, {}}; - ExportReplicatedMergeTreePartitionTaskEntry entry3{manifest3, ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING, {}, {}}; + ExportReplicatedMergeTreePartitionTaskEntry entry1{manifest1, ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING, {}, {}, {}, {}}; + ExportReplicatedMergeTreePartitionTaskEntry entry2{manifest2, ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING, {}, {}, {}, {}}; + ExportReplicatedMergeTreePartitionTaskEntry entry3{manifest3, ExportReplicatedMergeTreePartitionTaskEntry::Status::PENDING, {}, {}, {}, {}}; // Insert in reverse order by_key.insert(entry1); diff --git a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h index a8a94f530bac..f116afc73e4e 100644 --- a/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -221,7 +222,7 @@ class IDataLakeMetadata : boost::noncopyable throwNotImplemented("import"); } - virtual void commitExportPartitionTransaction( + virtual IStorage::ExportPartitionCommitInfo commitExportPartitionTransaction( std::shared_ptr /* catalog */, const StorageID & /* table_id */, const String & /* transaction_id */, diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index bd5686db3436..e5f1d20898d5 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -1670,7 +1670,7 @@ std::vector recomputeExportPartitionValues( } -bool IcebergMetadata::commitImportPartitionTransactionImpl( +std::optional IcebergMetadata::commitImportPartitionTransactionImpl( FileNamesGenerator & filename_generator, Poco::JSON::Object::Ptr & metadata, Poco::JSON::Object::Ptr & partition_spec, @@ -1698,7 +1698,13 @@ bool IcebergMetadata::commitImportPartitionTransactionImpl( LOG_INFO(log, "Export transaction {} already committed, skipping re-commit", transaction_id); - return true; + /// Surface a sentinel so the caller treats this as a successful attempt (non-empty + /// commit info), persists a commit_info znode, and makes the situation visible in + /// system.replicated_partition_exports.committed_metadata_file. We do not know the + /// original committer's paths from here. + IStorage::ExportPartitionCommitInfo already_committed_info; + already_committed_info.iceberg_metadata_file = ""; + return already_committed_info; } const auto & resolver = persistent_components.path_resolver; @@ -1893,7 +1899,7 @@ bool IcebergMetadata::commitImportPartitionTransactionImpl( { LOG_DEBUG(log, "Failed to write metadata {}, retrying", storage_metadata_name); cleanup(true); - return false; + return {}; } LOG_DEBUG(log, "Metadata file {} written", storage_metadata_name); @@ -1908,7 +1914,7 @@ bool IcebergMetadata::commitImportPartitionTransactionImpl( if (!catalog->updateMetadata(namespace_name, table_name, catalog_filename, new_snapshot)) { cleanup(true); - return false; + return {}; } /// Catalog has accepted the commit - the new snapshot is now live and references @@ -1951,11 +1957,16 @@ bool IcebergMetadata::commitImportPartitionTransactionImpl( /// post-publish work (e.g. metadata-cache invalidation). Running cleanup() /// here would delete manifest files referenced by the published snapshot /// and corrupt it. Log and swallow - any transient state (stale cache) - /// is self-healing on subsequent reads. + /// is self-healing on subsequent reads. Surface the published paths anyway + /// so the partition export task can persist them in ZooKeeper. tryLogCurrentException(log, "Post-publish work failed after Iceberg snapshot was committed; " "skipping manifest cleanup to preserve published snapshot"); - return true; + IStorage::ExportPartitionCommitInfo published_info; + published_info.iceberg_metadata_file = resolver.resolve(metadata_info.path); + published_info.iceberg_manifest_list = storage_manifest_list_name; + published_info.iceberg_manifest_file = storage_manifest_entry_name; + return published_info; } LOG_ERROR(log, "Failed to commit import partition transaction: {}", getCurrentExceptionMessage(false)); @@ -1963,10 +1974,18 @@ bool IcebergMetadata::commitImportPartitionTransactionImpl( throw; } - return true; + /// Record the storage paths of the files we just published so the partition + /// export task can persist them in ZooKeeper for observability. Only set here + /// (not on the retry / "already committed" paths) so the struct reflects + /// exactly what this attempt produced. + IStorage::ExportPartitionCommitInfo published_info; + published_info.iceberg_metadata_file = resolver.resolve(metadata_info.path); + published_info.iceberg_manifest_list = storage_manifest_list_name; + published_info.iceberg_manifest_file = storage_manifest_entry_name; + return published_info; } -void IcebergMetadata::commitExportPartitionTransaction( +IStorage::ExportPartitionCommitInfo IcebergMetadata::commitExportPartitionTransaction( std::shared_ptr catalog, const StorageID & table_id, const String & transaction_id, @@ -2005,7 +2024,9 @@ void IcebergMetadata::commitExportPartitionTransaction( LOG_INFO(log, "Export transaction {} already committed, skipping re-commit", transaction_id); - return; + IStorage::ExportPartitionCommitInfo already_committed_info; + already_committed_info.iceberg_metadata_file = ""; + return already_committed_info; } /// Fail fast if the table schema or partition spec changed between export-start and commit. @@ -2072,7 +2093,7 @@ void IcebergMetadata::commitExportPartitionTransaction( size_t attempt = 0; while (attempt < MAX_TRANSACTION_RETRIES) { - if (commitImportPartitionTransactionImpl( + auto commit_info = commitImportPartitionTransactionImpl( filename_generator, metadata, partition_spec, @@ -2092,10 +2113,10 @@ void IcebergMetadata::commitExportPartitionTransaction( table_id, configuration->getTypeName(), configuration->getNamespace(), - context)) - { - return; - } + context); + + if (commit_info) + return *commit_info; ++attempt; } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h index 87e8d5588172..d02bf91da236 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h @@ -163,7 +163,7 @@ class IcebergMetadata : public IDataLakeMetadata /// data_file_paths contains the metadata-path for each exported data file (as recorded in /// ZooKeeper). For every path a co-located sidecar Avro file (same path, ".avro" extension) /// must exist in the object storage; it supplies record_count and file_size_in_bytes. - void commitExportPartitionTransaction( + IStorage::ExportPartitionCommitInfo commitExportPartitionTransaction( std::shared_ptr catalog, const StorageID & table_id, const String & transaction_id, @@ -241,7 +241,12 @@ class IcebergMetadata : public IDataLakeMetadata Iceberg::IcebergDataSnapshotPtr getRelevantDataSnapshotFromTableStateSnapshot(Iceberg::TableStateSnapshot table_state_snapshot, ContextPtr local_context) const; - bool commitImportPartitionTransactionImpl( + /// Non-empty return value means the attempt succeeded (covers both the normal + /// publish path and the `isExportPartitionTransactionAlreadyCommitted` short-circuit). + /// An empty `ExportPartitionCommitInfo` means the caller must retry. The + /// short-circuit branch fills `iceberg_metadata_file` with a sentinel note since + /// the original committer's paths are not trivially recoverable from inside this call. + std::optional commitImportPartitionTransactionImpl( FileNamesGenerator & filename_generator, Poco::JSON::Object::Ptr & metadata, Poco::JSON::Object::Ptr & partition_spec, diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index 81140d94f207..01e4a1d4ff6e 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -722,7 +722,7 @@ SinkToStoragePtr StorageObjectStorage::import( local_context); } -void StorageObjectStorage::commitExportPartitionTransaction( +IStorage::ExportPartitionCommitInfo StorageObjectStorage::commitExportPartitionTransaction( const String & transaction_id, const String & partition_id, const Strings & exported_paths, @@ -745,7 +745,7 @@ void StorageObjectStorage::commitExportPartitionTransaction( configuration->lazyInitializeIfNeeded(object_storage, local_context); auto metadata_snapshot = getInMemoryMetadataPtr(local_context, false); - configuration->getExternalMetadata()->commitExportPartitionTransaction( + return configuration->getExternalMetadata()->commitExportPartitionTransaction( catalog, storage_id, transaction_id, @@ -756,16 +756,20 @@ void StorageObjectStorage::commitExportPartitionTransaction( exported_paths, configuration, local_context); - return; } const String commit_object = configuration->getRawPath().path + "/commit_" + partition_id + "_" + transaction_id; + ExportPartitionCommitInfo result; + result.commit_marker_file = commit_object; + /// if file already exists, nothing to be done if (object_storage->exists(StoredObject(commit_object))) { LOG_DEBUG(getLogger("StorageObjectStorage"), "Commit file already exists, nothing to be done: {}", commit_object); - return; + /// Still surface the path: observability does not require we wrote it, + /// only that it is the committed marker for this transaction. + return result; } auto out = object_storage->writeObject(StoredObject(commit_object), WriteMode::Rewrite, /* attributes= */ {}, DBMS_DEFAULT_BUFFER_SIZE, local_context->getWriteSettings()); @@ -775,6 +779,7 @@ void StorageObjectStorage::commitExportPartitionTransaction( out->write("\n", 1); } out->finalize(); + return result; } void StorageObjectStorage::truncate( diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.h b/src/Storages/ObjectStorage/StorageObjectStorage.h index 45807c7c89c4..5f1db8f2a527 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.h +++ b/src/Storages/ObjectStorage/StorageObjectStorage.h @@ -95,7 +95,7 @@ class StorageObjectStorage : public IStorage, public IBackgroundOperation const std::optional & /* format_settings_ */, ContextPtr /* context */) override; - void commitExportPartitionTransaction( + ExportPartitionCommitInfo commitExportPartitionTransaction( const String & transaction_id, const String & partition_id, const Strings & exported_paths, diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 63fcec56de3f..9eefd709aba1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -1144,7 +1144,7 @@ SinkToStoragePtr StorageObjectStorageCluster::import( context); } -void StorageObjectStorageCluster::commitExportPartitionTransaction( +IStorage::ExportPartitionCommitInfo StorageObjectStorageCluster::commitExportPartitionTransaction( const String & transaction_id, const String & partition_id, const Strings & exported_paths, @@ -1153,16 +1153,15 @@ void StorageObjectStorageCluster::commitExportPartitionTransaction( { if (pure_storage) { - pure_storage->commitExportPartitionTransaction( + return pure_storage->commitExportPartitionTransaction( transaction_id, partition_id, exported_paths, iceberg_commit_export_partition_arguments, local_context ); - return; } - IStorageCluster::commitExportPartitionTransaction( + return IStorageCluster::commitExportPartitionTransaction( transaction_id, partition_id, exported_paths, diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 1b73b95b20a4..6894bb76d2e1 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -44,7 +44,7 @@ class StorageObjectStorageCluster : public IStorageCluster const std::optional & format_settings_, ContextPtr context) override; - void commitExportPartitionTransaction( + ExportPartitionCommitInfo commitExportPartitionTransaction( const String & transaction_id, const String & partition_id, const Strings & exported_paths, diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index 9dd83a7ba872..b27167bf5d4f 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -134,7 +134,6 @@ #include #include -#include "Functions/generateSnowflakeID.h" #include "Interpreters/StorageID.h" #include "QueryPipeline/QueryPlanResourceHolder.h" #include "Storages/ExportReplicatedMergeTreePartitionManifest.h" @@ -8737,7 +8736,7 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & ExportReplicatedMergeTreePartitionManifest manifest; - manifest.transaction_id = generateSnowflakeIDString(); + manifest.transaction_id = toString(UUIDHelpers::generateV4()); manifest.query_id = query_context->getCurrentQueryId(); manifest.partition_id = partition_id; manifest.destination_database = dest_database; diff --git a/src/Storages/System/StorageSystemReplicatedPartitionExports.cpp b/src/Storages/System/StorageSystemReplicatedPartitionExports.cpp index 9aa9d1846273..6370e0f99433 100644 --- a/src/Storages/System/StorageSystemReplicatedPartitionExports.cpp +++ b/src/Storages/System/StorageSystemReplicatedPartitionExports.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -55,6 +56,16 @@ ColumnsDescription StorageSystemReplicatedPartitionExports::getColumnsDescriptio "Per-replica last exception entries. Each tuple records the most recent exception observed by that replica plus a best-effort within-replica count. Empty array if no replica has reported an exception for this task."}, {"exception_count", std::make_shared(), "Sum of per-replica exception counts. Each replica owns its own count, so the sum is exact w.r.t. the in-memory snapshot; within-replica updates remain best-effort and may under-count by one under concurrent failures."}, + {"destination_file_paths", std::make_shared(std::make_shared(), std::make_shared(std::make_shared())), + "Per-part destination file paths written to the destination object storage. Keyed by part name; values are the file paths produced by exporting that part. Mirrored from ZooKeeper on every poll while PENDING; partial during in-flight tasks. When the in-memory mirror could not fully refresh from Keeper (or a processed leaf is unreadable), the map may contain the key and/or path value '' for the affected part (or for the whole field if listing processed leaves failed); replaced on the next successful poll."}, + {"committed_metadata_file", std::make_shared(), + "For Iceberg destinations: path of the new metadata JSON file written at commit time. Empty for non-Iceberg destinations and for tasks that have not committed yet. May also be empty if the committing replica crashed between writing the object-storage files and persisting commit_info. If the export was already committed by a previous run (detected via the transaction id stored in the snapshot summary), this column holds a human-readable note instead of a path since the original committer's paths are not trivially recoverable."}, + {"committed_manifest_list", std::make_shared(), + "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", std::make_shared(), + "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", std::make_shared(), + "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."}, {"local_backoff_per_part", std::make_shared(backoff_tuple), "Per-part retry back-off local to this replica: parts currently waiting before their next attempt, with attempt count and the next eligible time. Not shared across replicas; empty if no part is backing off."}, }; @@ -163,6 +174,25 @@ void StorageSystemReplicatedPartitionExports::fillData(MutableColumns & res_colu res_columns[i++]->insert(per_replica); res_columns[i++]->insert(info.exception_count); + Map destination_paths_map; + destination_paths_map.reserve(info.destination_file_paths_per_part.size()); + for (const auto & [part_name, paths] : info.destination_file_paths_per_part) + { + Array paths_array; + paths_array.reserve(paths.size()); + for (const auto & path : paths) + paths_array.push_back(path); + destination_paths_map.emplace_back(Tuple{part_name, std::move(paths_array)}); + } + res_columns[i++]->insert(std::move(destination_paths_map)); + + res_columns[i++]->insert(info.committed_metadata_file); + res_columns[i++]->insert(info.committed_manifest_list); + + res_columns[i++]->insert(info.committed_manifest_file); + + res_columns[i++]->insert(info.committed_marker_file); + Array backoff_array; backoff_array.reserve(info.backoff_per_part.size()); for (const auto & b : info.backoff_per_part) diff --git a/src/Storages/System/StorageSystemReplicatedPartitionExports.h b/src/Storages/System/StorageSystemReplicatedPartitionExports.h index 09d7d3eaf9a2..768516c0a2ea 100644 --- a/src/Storages/System/StorageSystemReplicatedPartitionExports.h +++ b/src/Storages/System/StorageSystemReplicatedPartitionExports.h @@ -30,12 +30,31 @@ struct ReplicatedPartitionExportInfo /// count by one), matching the documented column semantics. size_t exception_count = 0; + /// Per-part destination file paths, keyed by part name. Mirrors the + /// /processed//paths_in_destination data from ZooKeeper. + /// Empty until parts complete; partial during PENDING. May contain + /// "" when a Keeper refresh was incomplete or a + /// processed leaf could not be parsed. + std::map> destination_file_paths_per_part; + + /// Iceberg commit-time paths surfaced from /commit_info. + /// All empty for non-Iceberg destinations or before commit lands. + String committed_metadata_file; + String committed_manifest_list; + String committed_manifest_file; + + /// Plain object storage commit marker file surfaced from + /// /commit_info. Empty for Iceberg destinations or before + /// commit lands. + String committed_marker_file; + struct PartBackoffEntry { String part; size_t attempts = 0; time_t next_retry_time = 0; }; + /// Parts of this task currently backing off (local to this replica). Empty if none. std::vector backoff_per_part; }; diff --git a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py index 383739b6b0c7..ad2deba8de19 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_iceberg/test.py @@ -167,6 +167,42 @@ def test_export_partition_to_iceberg(cluster): ) +def _destination_paths_has_sync_failed_marker(node, source_table, dest_table, partition_id): + """True when destination_file_paths contains the Keeper sync-failed marker value.""" + result = node.query( + f"SELECT has(arrayFlatten(mapValues(destination_file_paths)), '')" + f" FROM system.replicated_partition_exports" + f" WHERE source_table = '{source_table}'" + f" AND destination_table = '{dest_table}'" + f" AND partition_id = '{partition_id}'" + ).strip() + return result == "1" + + +def wait_for_destination_paths_sync_failed_marker( + node, source_table, dest_table, partition_id, expect_marker, timeout=90, poll_interval=0.5 +): + """Wait until destination_file_paths does/does not contain the sync-failed marker. + + The in-memory mirror refreshes on the manifest-updater poll (~30s), so the + default timeout allows at least one full cycle plus headroom. + """ + start_time = time.time() + last = None + while time.time() - start_time < timeout: + last = _destination_paths_has_sync_failed_marker( + node, source_table, dest_table, partition_id + ) + if last == expect_marker: + return + time.sleep(poll_interval) + + raise TimeoutError( + f"destination_file_paths sync-failed marker did not become {expect_marker}" + f" within {timeout}s (last={last})" + ) + + def test_export_two_partitions_to_iceberg(cluster): """ Export two partitions in a single ALTER TABLE statement and verify that both @@ -989,8 +1025,9 @@ def test_post_publish_exception_preserves_snapshot(cluster): post-publish region (after both the metadata file is written and `published = true` is set). With the fix in place: - the commit stays durable (snapshot is readable, manifests are intact); - - the export is marked COMPLETED because the idempotency check on retry - detects that the transaction is already committed and returns success; + - the export is marked COMPLETED because the outer `catch (...)` sees + `published == true` and returns the populated commit info with the real + paths produced by this attempt (no retry needed); - all exported rows are visible through the Iceberg table. """ node = cluster.instances["replica1"] @@ -1021,6 +1058,28 @@ def test_post_publish_exception_preserves_snapshot(cluster): f"Unexpected data after post-publish exception recovery:\n{result}" ) + # After a post-publish exception the catch handler with published==true returns + # the populated commit info (real metadata / manifest list / manifest file paths). + # ExportPartitionUtils::commit persists it to the commit_info znode, so the system + # table should show a real metadata path here, not the already-committed sentinel. + committed_metadata_file = node.query( + f""" + SELECT committed_metadata_file FROM system.replicated_partition_exports + WHERE source_table = '{mt_table}' + AND destination_table = '{iceberg_table}' + AND partition_id = '2020' + """ + ).strip() + assert committed_metadata_file, ( + "committed_metadata_file should be populated after a successful post-publish-catch return" + ) + assert not committed_metadata_file.startswith("<"), ( + f"committed_metadata_file should be a real metadata path, got the already-committed sentinel: {committed_metadata_file!r}" + ) + assert committed_metadata_file.endswith(".metadata.json"), ( + f"Expected a *.metadata.json path in committed_metadata_file, got: {committed_metadata_file!r}" + ) + def test_export_task_timeout_kills_stuck_pending_task(cluster): """ diff --git a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py index 44f35925f1e7..8d4589292e3c 100644 --- a/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py +++ b/tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py @@ -740,6 +740,29 @@ def test_export_partition_file_already_exists_policy(cluster): # wait for the exports to finish wait_for_export_status(node, mt_table, s3_table, "2020", "COMPLETED") + # plain object storage destinations surface the commit marker file path via + # system.replicated_partition_exports.committed_marker_file + committed_marker_file = node.query( + f""" + SELECT committed_marker_file FROM system.replicated_partition_exports + WHERE source_table = '{mt_table}' + AND destination_table = '{s3_table}' + AND partition_id = '2020' + """ + ).strip() + # `committed_marker_file` is the absolute key in the bucket (same convention as + # `destination_file_paths`); it may carry the s3_conn URL's in-bucket prefix on + # top of the table's `filename` argument, so use a "contains" check that does + # not depend on knowing that prefix. + assert f"{s3_table}/commit_2020_" in committed_marker_file, \ + f"Expected committed_marker_file under {s3_table}/, got: {committed_marker_file!r}" + # Path relative to the `s3_conn` URL, derived from the absolute key without + # assuming a particular URL prefix. + marker_relative_path = committed_marker_file[committed_marker_file.index(f"{s3_table}/"):] + assert node.query( + f"SELECT count() FROM s3(s3_conn, filename='{marker_relative_path}', format=LineAsString)" + ) == '1\n', f"Commit marker file does not exist at {committed_marker_file!r}" + # try to export the partition node.query( f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {s3_table} SETTINGS export_merge_tree_partition_force_export=1" diff --git a/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py b/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py index 94d4d6c1a017..431df7efbeb7 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg.py @@ -712,6 +712,19 @@ def test_idempotency_after_commit_crash(export_cluster): count = int(node.query(f"SELECT count() FROM {iceberg}").strip()) assert count == 3, f"Expected 3 rows (no duplicates), got {count}" + # The already-committed early-exit in commitExportPartitionTransaction surfaces + # a sentinel note in committed_metadata_file (the original committer's paths + # are not recoverable from inside the call). The sentinel makes the situation + # visible in system.replicated_partition_exports rather than leaving the + # commit_info columns empty. + committed_metadata_file = node.query( + f"SELECT committed_metadata_file FROM system.replicated_partition_exports " + f"WHERE source_table = '{source}' AND partition_id = '{pid}'" + ).strip() + assert committed_metadata_file == "", ( + f"Expected already-committed sentinel after idempotent retry, got: {committed_metadata_file!r}" + ) + # --------------------------------------------------------------------------- # Replicated tests — IcebergS3, no catalog diff --git a/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg_catalog.py b/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg_catalog.py index b32582829197..e05e762a5f84 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg_catalog.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_export_partition_iceberg_catalog.py @@ -373,6 +373,14 @@ def test_catalog_idempotent_retry(catalog_export_cluster): f"got {len(history)}" ) + committed_metadata_file = node.query( + f"SELECT committed_metadata_file FROM system.replicated_partition_exports " + f"WHERE source_table = '{source}' AND partition_id = '{pid}'" + ).strip() + assert committed_metadata_file == "", ( + f"Expected already-committed sentinel after idempotent retry, got: {committed_metadata_file!r}" + ) + # --------------------------------------------------------------------------- # Replicated catalog tests