wangyong9999 commented on code in PR #245:
URL: https://github.com/apache/paimon-cpp/pull/245#discussion_r3850606450
##########
src/paimon/core/operation/abstract_file_store_write.cpp:
##########
@@ -208,6 +211,29 @@ Result<std::vector<std::shared_ptr<CommitMessage>>>
AbstractFileStoreWrite::Prep
compact_increment.AddNewIndexFiles({dv_index_file_meta.value()});
}
}
+ if (writer_container.primary_key_index_maintainer) {
+ Status index_status =
+
writer_container.primary_key_index_maintainer->PrepareCommit(&increment);
+ if (!index_status.ok()) {
+ if (compact_deletion_file) {
+ const auto& new_index_files =
+ increment.GetCompactIncrement().NewIndexFiles();
+ for (const std::shared_ptr<IndexFileMeta>& index_file
: new_index_files) {
+ if (index_file != nullptr &&
+ index_file->IndexType() ==
+
DeletionVectorsIndexFile::DELETION_VECTORS_INDEX) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::string index_path,
+
dv_maintainer_factory_->GetIndexFileHandler()->FilePath(
+ partition, bucket, index_file));
+ [[maybe_unused]] Status cleanup_status =
+
options_.GetFileSystem()->Delete(index_path);
+ }
+ }
+ }
+ return index_status;
Review Comment:
Returning here discards `increment` — which `writer->PrepareCommit()`
already drained out of the writer — and also every `CommitMessage` collected
for the buckets processed earlier in this loop. A retry gets empty increments
from those writers, so data files that were written and would have been
committed silently drop out of the commit and stay on disk as orphans.
That window now holds a heavyweight, failure-prone step: a full re-read of a
data level plus an external sort, which hard-fails when the sort quota is
exhausted (`pk_sorted_index_builder.cpp`, "external-sort quota is exhausted").
A table with a level larger than `write-buffer-size` and no spill directory
fails every commit.
The read path is already built to tolerate this — `PkSortedBucketIndexState`
documents that levels without a valid group stay uncovered and are scanned
normally, and Java builds asynchronously for the same reason. Suggest degrading
a maintainer failure to "this level has no payload" (log, delete the payloads
built in this attempt, keep committing), or at least running the build before
the writer increment is drained.
##########
src/paimon/core/io/key_value_in_memory_record_reader.cpp:
##########
@@ -110,6 +115,21 @@ void KeyValueInMemoryRecordReader::Close() {
Result<std::shared_ptr<arrow::NumericArray<arrow::UInt64Type>>>
KeyValueInMemoryRecordReader::SortBatch() const {
+ if (sort_comparator_ != nullptr) {
+ std::vector<uint64_t>
indices(static_cast<size_t>(value_struct_array_->length()));
+ std::iota(indices.begin(), indices.end(), 0);
+ std::stable_sort(indices.begin(), indices.end(), [&](uint64_t left,
uint64_t right) {
+ ColumnarRowRef left_row(value_ctx_, left);
Review Comment:
`ColumnarRowRef`'s constructor takes `std::shared_ptr<ColumnarBatchContext>`
by value, so each comparison copies two shared pointers — about 2·n·log₂n
atomic refcount round-trips, ~40M for a one-million-row level, none of which
does any comparing.
Build the row refs once and index into them:
```cpp
std::vector<ColumnarRowRef> rows;
rows.reserve(indices.size());
for (uint64_t i = 0; i < indices.size(); ++i) rows.emplace_back(value_ctx_,
i);
std::stable_sort(indices.begin(), indices.end(), [&](uint64_t left, uint64_t
right) {
return sort_comparator_->CompareTo(rows[left], rows[right]) < 0;
});
```
--
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]
