From 2882c782707d1c99f84530bab8b31b572c139553 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Fri, 28 Aug 2026 20:20:04 -0400 Subject: [PATCH 1/2] 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"); From cd817b9d0a330a02ded9dbf65b9c3b53262d2840 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Tue, 1 Sep 2026 05:46:58 -0400 Subject: [PATCH 2/2] fix(scan): validate global index snapshot selection --- .../source/data_evolution_batch_scan.cpp | 27 +++++------ test/inte/global_index_test.cpp | 46 +++++++++++-------- 2 files changed, 42 insertions(+), 31 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 2a461bfb6..d7683d629 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp @@ -24,6 +24,7 @@ #include "paimon/core/global_index/global_index_scan_impl.h" #include "paimon/core/global_index/indexed_split_impl.h" +#include "paimon/core/snapshot.h" #include "paimon/core/table/source/data_split_impl.h" #include "paimon/global_index/bitmap_global_index_result.h" #include "paimon/global_index/global_index_scan.h" @@ -42,35 +43,36 @@ DataEvolutionBatchScan::DataEvolutionBatchScan( executor_(executor) {} 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::optional evaluated_index, EvalGlobalIndex()); - if (evaluated_index && evaluated_index->result) { + if (evaluated_index) { 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()); } - if (!row_ranges) { + if (!final_global_index_result) { return batch_scan_->CreatePlan(); } - if (row_ranges.value().empty()) { + if (core_options_.GetScanTagName() || core_options_.GetScanTimestampMillis()) { + return Status::NotImplemented("Global index scan does not support time travel"); + } + PAIMON_ASSIGN_OR_RAISE(std::vector 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(); + 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(global_index_snapshot_id, std::vector>()); } - 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 data_plan, batch_scan_->CreatePlan()); std::map id_to_score; @@ -154,7 +156,6 @@ DataEvolutionBatchScan::EvalGlobalIndex() const { 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"); diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index be94fb79d..2a1621e64 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -15,6 +15,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + +#include + #include "arrow/type.h" #include "gtest/gtest.h" #include "paimon/common/factories/io_hook.h" @@ -1489,7 +1492,7 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScan) { } } -TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexMissPreservesResolvedSnapshot) { +TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexSnapshotSelection) { CreateTable(); std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); auto schema = arrow::schema(fields_); @@ -1504,8 +1507,6 @@ TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexMissPreservesResolvedSnapsho 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, @@ -1513,27 +1514,36 @@ TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexMissPreservesResolvedSnapsho 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)); + ASSERT_EQ(latest_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)); + {{Options::SCAN_SNAPSHOT_ID, "1"}}, empty_index_result)); ASSERT_TRUE(supplied_explicit_plan->Splits().empty()); - ASSERT_EQ(supplied_explicit_plan->SnapshotId(), std::optional(2)); + ASSERT_EQ(supplied_explicit_plan->SnapshotId(), std::optional(1)); + + std::vector> time_travel_options = { + {{Options::SCAN_TAG_NAME, "tag"}}, + {{Options::SCAN_TIMESTAMP_MILLIS, std::to_string(std::numeric_limits::max())}}}; + for (const auto& options : time_travel_options) { + Result> result = + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, options, empty_index_result); + ASSERT_TRUE(result.status().IsNotImplemented()) << result.status().ToString(); + } + + auto unindexed_predicate = PredicateBuilder::Equal(/*field_index=*/3, /*field_name=*/"f3", + FieldType::DOUBLE, Literal(99.9)); + ASSERT_OK_AND_ASSIGN(auto fallback_plan, ScanGlobalIndexAndData(table_path, unindexed_predicate, + time_travel_options.back())); + ASSERT_EQ(fallback_plan->SnapshotId(), std::optional(2)); + + Result> nonexistent_snapshot_result = + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, + {{Options::SCAN_SNAPSHOT_ID, "999"}}, empty_index_result); + ASSERT_TRUE(nonexistent_snapshot_result.status().IsNotExist()) + << nonexistent_snapshot_result.status().ToString(); } TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithOnlyOnePartitionHasIndex) {