zjw1111 commented on code in PR #213:
URL: https://github.com/apache/paimon-cpp/pull/213#discussion_r3813342667
##########
src/paimon/core/operation/file_store_commit_impl.cpp:
##########
@@ -929,6 +929,19 @@ Result<int64_t> FileStoreCommitImpl::CommitWithProgress(
commit_messages.push_back(realtime_commit.commit_message);
}
+ PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> latest_snapshot,
+ snapshot_manager_->LatestSnapshot());
+ PAIMON_ASSIGN_OR_RAISE(RealtimeOffsetMap committed_offsets,
+
RealtimeCommitProperties::ReadOffsets(latest_snapshot, fs_));
+ PAIMON_ASSIGN_OR_RAISE(bool ranges_committed,
RealtimeCommitProperties::AreRangesCommitted(
Review Comment:
Question on the idempotency model here, not a change request.
This short-circuit decides "already committed" purely from realtime
offset-range coverage. It does not look at `(commitUser, commitIdentifier)`,
and it does not run `CheckFilesExistence`. Both differ from the non-realtime
path in this same file: `FilterAndCommit` calls `CheckFilesExistence` and
`FilterCommitted`, and `FilterCommitted` resolves the committed identifier via
`LatestSnapshotOfUser(commit_user_)` — matching Java's
`FileStoreCommitImpl.filterCommitted`. `CommitWithProgress` -> `Commit` ->
`TryCommit` bypasses both.
Two consequences I want to check are intentional:
1. Two writers sharing the same offset ranges are indistinguishable here. If
a zombie writer replays ranges that the current writer already committed, this
returns `latest_snapshot->Id()` and silently drops `commit_messages`, whereas
before this patch `RealtimeCommitProperties::Build` would have rejected the
commit as non-contiguous. Coverage of the offset range is treated as proof that
*this* writer's files are already in the snapshot, which only holds if offsets
are globally owned by one writer.
2. Without `CheckFilesExistence`, a retry after partial cleanup can be
accepted on offset coverage alone even if the referenced data files are gone.
Is offset coverage intended to be the sole idempotency key for realtime
commits — i.e. offset ownership is guaranteed single-writer by contract above
this layer — or is `(commitUser, commitIdentifier)` + `CheckFilesExistence`
still expected to be layered in later? The surviving `TODO(xinyu.lxy): Support
failure recovery and idempotent retry for real-time commits.` in `TryCommit`
suggests the latter, but this block reads like the former.
##########
src/paimon/core/realtime/arrow_realtime_store.cpp:
##########
@@ -264,16 +389,23 @@ Status ArrowRealtimeStore::Write(RealtimeWriteBatch&&
write_batch) {
}
std::shared_ptr<arrow::StructArray> struct_array =
checked_pointer_cast<arrow::StructArray>(data);
+ PAIMON_ASSIGN_OR_RAISE(std::optional<BatchStatistics> statistics,
Review Comment:
Question on the intended scope of the "optional pruning aid" contract, not a
change request.
`MayMatch` states it explicitly:
```cpp
// Statistics are only an optional pruning aid. An unsupported predicate or
incomplete
// statistic must retain the batch to avoid false negatives.
return !result.ok() || result.value();
```
and the read path is tested for it
(`TestMissingStatisticsRetainsNonMatchingBatch` tolerates `stored.min_values ==
nullptr`).
The write path here does the opposite: `CollectStatistics` failure
propagates out of `Write` and fails the write. `CollectStatistics` depends on
`arrow::compute::MinMax` and `arrow::StructArray::Make`, so any
type/kernel/allocation failure there turns into a write failure. Net effect:
`realtime.store.stats-mode=full` introduces a write failure mode that `none`
does not have, for data that is otherwise perfectly writable — statistics are
optional for pruning but mandatory for ingestion.
Is the "optional" contract meant to apply only to reads, with statistics
collection being a hard requirement on write? Or should a failure here degrade
to "no statistics for this batch" the way a missing statistic already degrades
on the read side?
##########
src/paimon/core/operation/expire_snapshots.cpp:
##########
@@ -157,8 +158,22 @@ Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t
earliest_snapshot_id,
std::vector<Snapshot> retained_snapshots;
PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot,
snapshot_manager_->LoadSnapshot(end_exclusive_id));
retained_snapshots.push_back(snapshot);
+ std::set<std::string> retained_offset_files;
+ PAIMON_ASSIGN_OR_RAISE(std::vector<Snapshot> all_snapshots,
Review Comment:
Perf point worth confirming — flagging only, no change requested.
`GetAllSnapshots()` is now called unconditionally on every expire round,
purely to build `retained_offset_files`. Before this patch the retained side of
`ExpireUntil` only did a point `LoadSnapshot(end_exclusive_id)`, so this adds a
full snapshot-directory listing plus one snapshot read per retained snapshot,
i.e. O(snapshot count) extra IO per expire round. On object storage that is a
listing round-trip plus N GETs, and it happens even for tables that never
produced a `.offsets` file at all — the common non-realtime case pays the full
cost to build an empty set.
Please confirm this is acceptable for tables with a large retained snapshot
count, or that you have measured it and it is noise relative to the manifest
work already done in this loop.
##########
src/paimon/core/operation/file_store_commit_impl.cpp:
##########
@@ -929,6 +929,19 @@ Result<int64_t> FileStoreCommitImpl::CommitWithProgress(
commit_messages.push_back(realtime_commit.commit_message);
}
+ PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> latest_snapshot,
+ snapshot_manager_->LatestSnapshot());
Review Comment:
Perf point worth confirming — flagging only, no change requested.
`LatestSnapshot()` here (line 933) and `ReadOffsets()` (line 935) are both
repeated further down the same commit path: `TryCommitOnce` does
`snapshot_manager_->LatestSnapshot()` again at line 978, and
`RealtimeCommitProperties::Build` re-reads the offsets file. So a single
realtime commit now does two latest-snapshot resolutions and two `.offsets`
reads, where the offsets read is a metadata-file GET on object storage.
That is on the hot path for every realtime commit, and since `Commit` is
invoked with `/*retry_on_conflict=*/false` (line 949) there is no retry loop to
amortize it against.
Please confirm the duplication is intentional (e.g. you want the coverage
check against a snapshot read strictly before the commit attempt, independent
of what `TryCommit` observes) rather than an accidental double read.
##########
src/paimon/core/realtime/arrow_realtime_store_test.cpp:
##########
@@ -86,6 +94,21 @@ class ArrowRealtimeStoreTest : public testing::Test {
return c_schema;
}
+ std::vector<int64_t> ReadIds(const BatchReader::ReadBatchWithBitmap&
batch) const {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ImportArray(batch.first.first.get(),
batch.first.second.get()).ValueOrDie();
+ std::shared_ptr<arrow::StructArray> struct_array =
+ std::static_pointer_cast<arrow::StructArray>(array);
Review Comment:
`docs/code-style.md` requires `checked_pointer_cast<T>` instead of
`std::static_pointer_cast<T>`, and the example given there is literally this
cast:
```cpp
// Good
auto struct_array = checked_pointer_cast<arrow::StructArray>(array);
```
Applies to line 103 as well. Please switch both to `checked_pointer_cast`.
##########
src/paimon/core/realtime/arrow_realtime_store_test.cpp:
##########
@@ -86,6 +94,21 @@ class ArrowRealtimeStoreTest : public testing::Test {
return c_schema;
}
+ std::vector<int64_t> ReadIds(const BatchReader::ReadBatchWithBitmap&
batch) const {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ImportArray(batch.first.first.get(),
batch.first.second.get()).ValueOrDie();
+ std::shared_ptr<arrow::StructArray> struct_array =
+ std::static_pointer_cast<arrow::StructArray>(array);
+ std::shared_ptr<arrow::Int64Array> ids =
+
std::static_pointer_cast<arrow::Int64Array>(struct_array->field(/*id=*/1));
Review Comment:
Nit: the parameter of `arrow::StructArray::field` is named `pos`, not `id`.
`.clang-tidy` enables `bugprone-argument-comment`, so this mismatched comment
is a lint violation. Please use `/*pos=*/1`.
##########
include/paimon/realtime/realtime_store.h:
##########
@@ -40,6 +40,14 @@ namespace paimon {
class MemoryPool;
class Predicate;
+/// Statistics collected by a `RealtimeStore` for query predicate pushdown.
+enum class StatisticsMode {
Review Comment:
Question on the intended semantics of this enum name.
`StatisticsMode` is declared in top-level `namespace paimon` (and
forward-declared in `core_options.h`), with a fully generic name. But
everything about it is realtime-specific: it lives in `realtime_store.h`, the
doc comment says "collected by a `RealtimeStore`", the option is
`realtime.store.stats-mode`, and the accessor is
`GetRealtimeStoreStatisticsMode()`.
The values are also a strict subset of Java's `metadata.stats-mode`, which
is `none | counts | truncate(n) | full` with default `truncate(16)`
(`CoreOptions.java`); Rust `core_options.rs` mirrors that set. Here it is `NONE
| FULL` only, and `FULL` does not truncate STRING/BINARY min/max.
So: is `StatisticsMode` meant to eventually carry the same semantics as
Java's `metadata.stats-mode` (in which case the missing `counts`/`truncate`
values and the realtime-only header placement are staging steps), or is it
deliberately a realtime-private concept that just happens to share the name?
Judging by the name alone a reader will assume the former; judging by the
implementation it is the latter. Worth pinning down now, since the name is
already exposed at `paimon` scope and hard to walk back later.
##########
src/paimon/core/core_options.h:
##########
@@ -107,6 +108,8 @@ class PAIMON_EXPORT CoreOptions {
std::optional<int64_t> GetScanSnapshotId() const;
std::optional<int64_t> GetScanTimestampMillis() const;
int64_t GetRealtimeReadViewTtlMillis() const;
+ /// Returns the statistics mode used by real-time memory indexers.
Review Comment:
"memory indexers" is stale terminology. The concept is `RealtimeStore` now —
the option is `realtime.store.stats-mode`, the accessor is
`GetRealtimeStoreStatisticsMode`, the enum doc in `realtime_store.h` says
"collected by a `RealtimeStore`", and this PR itself renames indexer -> store
elsewhere (e.g. `TestReusesIndexer...` -> `TestReusesStore...` in
`realtime_context_test.cpp`).
Suggest: `/// Returns the statistics mode used by real-time stores.`
##########
src/paimon/core/operation/expire_snapshots.cpp:
##########
@@ -169,10 +184,22 @@ Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t
earliest_snapshot_id,
PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot,
snapshot_manager_->LoadSnapshot(id));
PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.BaseManifestList(),
skipping_sets));
PAIMON_RETURN_NOT_OK(CleanUnusedManifests(snapshot.DeltaManifestList(),
skipping_sets));
+ std::optional<std::string> offsets_path =
+ RealtimeCommitProperties::GetOffsetsPath(snapshot);
+ if (offsets_path) {
+ expired_offset_files.insert(offsets_path.value());
+ }
auto status = fs_->Delete(snapshot_manager_->SnapshotPath(id));
// delete quietly will ignore any status error
(void)status;
}
+ for (const std::string& offsets_path : expired_offset_files) {
+ if (retained_offset_files.count(offsets_path) == 0) {
Review Comment:
This `retained_offset_files` guard is the one thing preventing a live
`.offsets` file from being deleted, and nothing in the test suite exercises the
case it exists for.
Sharing happens because `RealtimeCommitProperties::Build` forward-copies
`kOffsetsKey` when `realtime_ranges.empty()`, so a non-realtime snapshot
(compaction being the obvious one) inherits the `.offsets` path of the realtime
snapshot before it. Expiring the realtime snapshot then puts that path into
`expired_offset_files` while a retained snapshot still references it — exactly
what the `count(...) == 0` check must catch.
`TestRealtimeOffsetFileLifecycle` does not reach that state: it uses two
distinct `.offsets` paths and never commits a compaction. Across
`test/inte/realtime_write_inte_test.cpp`, `ExpireSnapshots()` only appears in
that one test and `CompactAndCommit` only in a different one, so no test ever
has an expiring and a retained snapshot pointing at the same `.offsets` file.
If the guard regresses, the failure is silent data loss — the offsets file is
gone and the realtime data referenced by the retained snapshot becomes
unreadable — and CI stays green.
Please add a test: realtime commit -> compaction commit (inherits the same
`.offsets` via the empty-ranges forward-copy) -> expire the realtime snapshot
-> assert the `.offsets` file still exists and `ReadOffsets` on the retained
snapshot still returns the committed ranges.
--
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]
