wangyong9999 commented on code in PR #258:
URL: https://github.com/apache/paimon-cpp/pull/258#discussion_r3905170654
##########
src/paimon/core/table/source/data_evolution_batch_scan.cpp:
##########
@@ -42,25 +43,36 @@ DataEvolutionBatchScan::DataEvolutionBatchScan(
executor_(executor) {}
Result<std::shared_ptr<Plan>> DataEvolutionBatchScan::CreatePlan() {
- std::optional<std::vector<Range>> row_ranges;
+ std::optional<int64_t> global_index_snapshot_id;
std::shared_ptr<GlobalIndexResult> final_global_index_result =
global_index_result_;
if (!final_global_index_result) {
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult>
index_result, EvalGlobalIndex());
- if (index_result) {
- final_global_index_result = index_result;
- PAIMON_ASSIGN_OR_RAISE(row_ranges, index_result->ToRanges());
+ PAIMON_ASSIGN_OR_RAISE(std::optional<EvaluatedGlobalIndex>
evaluated_index,
+ EvalGlobalIndex());
+ if (evaluated_index) {
+ final_global_index_result = evaluated_index->result;
+ global_index_snapshot_id = evaluated_index->snapshot_id;
}
- } else {
- PAIMON_ASSIGN_OR_RAISE(row_ranges,
final_global_index_result->ToRanges());
}
- if (!row_ranges) {
+ if (!final_global_index_result) {
return batch_scan_->CreatePlan();
}
- if (row_ranges.value().empty()) {
- return PlanImpl::EmptyPlan();
+ if (core_options_.GetScanTagName() ||
core_options_.GetScanTimestampMillis()) {
+ return Status::NotImplemented("Global index scan does not support time
travel");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<Range> row_ranges,
final_global_index_result->ToRanges());
+ if (row_ranges.empty()) {
+ if (!global_index_snapshot_id) {
+ PAIMON_ASSIGN_OR_RAISE(global_index_snapshot_id,
ResolveGlobalIndexSnapshotId());
+ if (!global_index_snapshot_id) {
+ return PlanImpl::EmptyPlan();
+ }
+
PAIMON_RETURN_NOT_OK(snapshot_reader_->GetSnapshotManager()->LoadSnapshot(
+ global_index_snapshot_id.value()));
+ }
+ return std::make_shared<PlanImpl>(global_index_snapshot_id,
+
std::vector<std::shared_ptr<Split>>());
}
- PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index,
- RowRangeIndex::Create(row_ranges.value()));
+ PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index,
RowRangeIndex::Create(row_ranges));
batch_scan_->WithRowRangeIndex(row_range_index);
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> data_plan,
batch_scan_->CreatePlan());
Review Comment:
The index side now resolves its snapshot explicitly, but the data scan here
still resolves latest on its own. A commit landing between the two applies row
ranges computed at snapshot N to a data plan at N+1: matches added in N+1 are
missed and the plan reports N+1 while the selection came from N. The race
predates this PR, but with the resolved id now in hand it is cheap to close:
when global_index_snapshot_id is set, compare data_plan->SnapshotId() against
it and fail on mismatch (or pin the data scan to that id if the API allows).
--
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]
