wangyong9999 commented on code in PR #258:
URL: https://github.com/apache/paimon-cpp/pull/258#discussion_r3912441123
##########
src/paimon/core/table/source/data_evolution_batch_scan.cpp:
##########
@@ -44,27 +90,43 @@ 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 (UsesUnsupportedTimeTravel(core_options_)) {
+ return Status::NotImplemented("Global index scan does not support time
travel");
}
- PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index,
- RowRangeIndex::Create(row_ranges.value()));
+ PAIMON_ASSIGN_OR_RAISE(std::vector<Range> row_ranges,
final_global_index_result->ToRanges());
+ if (row_ranges.empty()) {
+ if (!global_index_snapshot_id) {
+ const std::shared_ptr<SnapshotManager>& snapshot_manager =
+ snapshot_reader_->GetSnapshotManager();
+ PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot,
+
ResolveGlobalIndexScanSnapshot(core_options_, snapshot_manager));
+ if (!snapshot) {
Review Comment:
Fixed in 74003307. Supplied empty results now take the selected snapshot
from normal batch planning; we no longer infer it through a separate resolver.
Tag/timestamp are rejected before evaluating a Global Index candidate, and the
shared scanner change is removed.
##########
src/paimon/core/table/source/data_evolution_batch_scan.cpp:
##########
@@ -134,27 +144,41 @@ Result<std::shared_ptr<Plan>>
DataEvolutionBatchScan::WrapToIndexedSplits(
return std::make_shared<PlanImpl>(data_plan->SnapshotId(), indexed_splits);
}
-Result<std::shared_ptr<GlobalIndexResult>>
DataEvolutionBatchScan::EvalGlobalIndex() const {
+Result<std::optional<DataEvolutionBatchScan::EvaluatedGlobalIndex>>
+DataEvolutionBatchScan::EvalGlobalIndex() const {
auto predicate = batch_scan_->GetNonPartitionPredicate();
if (!predicate) {
- return std::shared_ptr<GlobalIndexResult>(nullptr);
+ return std::optional<EvaluatedGlobalIndex>();
}
if (!core_options_.GlobalIndexEnabled()) {
- return std::shared_ptr<GlobalIndexResult>(nullptr);
+ return std::optional<EvaluatedGlobalIndex>();
}
auto partition_filter = batch_scan_->GetPartitionPredicate();
// TODO(lisizhuo.lsz): support time travel
+ PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> snapshot_id,
ResolveGlobalIndexSnapshotId());
+ if (!snapshot_id) {
+ return Status::Invalid("not found latest snapshot");
+ }
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<GlobalIndexScan> index_scan,
- GlobalIndexScan::Create(table_path_,
core_options_.GetScanSnapshotId(), partition_filter,
- core_options_.ToMap(),
core_options_.GetFileSystem(), executor_,
- pool_));
+ GlobalIndexScan::Create(table_path_, snapshot_id, partition_filter,
core_options_.ToMap(),
+ core_options_.GetFileSystem(), executor_,
pool_));
auto index_scan_impl =
dynamic_cast<GlobalIndexScanImpl*>(index_scan.get());
if (!index_scan_impl) {
return Status::Invalid("invalid GlobalIndexScan, cannot cast to
GlobalIndexScanImpl");
}
- return index_scan_impl->Scan(predicate);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult> result,
+ index_scan_impl->Scan(predicate));
+ return std::optional<EvaluatedGlobalIndex>(EvaluatedGlobalIndex{result,
snapshot_id.value()});
+}
+
+Result<std::optional<int64_t>>
DataEvolutionBatchScan::ResolveGlobalIndexSnapshotId() const {
+ std::optional<int64_t> snapshot_id = core_options_.GetScanSnapshotId();
+ if (snapshot_id) {
+ return snapshot_id;
+ }
Review Comment:
Updated in 74003307. Supplied empty results now use normal batch planning,
so snapshot validation stays on the regular path. Tag/timestamp are rejected as
soon as a Global Index path is requested, including predicates without an
applicable index.
--
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]
