possibly fix issue, vibe coded for now #2294
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,8 @@ | ||
| #include <Storages/ObjectStorage/MultiFileStorageObjectStorageSink.h> | ||
| #include <Interpreters/Context.h> | ||
| #include <Common/logger_useful.h> | ||
| #include <IO/ReadBufferFromFileBase.h> | ||
| #include <IO/ReadHelpers.h> | ||
| #include <filesystem> | ||
|
|
||
| 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,24 +30,49 @@ MultiFileStorageObjectStorageSink::MultiFileStorageObjectStorageSink( | |
| StorageObjectStorageConfigurationPtr configuration_, | ||
| std::size_t max_bytes_per_file_, | ||
| std::size_t max_rows_per_file_, | ||
| FileAlreadyExistsPolicy file_already_exists_policy_, | ||
| const std::function<void(const std::string &)> & new_file_path_callback_, | ||
| const std::optional<FormatSettings> & format_settings_, | ||
| SharedHeader sample_block_, | ||
| ContextPtr context_) | ||
| : 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_), | ||
| 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; | ||
|
There was a problem hiding this comment.
When no commit marker exists but a generated data path already exists—for example, an independently uploaded object, a removed marker, or an older incomplete export— AGENTS.md reference: AGENTS.md:L153-L153 Useful? React with 👍 / 👎. |
||
|
|
||
| current_sink = createNewSink(); | ||
| } | ||
|
|
||
|
|
@@ -72,15 +109,13 @@ std::shared_ptr<StorageObjectStorageSink> MultiFileStorageObjectStorageSink::cre | |
| { | ||
| auto new_path = generateNewFilePath(); | ||
|
|
||
| /// 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_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(); | ||
| } | ||
|
|
||
| std::optional<std::vector<std::string>> MultiFileStorageObjectStorageSink::tryReadCommittedPaths() const | ||
| { | ||
| if (!object_storage->exists(StoredObject(commit_file_path))) | ||
| return {}; | ||
|
|
||
| auto in = object_storage->readObject(StoredObject(commit_file_path), context->getReadSettings()); | ||
|
|
||
| std::vector<std::string> 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)); | ||
| } | ||
|
|
||
| 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); | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When a commit marker survives but one of its referenced data objects has been deleted or expired, this branch reports every recorded path and turns the resulting
FILE_ALREADY_EXISTSinto a successfulskipexport without checking that those objects still exist. For the common single-file export, the previous path-level existence check would instead rewrite a missing object; this change can now mark the exportCOMPLETEDand publish a path that returns no data. Validate every committed path before accepting the marker, and reconstruct or fail if any are missing.Useful? React with 👍 / 👎.