wangyong9999 commented on code in PR #227:
URL: https://github.com/apache/paimon-cpp/pull/227#discussion_r3859526697
##########
src/paimon/core/operation/abstract_file_store_write.cpp:
##########
@@ -188,6 +190,22 @@ Result<std::vector<std::shared_ptr<CommitMessage>>>
AbstractFileStoreWrite::Prep
}
std::vector<std::shared_ptr<CommitMessage>> result;
+ // A writer hands its files over the moment it drains an increment, so
once a message exists
+ // nothing the writer does removes its data files, sidecars or managed
blob packs. Failing
+ // out of this loop never hands `result` to the caller, so the messages
built so far would
+ // be stranded with no one left to clean them up.
+ ScopeGuard prepare_guard([this, &result]() {
+ if (result.empty()) {
+ return;
+ }
+ Status status = UncommittedFileCleaner::Delete(
Review Comment:
`result` may contain a metadata-only upgrade, where `CompactBefore` and
`CompactAfter` reuse the same `file_name`. If a later bucket fails during
prepare, this guard invokes the cleaner, which treats that file as an
uncommitted output and deletes it even though the previous snapshot still
references it, including its `.blobref`. Before deleting `CompactAfter`,
exclude files also present in `CompactBefore`, as `MergeTreeWriter::DoClose`
already does.
##########
src/paimon/core/operation/expire_snapshots.cpp:
##########
@@ -277,14 +280,31 @@ Status ExpireSnapshots::CleanUnusedDataFiles(const
std::string& manifest_list_na
}
}
+ // One data file path factory per (partition, bucket): creating one
re-derives the
+ // partition path and external path provider, too heavy to repeat for
every file.
+ std::unordered_map<BinaryRow, std::map<int32_t,
std::shared_ptr<DataFilePathFactory>>>
+ data_file_path_factories;
std::vector<std::future<void>> futures;
ScopeGuard guard([&futures]() { Wait(futures); });
for (const auto& [data_file_to_delete, entry] : data_files_to_delete) {
- auto delete_file_path = data_file_to_delete;
- futures.push_back(Via(executor_.get(), [this, delete_file_path]() {
- auto status = fs_->Delete(delete_file_path);
- // delete quietly will ignore any status error
- (void)status;
+ // An expired data file takes its companion files with it (e.g.
the managed blob
+ // reference sidecar). CollectFiles resolves their paths, honoring
external data
+ // paths.
+ std::shared_ptr<DataFilePathFactory>& data_file_path_factory =
+ data_file_path_factories[entry.Partition()][entry.Bucket()];
+ if (data_file_path_factory == nullptr) {
+ PAIMON_ASSIGN_OR_RAISE(
+ data_file_path_factory,
+
path_factory_->CreateDataFilePathFactory(entry.Partition(), entry.Bucket()));
+ }
+ std::vector<std::string> delete_file_paths =
Review Comment:
Snapshot expiration deletes only the data file and its `.blobref` here,
while the orphan cleaner always skips `.managed.blob` and does not support
primary-key tables. After an update, first-row merge, or compaction drops the
last descriptor reference, the pack remains forever, so managed BLOB storage
grows without bound under normal workloads. Build a live set from sidecars
reachable from retained snapshots, tags, and branches, then reclaim
unreferenced packs.
##########
src/paimon/core/io/primary_key_blob_externalizer.cpp:
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/primary_key_blob_externalizer.h"
+
+#include <map>
+#include <optional>
+#include <set>
+#include <string>
+#include <utility>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "fmt/format.h"
+#include "paimon/common/data/blob_defs.h"
+#include "paimon/common/data/blob_descriptor.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/defs.h"
+#include "paimon/format/file_format.h"
+#include "paimon/format/file_format_factory.h"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/writer_builder.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/memory/bytes.h"
+
+namespace paimon {
+
+namespace {
+
+/// The format managed blob packs are written in. Named once: the writer is
built through the
+/// format factory, and the same name is what an error has to report.
+constexpr const char kPackFormat[] = "blob";
+
+} // namespace
+
+/// Rolls `.managed.blob` packs for one managed blob field. Each written value
appends one blob
+/// format record to the current pack; the pack is sealed once it reaches the
blob target file
+/// size. Descriptors point straight at the payload bytes inside the pack, so
reading one back
+/// is a single ranged read that needs no pack footer.
+class PrimaryKeyBlobExternalizer::ManagedBlobPackWriter {
+ public:
+ ManagedBlobPackWriter(const CoreOptions& options, const
std::shared_ptr<arrow::Field>& field,
+ const std::shared_ptr<DataFilePathFactory>&
path_factory,
+ std::vector<std::string>* uncommitted_packs,
+ const std::shared_ptr<MemoryPool>& pool)
+ : options_(options),
+ field_(field),
+ path_factory_(path_factory),
+ uncommitted_packs_(uncommitted_packs),
+ target_file_size_(options.GetBlobTargetFileSize()),
+ pool_(pool) {}
+
+ /// Copies row `row` of `column` into the current pack and returns the
serialized
+ /// descriptor of the copied payload. The value may itself be a serialized
descriptor; the
+ /// blob format writer then streams the referenced bytes in,
re-materializing them.
+ Result<PAIMON_UNIQUE_PTR<Bytes>> Write(const
std::shared_ptr<arrow::Array>& column,
+ int64_t row) {
+ if (writer_ == nullptr) {
+ PAIMON_RETURN_NOT_OK(OpenCurrent());
+ }
+
+ std::shared_ptr<arrow::Array> element = column->Slice(row, 1);
+ std::shared_ptr<arrow::Array> pack_row;
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(pack_row,
+
arrow::StructArray::Make({std::move(element)}, {field_}));
+ ArrowArray c_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*pack_row,
&c_array));
+ PAIMON_RETURN_NOT_OK(writer_->AddBatch(&c_array));
+
+ // The format writer reports the payload bytes of the record it just
stored, so the
+ // record layout stays inside the format. Asked through the base
interface rather than
+ // by downcasting to the blob writer: the blob format lives in a
plugin library, so a
+ // checked cast would need its type info here, and an unchecked one
would be undefined
+ // behaviour the moment the factory returned anything else.
+ std::optional<std::pair<int64_t, int64_t>> payload_range =
writer_->LastPayloadRange();
+ if (!payload_range) {
+ return Status::Invalid(
+ fmt::format("Managed blob pack {} did not produce a payload
record. The '{}' "
+ "format must store one addressable payload per
record.",
+ current_path_, kPackFormat));
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<BlobDescriptor> descriptor,
+ BlobDescriptor::Create(current_path_, payload_range->first,
payload_range->second));
+ PAIMON_UNIQUE_PTR<Bytes> serialized = descriptor->Serialize(pool_);
+
+ PAIMON_ASSIGN_OR_RAISE(
+ bool reach_target_size,
+ writer_->ReachTargetSize(/*suggested_check=*/true,
target_file_size_));
+ if (reach_target_size) {
+ PAIMON_RETURN_NOT_OK(CloseCurrent());
+ }
+ return serialized;
+ }
+
+ /// Seals the current pack: writes the blob format footer and closes the
stream. The
+ /// underlying stream is closed even when writing the footer fails, so a
failed seal never
+ /// leaks the stream.
+ Status CloseCurrent() {
+ if (writer_ == nullptr) {
+ return Status::OK();
+ }
+ Status status = writer_->Finish();
+ Status close_status = out_->Close();
+ if (status.ok()) {
+ status = close_status;
+ }
+ writer_.reset();
+ out_.reset();
+ current_path_.clear();
+ return status;
+ }
+
+ /// Quietly drops the current pack writer; the file itself is removed
through the
+ /// uncommitted pack list.
+ void AbortCurrent() {
+ if (out_) {
+ [[maybe_unused]] Status status = out_->Close();
+ }
+ writer_.reset();
+ out_.reset();
+ current_path_.clear();
+ }
+
+ private:
+ Status OpenCurrent() {
+ std::string path = path_factory_->NewManagedBlobPath();
+ uncommitted_packs_->push_back(path);
+ // The blob format lives in its own plugin library, so the writer is
created through
+ // the format factory like every other core write path; core must not
reference the
+ // plugin's out-of-line symbols directly.
+ std::map<std::string, std::string> format_options = options_.ToMap();
+ // Managed pack writes never convert fetch failures to NULL payloads
and never
+ // interpret placeholder sentinels: strip the user-facing toggles so
the format
+ // defaults (false) apply.
+ format_options.erase(Options::BLOB_WRITE_NULL_ON_MISSING_FILE);
+ format_options.erase(Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE);
+ BlobDefs::EraseInternalPlaceholderOptions(&format_options);
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> format,
+ FileFormatFactory::Get(kPackFormat,
format_options));
+ ::ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ arrow::ExportSchema(*arrow::schema(arrow::FieldVector{field_}),
&c_schema));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<WriterBuilder> writer_builder,
+ format->CreateWriterBuilder(&c_schema,
/*batch_size=*/1));
+ writer_builder->WithMemoryPool(pool_);
+ if (auto* specific_fs_builder =
+ dynamic_cast<SpecificFSWriterBuilder*>(writer_builder.get())) {
+ specific_fs_builder->WithFileSystem(options_.GetFileSystem());
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<OutputStream> out,
+ options_.GetFileSystem()->Create(path,
/*overwrite=*/false));
+ out_ = std::move(out);
+ PAIMON_ASSIGN_OR_RAISE(writer_, writer_builder->Build(out_,
/*compression=*/"none"));
+ current_path_ = path;
+ return Status::OK();
+ }
+
+ CoreOptions options_;
+ std::shared_ptr<arrow::Field> field_;
+ std::shared_ptr<DataFilePathFactory> path_factory_;
+ std::vector<std::string>* uncommitted_packs_;
+ int64_t target_file_size_;
+ std::shared_ptr<MemoryPool> pool_;
+
+ std::string current_path_;
+ std::shared_ptr<OutputStream> out_;
+ std::unique_ptr<FormatWriter> writer_;
+};
+
+Result<std::unique_ptr<PrimaryKeyBlobExternalizer>>
PrimaryKeyBlobExternalizer::Create(
+ const CoreOptions& options, const std::shared_ptr<arrow::Schema>&
value_schema,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::shared_ptr<MemoryPool>& pool) {
+ std::vector<std::string> inline_field_names =
options.GetBlobInlineFields();
+ std::set<std::string> inline_fields(inline_field_names.begin(),
inline_field_names.end());
+ std::vector<std::string> managed_field_names =
+ BlobUtils::ManagedBlobFieldNames(value_schema, inline_fields);
+ if (managed_field_names.empty()) {
+ return std::unique_ptr<PrimaryKeyBlobExternalizer>();
+ }
+ // Guarded by SchemaValidation for tables created through the catalog;
checked again here
+ // because pack rolling cannot work with a non-positive target size.
+ if (options.GetBlobTargetFileSize() <= 0) {
+ return Status::Invalid(
+ fmt::format("Managed blob target file size must be positive, "
+ "but got {}.",
+ options.GetBlobTargetFileSize()));
+ }
+ std::vector<int32_t> managed_field_indices;
+ managed_field_indices.reserve(managed_field_names.size());
+ for (const auto& field_name : managed_field_names) {
+
managed_field_indices.push_back(value_schema->GetFieldIndex(field_name));
+ }
+ auto value_type = arrow::struct_(value_schema->fields());
+ return std::unique_ptr<PrimaryKeyBlobExternalizer>(new
PrimaryKeyBlobExternalizer(
+ options, value_type, std::move(managed_field_indices), path_factory,
pool));
+}
+
+PrimaryKeyBlobExternalizer::PrimaryKeyBlobExternalizer(
+ const CoreOptions& options, const std::shared_ptr<arrow::DataType>&
value_type,
+ std::vector<int32_t> managed_field_indices,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::shared_ptr<MemoryPool>& pool)
+ : options_(options),
+ value_type_(value_type),
+ managed_field_indices_(std::move(managed_field_indices)),
+ path_factory_(path_factory),
+ pool_(pool),
+ arrow_pool_(GetArrowPool(pool)),
+ logger_(Logger::GetLogger("PrimaryKeyBlobExternalizer")) {
+ auto struct_type = checked_pointer_cast<arrow::StructType>(value_type_);
+ pack_writers_.reserve(managed_field_indices_.size());
+ for (int32_t field_index : managed_field_indices_) {
+ pack_writers_.push_back(std::make_unique<ManagedBlobPackWriter>(
+ options_, struct_type->field(field_index), path_factory_,
&uncommitted_packs_, pool_));
+ }
+}
+
+PrimaryKeyBlobExternalizer::~PrimaryKeyBlobExternalizer() {
+ Abort();
+}
+
+Result<std::unique_ptr<RecordBatch>> PrimaryKeyBlobExternalizer::Externalize(
+ std::unique_ptr<RecordBatch>&& moved_batch) {
+ std::unique_ptr<RecordBatch> batch = std::move(moved_batch);
+ Result<std::unique_ptr<RecordBatch>> result = [&]() ->
Result<std::unique_ptr<RecordBatch>> {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array>
arrow_array,
+ arrow::ImportArray(batch->GetData(),
value_type_));
+ auto struct_array =
std::dynamic_pointer_cast<arrow::StructArray>(arrow_array);
+ if (struct_array == nullptr) {
+ return Status::Invalid(
+ "PrimaryKeyBlobExternalizer expects a StructArray record
batch.");
+ }
+ const std::vector<RecordBatch::RowKind>& row_kinds =
batch->GetRowKind();
+ if (!row_kinds.empty() &&
+ static_cast<int64_t>(row_kinds.size()) != struct_array->length()) {
+ return Status::Invalid(
+ "PrimaryKeyBlobExternalizer batch row kinds do not match the
row count.");
+ }
+ arrow::ArrayVector new_children = struct_array->fields();
+ for (size_t writer_index = 0; writer_index <
managed_field_indices_.size();
+ writer_index++) {
+ int32_t field_index = managed_field_indices_[writer_index];
+ const auto& column = struct_array->field(field_index);
+ auto blob_column =
std::dynamic_pointer_cast<arrow::LargeBinaryArray>(column);
+ if (blob_column == nullptr) {
+ return Status::Invalid(
+ fmt::format("PrimaryKeyBlobExternalizer expects managed
blob column {} to be a "
+ "LargeBinaryArray.",
+
struct_array->struct_type()->field(field_index)->name()));
+ }
+ // The member pool, not a local one: these buffers enter the write
buffer and
+ // must stay allocatable-from until the batch is flushed and
released.
+ arrow::LargeBinaryBuilder builder(arrow_pool_.get());
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(blob_column->length()));
+ for (int64_t row = 0; row < blob_column->length(); row++) {
+ bool retract =
+ !row_kinds.empty() && (row_kinds[row] ==
RecordBatch::RowKind::UPDATE_BEFORE ||
+ row_kinds[row] ==
RecordBatch::RowKind::DELETE);
+ // A retract row never keeps a payload: the managed value is
dropped without
+ // writing anything to a pack.
+ if (retract || blob_column->IsNull(row)) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendNull());
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> descriptor,
+
pack_writers_[writer_index]->Write(blob_column, row));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append(
+ reinterpret_cast<const uint8_t*>(descriptor->data()),
descriptor->size()));
+ }
+ std::shared_ptr<arrow::Array> descriptor_column;
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&descriptor_column));
+ new_children[field_index] = std::move(descriptor_column);
+ }
+
+ auto struct_type =
checked_pointer_cast<arrow::StructType>(value_type_);
+ std::shared_ptr<arrow::StructArray> externalized_array;
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ externalized_array, arrow::StructArray::Make(new_children,
struct_type->fields()));
+ ArrowArray c_array;
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*externalized_array,
&c_array));
+ return std::make_unique<RecordBatch>(batch->GetPartition(),
batch->GetBucket(), row_kinds,
+ &c_array);
+ }();
+ if (!result.ok()) {
Review Comment:
`Abort()` deletes every successfully written entry in `uncommitted_packs_`,
even though the owning writer may already hold buffered or pending data that
references those packs. Neither writer is poisoned after the error, so a later
`PrepareCommit` can commit dangling descriptors; the same issue occurs after a
seal failure because `new_files_` has already been flushed. The failure path
must either roll back the pending data and sidecars and make the writer
unusable, or delete only packs not referenced by pending files.
##########
src/paimon/core/append/append_compact_coordinator.cpp:
##########
@@ -361,4 +859,210 @@ Result<std::vector<std::shared_ptr<CommitMessage>>>
AppendCompactCoordinator::Ru
core_options, executor, pool);
}
+Result<int32_t> AppendCompactCoordinator::RunAndCommit(
+ const std::string& table_path, const std::map<std::string, std::string>&
options,
+ const std::vector<std::map<std::string, std::string>>& partitions,
+ const std::string& commit_user, const std::shared_ptr<FileSystem>&
file_system,
+ const std::shared_ptr<MemoryPool>& input_pool, int64_t
candidate_files_per_round) {
+ if (commit_user.empty()) {
+ return Status::Invalid("AppendCompactCoordinator::RunAndCommit needs a
commit user.");
+ }
+ if (candidate_files_per_round <= 0) {
+ return Status::Invalid(fmt::format(
+ "candidate_files_per_round must be positive, but was {}.",
candidate_files_per_round));
+ }
+ auto pool = input_pool ? input_pool : GetDefaultPool();
+ std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+ auto logger = Logger::GetLogger("AppendCompactCoordinator");
+
+ std::pair<std::shared_ptr<TableSchema>, CoreOptions> schema_and_options;
+ PAIMON_ASSIGN_OR_RAISE(schema_and_options,
+ LoadSchemaAndOptions(table_path, options,
file_system));
+ const auto& [table_schema, core_options] = schema_and_options;
+ PAIMON_RETURN_NOT_OK(ValidateTable(table_schema, core_options));
+
+ auto arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<arrow::Schema> partition_schema,
+ FieldMapping::GetPartitionSchema(arrow_schema,
table_schema->PartitionKeys()));
+ auto snapshot_manager =
+ std::make_shared<SnapshotManager>(core_options.GetFileSystem(),
table_path);
+ auto schema_manager =
std::make_shared<SchemaManager>(core_options.GetFileSystem(), table_path);
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ BuildPathFactory(table_path, table_schema, arrow_schema, core_options,
pool));
+
+ if (!core_options.DataEvolutionEnabled()) {
+ // A plain append table has no row id space to split on, so it keeps
the single-round
+ // behaviour and only gains the commit.
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<CommitMessage>>
messages,
+ Run(table_path, options, partitions,
file_system, pool));
+ if (messages.empty()) {
+ return 0;
+ }
+ PAIMON_RETURN_NOT_OK(
+ CommitRound(table_path, commit_user, core_options, messages,
executor, pool));
+ return 1;
+ }
+ if (core_options.DataEvolutionCompactionRewriteRowIds()) {
+ return Status::Invalid(fmt::format(
+ "'{}' is no longer supported: normal data-evolution compaction
preserves row ids "
+ "and logical deletions.",
+ Options::DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS));
+ }
+
+ // The windows are planned once, against the snapshot the run starts from;
every round
+ // then re-scans and sees the rounds committed before it. That re-scan
re-reads the
+ // snapshot and the manifest list, but not most of the manifest files: a
manifest is
+ // pruned by its recorded row id range, and because windows are cut only
at coverage gaps
+ // it belongs to exactly one round.
`DataEvolutionCompactPlanner::PlanRowIdWindows`
+ // documents the one exception, a delete-only manifest without row id
statistics.
+ PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot,
snapshot_manager->LatestSnapshot());
+ if (!snapshot.has_value()) {
+ return 0;
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<ManifestList> manifest_list,
+ ManifestList::Create(core_options.GetFileSystem(),
core_options.GetManifestFormat(),
+ core_options.GetManifestCompression(),
path_factory,
+ core_options.GetCache(), pool));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<PredicateFilter> partition_filter,
+ FileStoreScan::CreatePartitionPredicate(
+ table_schema->PartitionKeys(),
+ core_options.GetPartitionDefaultName(),
arrow_schema, partitions));
+ PAIMON_ASSIGN_OR_RAISE(std::vector<Range> windows,
+ DataEvolutionCompactPlanner::PlanRowIdWindows(
+ manifest_list, snapshot.value(),
partition_filter, partition_schema,
+ candidate_files_per_round, logger.get()));
+ if (windows.empty()) {
+ // Nothing to split on; one unbounded round still commits correctly.
+ windows.emplace_back(0, std::numeric_limits<int64_t>::max());
+ }
+ PAIMON_LOG_DEBUG(logger, "Compacting table %s in %zu round(s)",
table_path.c_str(),
+ windows.size());
+
+ int32_t committed_rounds = 0;
+ for (const auto& window : windows) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::shared_ptr<CommitMessage>> messages,
+ RunDataEvolutionRound(table_path, snapshot_manager,
schema_manager, table_schema,
+ arrow_schema, partition_schema,
core_options, path_factory,
+ partitions, window, executor, pool));
+ if (messages.empty()) {
+ continue;
+ }
+ Status commit_status =
+ CommitRound(table_path, commit_user, core_options, messages,
executor, pool);
+ if (!commit_status.ok()) {
+ // The round's rewritten data files are unreachable once its
commit failed, and the
+ // rounds already committed stay committed: compaction is
idempotent, a later run
+ // re-plans whatever is left. A deletion-vector index file the
round may have
+ // written is left to orphan cleaning, which is what collects
unreferenced index
+ // files.
+ CleanupCompactOutputs(messages, path_factory, core_options);
Review Comment:
`FileStoreCommitImpl::TryCommitOnce` deliberately treats an atomic
snapshot-write error as an uncertain result and keeps outputs for recovery via
`FilterAndCommit`. This branch instead deletes every `CompactAfter` on any
commit error; if the snapshot was committed but the client timed out, the
latest snapshot now references missing data files. Give each round a
recoverable unique identifier and delete outputs only after confirming the
commit did not land; otherwise keep them.
##########
src/paimon/core/operation/commit/conflict_detection.cpp:
##########
@@ -434,11 +801,76 @@ Status ConflictDetection::CheckKeyRange(const
std::vector<ManifestEntry>& merged
Status ConflictDetection::CheckRowIdExistence(const
std::vector<ManifestEntry>& base_entries,
const
std::vector<ManifestEntry>& delta_entries,
- const std::optional<int64_t>&
next_row_id) const {
+ const std::optional<int64_t>&
next_row_id,
+ const Snapshot::CommitKind&
commit_kind) const {
if (!options_.DataEvolutionEnabled()) {
return Status::OK();
}
+ std::vector<const ManifestEntry*> existing_data_files;
Review Comment:
This path does not first verify that a delta DELETE still matches the
current file with the same `Identifier` by `first_row_id` and `row_count`.
Because those fields are not part of `Identifier`, a stale DELETE can cancel a
concurrently reassigned ADD; the materialization output has no row ID and
therefore skips the later range check, potentially replacing current data with
a stale rewrite. Compare the base ADD and delta DELETE metadata first and
reject the commit on any mismatch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
