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` 可能包含 metadata-only upgrade：它的 `CompactBefore/After` 复用同一个
`file_name`。这里在后续 bucket prepare 失败时调用 cleaner，会把该文件当成未提交输出删除，旧 snapshot
仍然引用它，数据文件和 `.blobref` 都会丢失。Cleaner 删除 `CompactAfter` 前需排除同时存在于 `CompactBefore`
的文件，沿用 `MergeTreeWriter::DoClose` 的判断。
##########
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:
这里过期数据文件时只删除 data file 和 `.blobref`；orphan cleaner 又永久跳过 `.managed.blob`
且不支持主键表。正常 update、first-row merge 或 compaction 让 pack 失去最后引用后，它仍会永久占用空间，存储会随历史
BLOB 无界增长。需要基于所有保留 snapshot/tag/branch 的 sidecar 构建 live set，再回收无引用 pack。
##########
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()` 会删除此前成功写入的全部 `uncommitted_packs_`，但 owner writer 里可能已经保留引用这些 pack
的 buffered/pending data。两个 writer 都没有失败状态，之后再次 `PrepareCommit` 会提交悬空
descriptor；封包失败时 `new_files_` 已经 flush，问题同样存在。失败时必须同步回滚 pending data/sidecar
并禁止继续提交，或只删除尚未被 pending 文件引用的 pack。
##########
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` 明确把 snapshot 原子写异常视为结果不确定，并保留输出等待
`FilterAndCommit`。这里却对所有 commit error 立即删除 `CompactAfter`；服务端已提交、客户端超时时，最新
snapshot 会直接引用不存在的数据文件。每轮需使用可恢复的唯一 identifier，确认提交未生效后才能清理；无法确认时应保留输出。
##########
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:
这里没有先校验 delta DELETE 与当前同一 `Identifier` 的 `first_row_id/row_count`。由于
Identifier 不包含这两个字段，文件被并发重新分配后 stale DELETE 仍会抵消当前 ADD；materialize 输出又没有 row
ID，后续 range 检查会跳过，最终可能用旧数据替换当前文件。应先比较 base ADD 与 delta DELETE 的 row
metadata，不一致直接拒绝提交。
--
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]
