lxy-9602 commented on code in PR #258:
URL: https://github.com/apache/paimon-cpp/pull/258#discussion_r3885882106
##########
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:
This behavior seems to predate the PR: the global-index scan already appears
to ignore tag- and timestamp-based time travel and fall back to the latest
snapshot. What changes here is that the new resolver makes this more visible by
attaching that latest snapshot ID to an empty plan. Since global-index time
travel does not seem to be supported at the moment, would it make sense to
reject these modes explicitly instead of silently scanning and reporting the
latest snapshot?
##########
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:
Please validate the snapshot before returning the empty plan. With a
pre-supplied empty `GlobalIndexResult`, a nonexistent `scan.snapshot-id` (for
example, 999) is currently copied directly into a successful plan. Please
resolve/load the snapshot here and add a regression test for nonexistent or
expired snapshot IDs.
--
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]
