diff --git a/src/Storages/IStorage.h b/src/Storages/IStorage.h index a77f425b9f75..fb73f574c018 100644 --- a/src/Storages/IStorage.h +++ b/src/Storages/IStorage.h @@ -62,6 +62,9 @@ struct StreamLocalLimits; class EnabledQuota; struct SelectQueryInfo; +/// Declared opaquely (definition in Core/SettingsEnums.h) to keep this widely included header light. +enum class MergeTreePartExportFileAlreadyExistsPolicy : uint8_t; + using NameDependencies = std::unordered_map>; using DatabaseAndTableName = std::pair; @@ -459,7 +462,7 @@ It is currently only implemented in StorageObjectStorage. const std::string & /* file_name */, Block & /* block_with_partition_values */, const std::function & /* new_file_path_callback */, - bool /* overwrite_if_exists */, + MergeTreePartExportFileAlreadyExistsPolicy /* file_already_exists_policy */, std::size_t /* max_bytes_per_file */, std::size_t /* max_rows_per_file */, const std::optional & /* iceberg_metadata_json_string */, diff --git a/src/Storages/MergeTree/ExportPartTask.cpp b/src/Storages/MergeTree/ExportPartTask.cpp index f86eb02b176a..aa1f49a97a9f 100644 --- a/src/Storages/MergeTree/ExportPartTask.cpp +++ b/src/Storages/MergeTree/ExportPartTask.cpp @@ -316,7 +316,7 @@ bool ExportPartTask::executeStep() filename, block_with_partition_values, new_file_path_callback, - manifest.file_already_exists_policy == MergeTreePartExportManifest::FileAlreadyExistsPolicy::overwrite, + manifest.file_already_exists_policy, manifest.settings[Setting::export_merge_tree_part_max_bytes_per_file], manifest.settings[Setting::export_merge_tree_part_max_rows_per_file], manifest.iceberg_metadata_json, diff --git a/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.cpp b/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.cpp index c14223d94044..4c419e0acb90 100644 --- a/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.cpp +++ b/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.cpp @@ -1,6 +1,8 @@ #include #include #include +#include +#include #include namespace DB @@ -9,6 +11,16 @@ namespace DB namespace ErrorCodes { extern const int FILE_ALREADY_EXISTS; + extern const int CORRUPTED_DATA; +} + +namespace +{ + /// The commit file lives in the same directory as the data files. + std::string commitFilePath(const std::string & base_path, const String & transaction_id) + { + return (std::filesystem::path(base_path).parent_path() / ("commit_" + transaction_id)).string(); + } } MultiFileStorageObjectStorageSink::MultiFileStorageObjectStorageSink( @@ -18,7 +30,7 @@ MultiFileStorageObjectStorageSink::MultiFileStorageObjectStorageSink( StorageObjectStorageConfigurationPtr configuration_, std::size_t max_bytes_per_file_, std::size_t max_rows_per_file_, - bool overwrite_if_exists_, + FileAlreadyExistsPolicy file_already_exists_policy_, const std::function & new_file_path_callback_, const std::optional & format_settings_, SharedHeader sample_block_, @@ -26,16 +38,41 @@ MultiFileStorageObjectStorageSink::MultiFileStorageObjectStorageSink( : SinkToStorage(sample_block_), base_path(base_path_), transaction_id(transaction_id_), + commit_file_path(commitFilePath(base_path_, transaction_id_)), object_storage(object_storage_), configuration(configuration_), max_bytes_per_file(max_bytes_per_file_), max_rows_per_file(max_rows_per_file_), - overwrite_if_exists(overwrite_if_exists_), + file_already_exists_policy(file_already_exists_policy_), new_file_path_callback(new_file_path_callback_), format_settings(format_settings_), sample_block(sample_block_), context(context_) { + if (file_already_exists_policy != FileAlreadyExistsPolicy::overwrite) + { + if (auto committed_paths = tryReadCommittedPaths()) + { + if (committed_paths->empty()) + throw Exception(ErrorCodes::CORRUPTED_DATA, + "Commit file {} lists no data files", commit_file_path); + + /// Report the whole committed set before throwing: a caller applying `skip` takes these + /// paths as the part's export result, so it needs every file and not just the first. + for (const auto & committed_path : *committed_paths) + new_file_path_callback(committed_path); + + throw Exception(ErrorCodes::FILE_ALREADY_EXISTS, + "Part was already exported as {} file(s), see commit file {}", + committed_paths->size(), commit_file_path); + } + } + + /// No commit file: either a fresh export, or an attempt that died before finalizing every + /// file. `error` still reports the leftovers as a conflict, but `skip` has to rewrite them -- + /// the files that attempt never reached carry rows no later attempt produces. + overwrite_data_files = file_already_exists_policy != FileAlreadyExistsPolicy::error; + current_sink = createNewSink(); } @@ -72,15 +109,13 @@ std::shared_ptr MultiFileStorageObjectStorageSink::cre { auto new_path = generateNewFilePath(); - /// todo - /// sounds like bad design, but callers might decide to ignore the exception, and if we throw it before the callback - /// they will not be able to grab the file path. - /// maybe I should consider moving the file already exists policy in here? + /// The callback runs before the conflict check on purpose: under `error` the caller discards + /// the reported path along with the failure, and under the other policies this check is off. new_file_path_callback(new_path); file_paths.emplace_back(std::move(new_path)); - if (!overwrite_if_exists && object_storage->exists(StoredObject(file_paths.back()))) + if (!overwrite_data_files && object_storage->exists(StoredObject(file_paths.back()))) { throw Exception(ErrorCodes::FILE_ALREADY_EXISTS, "File {} already exists", file_paths.back()); } @@ -125,12 +160,33 @@ void MultiFileStorageObjectStorageSink::onFinish() commit(); } -void MultiFileStorageObjectStorageSink::commit() +std::optional> MultiFileStorageObjectStorageSink::tryReadCommittedPaths() const { - /// the commit file path should be in the same directory as the data files - const auto commit_file_path = fs::path(base_path).parent_path() / ("commit_" + transaction_id); + if (!object_storage->exists(StoredObject(commit_file_path))) + return {}; + + auto in = object_storage->readObject(StoredObject(commit_file_path), context->getReadSettings()); + + std::vector committed_paths; + while (!in->eof()) + { + String committed_path; + readStringUntilNewlineInto(committed_path, *in); + if (!in->eof()) + in->ignore(1); + if (!committed_path.empty()) + committed_paths.emplace_back(std::move(committed_path)); + } - if (!overwrite_if_exists && object_storage->exists(StoredObject(commit_file_path))) + return committed_paths; +} + +void MultiFileStorageObjectStorageSink::commit() +{ + /// The constructor already ruled out a pre-existing commit file for every policy but + /// `overwrite`, so seeing one here means another exporter committed this part while we wrote. + if (file_already_exists_policy != FileAlreadyExistsPolicy::overwrite + && object_storage->exists(StoredObject(commit_file_path))) { throw Exception(ErrorCodes::FILE_ALREADY_EXISTS, "Commit file {} already exists, aborting {} export", commit_file_path, transaction_id); } diff --git a/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.h b/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.h index 51f6b8094232..5febf0e65da3 100644 --- a/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.h +++ b/src/Storages/ObjectStorage/MultiFileStorageObjectStorageSink.h @@ -1,5 +1,6 @@ #pragma once +#include #include namespace DB @@ -11,6 +12,8 @@ namespace DB class MultiFileStorageObjectStorageSink : public SinkToStorage { public: + using FileAlreadyExistsPolicy = MergeTreePartExportFileAlreadyExistsPolicy; + MultiFileStorageObjectStorageSink( const std::string & base_path_, const String & transaction_id_, @@ -18,7 +21,7 @@ class MultiFileStorageObjectStorageSink : public SinkToStorage StorageObjectStorageConfigurationPtr configuration_, std::size_t max_bytes_per_file_, std::size_t max_rows_per_file_, - bool overwrite_if_exists_, + FileAlreadyExistsPolicy file_already_exists_policy_, const std::function & new_file_path_callback_, const std::optional & format_settings_, SharedHeader sample_block_, @@ -35,11 +38,17 @@ class MultiFileStorageObjectStorageSink : public SinkToStorage private: const std::string base_path; const String transaction_id; + /// Written by `commit` only after every data file has been finalized, so its presence -- + /// unlike that of any individual data file -- proves a previous export of this part + /// produced the whole set. + const std::string commit_file_path; ObjectStoragePtr object_storage; StorageObjectStorageConfigurationPtr configuration; std::size_t max_bytes_per_file; std::size_t max_rows_per_file; - bool overwrite_if_exists; + FileAlreadyExistsPolicy file_already_exists_policy; + /// Data files left behind by an attempt that never reached `commit` have to be rewritten. + bool overwrite_data_files = false; std::function new_file_path_callback; const std::optional format_settings; SharedHeader sample_block; @@ -51,6 +60,8 @@ class MultiFileStorageObjectStorageSink : public SinkToStorage std::string generateNewFilePath(); std::shared_ptr createNewSink(); + /// The data files a previous export of this part committed, or nothing when it never committed. + std::optional> tryReadCommittedPaths() const; void commit(); }; diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.cpp b/src/Storages/ObjectStorage/StorageObjectStorage.cpp index abd0cb896222..269027ccecc9 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorage.cpp @@ -673,7 +673,7 @@ SinkToStoragePtr StorageObjectStorage::import( const std::string & file_name, Block & block_with_partition_values, const std::function & new_file_path_callback, - bool overwrite_if_exists, + MergeTreePartExportFileAlreadyExistsPolicy file_already_exists_policy, std::size_t max_bytes_per_file, std::size_t max_rows_per_file, const std::optional & iceberg_metadata_json_string, @@ -733,7 +733,7 @@ SinkToStoragePtr StorageObjectStorage::import( configuration, max_bytes_per_file, max_rows_per_file, - overwrite_if_exists, + file_already_exists_policy, new_file_path_callback, format_settings_ ? format_settings_ : format_settings, std::make_shared(metadata_snapshot->getSampleBlock()), diff --git a/src/Storages/ObjectStorage/StorageObjectStorage.h b/src/Storages/ObjectStorage/StorageObjectStorage.h index 5f1db8f2a527..097497017380 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorage.h +++ b/src/Storages/ObjectStorage/StorageObjectStorage.h @@ -88,7 +88,7 @@ class StorageObjectStorage : public IStorage, public IBackgroundOperation const std::string & /* file_name */, Block & /* block_with_partition_values */, const std::function & new_file_path_callback, - bool /* overwrite_if_exists */, + MergeTreePartExportFileAlreadyExistsPolicy /* file_already_exists_policy */, std::size_t /* max_bytes_per_file */, std::size_t /* max_rows_per_file */, const std::optional & /* iceberg_metadata_json_string */, diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index 9eefd709aba1..53df3415f40b 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -1114,7 +1114,7 @@ SinkToStoragePtr StorageObjectStorageCluster::import( const std::string & file_name, Block & block_with_partition_values, const std::function & new_file_path_callback, - bool overwrite_if_exists, + MergeTreePartExportFileAlreadyExistsPolicy file_already_exists_policy, std::size_t max_bytes_per_file, std::size_t max_rows_per_file, const std::optional & iceberg_metadata_json_string, @@ -1126,7 +1126,7 @@ SinkToStoragePtr StorageObjectStorageCluster::import( file_name, block_with_partition_values, new_file_path_callback, - overwrite_if_exists, + file_already_exists_policy, max_bytes_per_file, max_rows_per_file, iceberg_metadata_json_string, @@ -1136,7 +1136,7 @@ SinkToStoragePtr StorageObjectStorageCluster::import( file_name, block_with_partition_values, new_file_path_callback, - overwrite_if_exists, + file_already_exists_policy, max_bytes_per_file, max_rows_per_file, iceberg_metadata_json_string, diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h index 6894bb76d2e1..1847f5340e77 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.h @@ -37,7 +37,7 @@ class StorageObjectStorageCluster : public IStorageCluster const std::string & file_name, Block & block_with_partition_values, const std::function & new_file_path_callback, - bool overwrite_if_exists, + MergeTreePartExportFileAlreadyExistsPolicy file_already_exists_policy, std::size_t max_bytes_per_file, std::size_t max_rows_per_file, const std::optional & iceberg_metadata_json_string, 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 b6c721c87dbf..cf425f76b7bd 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 @@ -831,6 +831,219 @@ def test_export_partition_file_already_exists_policy(cluster): ) == '1\n', "Expected the export to be marked as FAILED" +def export_transaction_id(node, mt_table, s3_table): + return node.query( + f""" + SELECT transaction_id FROM system.replicated_partition_exports + WHERE source_table = '{mt_table}' + AND destination_table = '{s3_table}' + AND partition_id = '2020' + """ + ).strip() + + +def wait_for_new_export_transaction(node, mt_table, s3_table, previous_transaction_id, timeout=60): + """Wait until the export entry carries a transaction id other than *previous_transaction_id*. + + A force re-export replaces the entry. Without this wait, the COMPLETED status of the export + being replaced can still be visible in the in-memory mirror and satisfy a status wait + immediately, before the new export has even started. + """ + start_time = time.time() + last_transaction_id = None + while time.time() - start_time < timeout: + last_transaction_id = export_transaction_id(node, mt_table, s3_table) + if last_transaction_id and last_transaction_id != previous_transaction_id: + return last_transaction_id + time.sleep(0.2) + + raise TimeoutError( + f"Export transaction id did not change from {previous_transaction_id!r} within {timeout}s. " + f"Last seen: {last_transaction_id!r}" + ) + + +def export_partition_split_into_files( + node, mt_table, s3_table, force=False, policy=None, previous_transaction_id=None +): + """Export partition 2020 with one row per destination file and wait for completion.""" + settings = ["export_merge_tree_part_max_rows_per_file = 1"] + if force: + settings.append("export_merge_tree_partition_force_export = 1") + if policy: + settings.append(f"export_merge_tree_part_file_already_exists_policy = '{policy}'") + + node.query( + f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {s3_table} " + f"SETTINGS {', '.join(settings)}" + ) + + if previous_transaction_id is not None: + wait_for_new_export_transaction(node, mt_table, s3_table, previous_transaction_id) + + wait_for_export_status(node, mt_table, s3_table, "2020", "COMPLETED") + + +def recorded_export_paths(node, mt_table, s3_table): + """Destination file paths recorded for the exported parts, in the order the sink wrote them. + + Mirrors the `/processed//paths_in_destination` data in ZooKeeper, which + is what the commit phase turns into the partition commit marker. + """ + paths = node.query( + f""" + SELECT arrayJoin(arrayFlatten(mapValues(destination_file_paths))) + FROM system.replicated_partition_exports + WHERE source_table = '{mt_table}' + AND destination_table = '{s3_table}' + AND partition_id = '2020' + """ + ) + return [path for path in paths.splitlines() if path] + + +def partition_commit_marker_lines(node, mt_table, s3_table): + """Data-file paths listed inside the partition-level commit marker.""" + 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() + 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 the + # URL's in-bucket prefix. + marker_relative_path = committed_marker_file[committed_marker_file.index(f"{s3_table}/"):] + lines = node.query( + f"SELECT * FROM s3(s3_conn, filename='{marker_relative_path}', format=LineAsString)" + ) + return [line for line in lines.splitlines() if line] + + +def list_partition_directory(cluster, data_path): + """Object keys sitting next to *data_path*, split into data files and commit markers. + + The per-part commit marker is written by `MultiFileStorageObjectStorageSink::commit` in the + same directory as the data files, named `commit_`. + """ + directory = data_path.rsplit("/", 1)[0] + "/" + object_names = sorted( + obj.object_name + for obj in cluster.minio_client.list_objects( + cluster.minio_bucket, prefix=directory, recursive=True + ) + ) + data_files = [n for n in object_names if not n.rsplit("/", 1)[-1].startswith("commit_")] + markers = [n for n in object_names if n.rsplit("/", 1)[-1].startswith("commit_")] + return data_files, markers + + +def test_export_partition_skip_policy_reports_every_split_file(cluster): + """A `skip` re-export of an already-exported multi-file part must record every destination + file, not just the first one. + + The recorded list is what the commit phase turns into the partition commit marker, so + dropping the later split files from it misrepresents the export even though the data is all + there. + """ + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"skip_reports_all_files_mt_table_{postfix}" + s3_table = f"skip_reports_all_files_s3_table_{postfix}" + + create_tables_and_insert_data(node, mt_table, s3_table, "replica1") + # The destination file name is derived from the part name, so part names have to stay stable + # across the two exports, otherwise the second one writes to fresh paths and skips nothing. + node.query(f"SYSTEM STOP MERGES {mt_table}") + + export_partition_split_into_files(node, mt_table, s3_table) + first_transaction_id = export_transaction_id(node, mt_table, s3_table) + + exported_paths = recorded_export_paths(node, mt_table, s3_table) + assert len(exported_paths) == 3, \ + f"Expected the 3-row partition to split into 3 files, got {exported_paths}" + assert len(partition_commit_marker_lines(node, mt_table, s3_table)) == 3 + + # Re-export. Every destination file is already there, so `skip` short-circuits the part -- + # but it must do so with the complete file list. + export_partition_split_into_files( + node, mt_table, s3_table, force=True, policy="skip", + previous_transaction_id=first_transaction_id, + ) + + skipped_paths = recorded_export_paths(node, mt_table, s3_table) + assert sorted(skipped_paths) == sorted(exported_paths), ( + f"Skipped re-export recorded {skipped_paths} instead of all 3 split files {exported_paths}" + ) + + committed = partition_commit_marker_lines(node, mt_table, s3_table) + assert len(committed) == 3, \ + f"Skipped re-export committed {len(committed)} path(s) instead of all 3 split files: {committed}" + + +def test_export_partition_skip_policy_reexports_incomplete_part(cluster): + """A part whose multi-file export was interrupted must be re-exported in full under `skip`. + + The first split file existing proves nothing on its own: only the per-part commit marker, + written after the last file is finalized, proves the part was fully exported. Removing the + trailing files together with the marker reproduces what an attempt that died mid-part leaves + behind, and the retry has to rewrite them -- the rows in those files are produced by no other + attempt. + """ + node = cluster.instances["replica1"] + + postfix = str(uuid.uuid4()).replace("-", "_") + mt_table = f"skip_reexports_partial_mt_table_{postfix}" + s3_table = f"skip_reexports_partial_s3_table_{postfix}" + + create_tables_and_insert_data(node, mt_table, s3_table, "replica1") + node.query(f"SYSTEM STOP MERGES {mt_table}") + + export_partition_split_into_files(node, mt_table, s3_table) + first_transaction_id = export_transaction_id(node, mt_table, s3_table) + + written_in_order = recorded_export_paths(node, mt_table, s3_table) + assert len(written_in_order) == 3, \ + f"Expected the 3-row partition to split into 3 files, got {written_in_order}" + + data_files, markers = list_partition_directory(cluster, written_in_order[0]) + assert data_files == sorted(written_in_order), \ + f"Objects in the partition directory {data_files} do not match the recorded paths {written_in_order}" + assert len(markers) == 1, f"Expected one per-part commit marker, got {markers}" + + # Roll the destination back to "first file finalized, nothing else": drop the trailing files + # and the marker that would otherwise prove the part complete. + for key in written_in_order[1:] + markers: + cluster.minio_client.remove_object(cluster.minio_bucket, key) + + surviving_data_files, surviving_markers = list_partition_directory(cluster, written_in_order[0]) + assert surviving_data_files == [written_in_order[0]], \ + f"Expected only the first split file to remain, got {surviving_data_files}" + assert surviving_markers == [], \ + f"Expected the per-part commit marker to be gone, got {surviving_markers}" + + export_partition_split_into_files( + node, mt_table, s3_table, force=True, policy="skip", + previous_transaction_id=first_transaction_id, + ) + + data_files_after, markers_after = list_partition_directory(cluster, written_in_order[0]) + assert len(data_files_after) == 3, ( + f"Retry left the part partially exported: {data_files_after} " + f"(the interrupted attempt's missing files were never rewritten)" + ) + assert len(markers_after) == 1, \ + f"Retry did not rewrite the per-part commit marker: {markers_after}" + assert node.query(f"SELECT count() FROM {s3_table} WHERE year = 2020") == "3\n", \ + "Rows from the split files the interrupted attempt never wrote are missing from the destination" + assert len(partition_commit_marker_lines(node, mt_table, s3_table)) == 3 + + def test_export_partition_feature_is_disabled(cluster): replica_with_export_disabled = cluster.instances["replica_with_export_disabled"]