From 2882c782707d1c99f84530bab8b31b572c139553 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Fri, 28 Aug 2026 20:20:04 -0400 Subject: [PATCH] fix(scan): preserve snapshot id for empty data evolution index plans --- .../source/data_evolution_batch_scan.cpp | 48 ++++++++++++++----- .../table/source/data_evolution_batch_scan.h | 9 +++- test/inte/global_index_test.cpp | 47 ++++++++++++++++++ 3 files changed, 91 insertions(+), 13 deletions(-) diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.cpp b/src/paimon/core/table/source/data_evolution_batch_scan.cpp index ed09b693b..2a461bfb6 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp @@ -43,12 +43,15 @@ DataEvolutionBatchScan::DataEvolutionBatchScan( Result> DataEvolutionBatchScan::CreatePlan() { std::optional> row_ranges; + std::optional global_index_snapshot_id; std::shared_ptr final_global_index_result = global_index_result_; if (!final_global_index_result) { - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr 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 evaluated_index, + EvalGlobalIndex()); + if (evaluated_index && evaluated_index->result) { + final_global_index_result = evaluated_index->result; + global_index_snapshot_id = evaluated_index->snapshot_id; + PAIMON_ASSIGN_OR_RAISE(row_ranges, evaluated_index->result->ToRanges()); } } else { PAIMON_ASSIGN_OR_RAISE(row_ranges, final_global_index_result->ToRanges()); @@ -57,7 +60,14 @@ Result> DataEvolutionBatchScan::CreatePlan() { return batch_scan_->CreatePlan(); } if (row_ranges.value().empty()) { - return PlanImpl::EmptyPlan(); + if (!global_index_snapshot_id) { + PAIMON_ASSIGN_OR_RAISE(global_index_snapshot_id, ResolveGlobalIndexSnapshotId()); + } + if (!global_index_snapshot_id) { + return PlanImpl::EmptyPlan(); + } + return std::make_shared(global_index_snapshot_id, + std::vector>()); } PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index, RowRangeIndex::Create(row_ranges.value())); @@ -134,27 +144,41 @@ Result> DataEvolutionBatchScan::WrapToIndexedSplits( return std::make_shared(data_plan->SnapshotId(), indexed_splits); } -Result> DataEvolutionBatchScan::EvalGlobalIndex() const { +Result> +DataEvolutionBatchScan::EvalGlobalIndex() const { auto predicate = batch_scan_->GetNonPartitionPredicate(); if (!predicate) { - return std::shared_ptr(nullptr); + return std::optional(); } if (!core_options_.GlobalIndexEnabled()) { - return std::shared_ptr(nullptr); + return std::optional(); } auto partition_filter = batch_scan_->GetPartitionPredicate(); // TODO(lisizhuo.lsz): support time travel + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot_id, ResolveGlobalIndexSnapshotId()); + if (!snapshot_id) { + return Status::Invalid("not found latest snapshot"); + } PAIMON_ASSIGN_OR_RAISE( std::unique_ptr 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(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 result, + index_scan_impl->Scan(predicate)); + return std::optional(EvaluatedGlobalIndex{result, snapshot_id.value()}); +} + +Result> DataEvolutionBatchScan::ResolveGlobalIndexSnapshotId() const { + std::optional snapshot_id = core_options_.GetScanSnapshotId(); + if (snapshot_id) { + return snapshot_id; + } + return snapshot_reader_->GetSnapshotManager()->LatestSnapshotId(); } } // namespace paimon diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.h b/src/paimon/core/table/source/data_evolution_batch_scan.h index cfa297857..16fdcccd8 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.h +++ b/src/paimon/core/table/source/data_evolution_batch_scan.h @@ -21,6 +21,7 @@ #include #include #include +#include #include #include @@ -49,7 +50,13 @@ class DataEvolutionBatchScan : public AbstractTableScan { const std::map& id_to_score); private: - Result> EvalGlobalIndex() const; + struct EvaluatedGlobalIndex { + std::shared_ptr result; + int64_t snapshot_id; + }; + + Result> EvalGlobalIndex() const; + Result> ResolveGlobalIndexSnapshotId() const; private: std::shared_ptr pool_; diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 84ab22df8..0f28da192 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -1489,6 +1489,53 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScan) { } } +TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexMissPreservesResolvedSnapshot) { + CreateTable(); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + auto schema = arrow::schema(fields_); + std::vector write_cols = schema->field_names(); + auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([ +["Alice", 10, 1, 11.1], +["Bob", 20, 0, 12.1] + ])") + .ValueOrDie(); + + ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, write_cols, src_array)); + ASSERT_OK(Commit(table_path, commit_msgs)); + ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0", "bitmap", /*options=*/{}, + Range(0, 1))); + ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f1", "bitmap", /*options=*/{}, + Range(0, 1))); + + auto predicate = + PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING, + Literal(FieldType::STRING, "missing", 7)); + + ASSERT_OK_AND_ASSIGN(auto latest_plan, ScanGlobalIndexAndData(table_path, predicate)); + ASSERT_TRUE(latest_plan->Splits().empty()); + ASSERT_EQ(latest_plan->SnapshotId(), std::optional(3)); + + ASSERT_OK_AND_ASSIGN( + auto explicit_plan, + ScanGlobalIndexAndData(table_path, predicate, {{Options::SCAN_SNAPSHOT_ID, "2"}})); + ASSERT_TRUE(explicit_plan->Splits().empty()); + ASSERT_EQ(explicit_plan->SnapshotId(), std::optional(2)); + + auto empty_index_result = BitmapGlobalIndexResult::FromRanges({}); + ASSERT_OK_AND_ASSIGN(auto supplied_latest_plan, + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, /*options=*/{}, + empty_index_result)); + ASSERT_TRUE(supplied_latest_plan->Splits().empty()); + ASSERT_EQ(supplied_latest_plan->SnapshotId(), std::optional(3)); + + ASSERT_OK_AND_ASSIGN( + auto supplied_explicit_plan, + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, + {{Options::SCAN_SNAPSHOT_ID, "2"}}, empty_index_result)); + ASSERT_TRUE(supplied_explicit_plan->Splits().empty()); + ASSERT_EQ(supplied_explicit_plan->SnapshotId(), std::optional(2)); +} + TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithOnlyOnePartitionHasIndex) { CreateTable(/*partition_keys=*/{"f1"}); std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");