From 2882c782707d1c99f84530bab8b31b572c139553 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Fri, 28 Aug 2026 20:20:04 -0400 Subject: [PATCH 1/4] 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/4] 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) { From f6b9cb0aea7cd69feb49414083fb81f6e944cc59 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Tue, 1 Sep 2026 22:44:42 -0400 Subject: [PATCH 3/4] fix(scan): align global index snapshot planning --- .../source/data_evolution_batch_scan.cpp | 78 ++++++++++++++----- .../static_from_snapshot_starting_scanner.h | 38 +++++---- test/inte/global_index_test.cpp | 33 ++++++-- 3 files changed, 108 insertions(+), 41 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 91ee97b19..9e0ea93b3 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp @@ -26,10 +26,55 @@ #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/core/table/source/snapshot/static_from_snapshot_starting_scanner.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/global_index/bitmap_global_index_result.h" namespace paimon { +namespace { + +bool UsesUnsupportedTimeTravel(const CoreOptions& core_options) { + const StartupMode startup_mode = core_options.GetStartupMode(); + if (startup_mode == StartupMode::FromTimestamp()) { + return core_options.GetScanTimestampMillis().has_value(); + } + return startup_mode == StartupMode::FromSnapshot() && + !core_options.GetScanSnapshotId().has_value() && + core_options.GetScanTagName().has_value(); +} + +Result> ResolveGlobalIndexScanSnapshot( + const CoreOptions& core_options, const std::shared_ptr& snapshot_manager) { + const StartupMode startup_mode = core_options.GetStartupMode(); + if (startup_mode == StartupMode::FromSnapshot() || + startup_mode == StartupMode::FromSnapshotFull()) { + if (const std::optional& snapshot_id = core_options.GetScanSnapshotId()) { + return StaticFromSnapshotStartingScanner::ResolveSnapshot(snapshot_manager, + snapshot_id.value()); + } + if (startup_mode == StartupMode::FromSnapshotFull()) { + return Status::Invalid( + "scan.snapshot-id must be set when startup mode is FROM_SNAPSHOT_FULL"); + } + if (!core_options.GetScanTagName()) { + return Status::Invalid( + "scan.snapshot-id or scan.tag-name must be set when startup mode is " + "FROM_SNAPSHOT"); + } + } else if (startup_mode == StartupMode::FromTimestamp() && + !core_options.GetScanTimestampMillis()) { + return Status::Invalid( + "scan.timestamp-millis or scan.timestamp must be set when startup mode is " + "FROM_TIMESTAMP"); + } + + // Tag and timestamp scans are rejected only when the predicate uses a Global Index. Use the + // latest snapshot here to check whether an applicable index exists. + return snapshot_manager->LatestSnapshot(); +} + +} // namespace + DataEvolutionBatchScan::DataEvolutionBatchScan( const std::string& table_path, const std::shared_ptr& snapshot_reader, std::unique_ptr&& batch_scan, @@ -58,7 +103,7 @@ Result> DataEvolutionBatchScan::CreatePlan() { if (!final_global_index_result) { return batch_scan_->CreatePlan(); } - if (core_options_.GetScanTagName() || core_options_.GetScanTimestampMillis()) { + if (UsesUnsupportedTimeTravel(core_options_)) { return Status::NotImplemented("Global index scan does not support time travel"); } PAIMON_ASSIGN_OR_RAISE(std::vector row_ranges, final_global_index_result->ToRanges()); @@ -66,18 +111,12 @@ Result> DataEvolutionBatchScan::CreatePlan() { if (!global_index_snapshot_id) { const std::shared_ptr& snapshot_manager = snapshot_reader_->GetSnapshotManager(); - if (const std::optional& snapshot_id = core_options_.GetScanSnapshotId()) { - PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, - snapshot_manager->LoadSnapshot(snapshot_id.value())); - global_index_snapshot_id = snapshot.Id(); - } else { - PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, - snapshot_manager->LatestSnapshot()); - if (!snapshot) { - return PlanImpl::EmptyPlan(); - } - global_index_snapshot_id = snapshot->Id(); + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, + ResolveGlobalIndexScanSnapshot(core_options_, snapshot_manager)); + if (!snapshot) { + return PlanImpl::EmptyPlan(); } + global_index_snapshot_id = snapshot->Id(); } return std::make_shared(global_index_snapshot_id, std::vector>()); @@ -85,6 +124,9 @@ Result> DataEvolutionBatchScan::CreatePlan() { 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()); + if (global_index_snapshot_id && data_plan->SnapshotId() != global_index_snapshot_id) { + return Status::Invalid("Global index and data scan resolved different snapshots"); + } std::map id_to_score; if (auto scored_result = std::dynamic_pointer_cast(final_global_index_result)) { @@ -166,18 +208,12 @@ DataEvolutionBatchScan::EvalGlobalIndex() const { return std::optional(); } auto partition_filter = batch_scan_->GetPartitionPredicate(); - std::optional snapshot; const std::shared_ptr& snapshot_manager = snapshot_reader_->GetSnapshotManager(); - if (const std::optional& snapshot_id = core_options_.GetScanSnapshotId()) { - PAIMON_ASSIGN_OR_RAISE(Snapshot loaded_snapshot, - snapshot_manager->LoadSnapshot(snapshot_id.value())); - snapshot = std::move(loaded_snapshot); - } else { - PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot()); - } + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, + ResolveGlobalIndexScanSnapshot(core_options_, snapshot_manager)); if (!snapshot) { - return Status::Invalid("not found latest snapshot"); + return std::optional(); } PAIMON_ASSIGN_OR_RAISE( diff --git a/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h b/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h index 79ebb79a4..2ed28fb2e 100644 --- a/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h +++ b/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h @@ -20,6 +20,7 @@ #pragma once #include +#include #include "paimon/core/table/source/snapshot/starting_scanner.h" #include "paimon/logging.h" @@ -36,29 +37,36 @@ class StaticFromSnapshotStartingScanner : public StartingScanner { starting_snapshot_id_ = snapshot_id; } - Result> Scan( - const std::shared_ptr& snapshot_reader) override { + static Result> ResolveSnapshot( + const std::shared_ptr& snapshot_manager, int64_t snapshot_id) { PAIMON_ASSIGN_OR_RAISE(std::optional earliest, - snapshot_manager_->EarliestSnapshotId()); - PAIMON_ASSIGN_OR_RAISE(std::optional latest, - snapshot_manager_->LatestSnapshotId()); - if (earliest == std::nullopt || latest == std::nullopt) { - PAIMON_LOG_INFO( - logger_, "There is currently no snapshot. Waiting for snapshot generation.%s", ""); - return std::make_shared(); + snapshot_manager->EarliestSnapshotId()); + PAIMON_ASSIGN_OR_RAISE(std::optional latest, snapshot_manager->LatestSnapshotId()); + if (!earliest || !latest) { + return std::optional(); } - if (starting_snapshot_id_.value() < earliest.value() || - starting_snapshot_id_.value() > latest.value()) { + if (snapshot_id < earliest.value() || snapshot_id > latest.value()) { return Status::Invalid( fmt::format("The specified scan snapshotId {} is out of " "available snapshotId range [{}, {}].", - starting_snapshot_id_.value(), earliest.value(), latest.value())); + snapshot_id, earliest.value(), latest.value())); + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager->LoadSnapshot(snapshot_id)); + return std::optional(std::move(snapshot)); + } + + Result> Scan( + const std::shared_ptr& snapshot_reader) override { + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, + ResolveSnapshot(snapshot_manager_, starting_snapshot_id_.value())); + if (!snapshot) { + PAIMON_LOG_INFO( + logger_, "There is currently no snapshot. Waiting for snapshot generation.%s", ""); + return std::make_shared(); } - PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, - snapshot_manager_->LoadSnapshot(starting_snapshot_id_.value())); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr plan, - snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot)->Read()); + snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot.value())->Read()); return std::make_shared(plan); } diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 3a8be9837..3d4f2f7c3 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -1536,14 +1536,37 @@ TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexSnapshotSelection) { ASSERT_TRUE(latest_plan->Splits().empty()); ASSERT_EQ(latest_plan->SnapshotId(), std::optional(2)); + const std::map explicit_latest_options = { + {Options::SCAN_MODE, "latest"}, + {Options::SCAN_SNAPSHOT_ID, "999"}, + {Options::SCAN_TAG_NAME, "ignored"}, + {Options::SCAN_TIMESTAMP_MILLIS, "0"}}; + ASSERT_OK_AND_ASSIGN(auto explicit_latest_plan, + ScanGlobalIndexAndData(table_path, predicate, explicit_latest_options)); + ASSERT_TRUE(explicit_latest_plan->Splits().empty()); + ASSERT_EQ(explicit_latest_plan->SnapshotId(), std::optional(2)); + auto empty_index_result = BitmapGlobalIndexResult::FromRanges({}); - ASSERT_OK_AND_ASSIGN( - auto supplied_explicit_plan, - ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, - {{Options::SCAN_SNAPSHOT_ID, "1"}}, empty_index_result)); + ASSERT_OK_AND_ASSIGN(auto supplied_latest_plan, + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, + explicit_latest_options, empty_index_result)); + ASSERT_EQ(supplied_latest_plan->SnapshotId(), std::optional(2)); + + ASSERT_OK_AND_ASSIGN(auto supplied_explicit_plan, + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, + {{Options::SCAN_SNAPSHOT_ID, "1"}, + {Options::SCAN_TAG_NAME, "ignored"}, + {Options::SCAN_TIMESTAMP_MILLIS, "0"}}, + empty_index_result)); ASSERT_TRUE(supplied_explicit_plan->Splits().empty()); ASSERT_EQ(supplied_explicit_plan->SnapshotId(), std::optional(1)); + Result> missing_selector_result = + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, + {{Options::SCAN_MODE, "from-snapshot"}}, empty_index_result); + ASSERT_TRUE(missing_selector_result.status().IsInvalid()) + << missing_selector_result.status().ToString(); + std::vector> time_travel_options = { {{Options::SCAN_TAG_NAME, "tag"}}, {{Options::SCAN_TIMESTAMP_MILLIS, std::to_string(std::numeric_limits::max())}}}; @@ -1562,7 +1585,7 @@ TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexSnapshotSelection) { Result> nonexistent_snapshot_result = ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, {{Options::SCAN_SNAPSHOT_ID, "999"}}, empty_index_result); - ASSERT_TRUE(nonexistent_snapshot_result.status().IsNotExist()) + ASSERT_TRUE(nonexistent_snapshot_result.status().IsInvalid()) << nonexistent_snapshot_result.status().ToString(); } From 740033073b8c125de7b54bf6208f84e4b81af850 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Wed, 2 Sep 2026 04:56:21 -0400 Subject: [PATCH 4/4] fix(scan): avoid inferring external index snapshots --- .../source/data_evolution_batch_scan.cpp | 83 +++++++++---------- .../static_from_snapshot_starting_scanner.h | 38 ++++----- test/inte/global_index_test.cpp | 12 ++- 3 files changed, 61 insertions(+), 72 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 9e0ea93b3..1b97bfd8e 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp @@ -26,7 +26,6 @@ #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/core/table/source/snapshot/static_from_snapshot_starting_scanner.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/global_index/bitmap_global_index_result.h" @@ -43,36 +42,6 @@ bool UsesUnsupportedTimeTravel(const CoreOptions& core_options) { core_options.GetScanTagName().has_value(); } -Result> ResolveGlobalIndexScanSnapshot( - const CoreOptions& core_options, const std::shared_ptr& snapshot_manager) { - const StartupMode startup_mode = core_options.GetStartupMode(); - if (startup_mode == StartupMode::FromSnapshot() || - startup_mode == StartupMode::FromSnapshotFull()) { - if (const std::optional& snapshot_id = core_options.GetScanSnapshotId()) { - return StaticFromSnapshotStartingScanner::ResolveSnapshot(snapshot_manager, - snapshot_id.value()); - } - if (startup_mode == StartupMode::FromSnapshotFull()) { - return Status::Invalid( - "scan.snapshot-id must be set when startup mode is FROM_SNAPSHOT_FULL"); - } - if (!core_options.GetScanTagName()) { - return Status::Invalid( - "scan.snapshot-id or scan.tag-name must be set when startup mode is " - "FROM_SNAPSHOT"); - } - } else if (startup_mode == StartupMode::FromTimestamp() && - !core_options.GetScanTimestampMillis()) { - return Status::Invalid( - "scan.timestamp-millis or scan.timestamp must be set when startup mode is " - "FROM_TIMESTAMP"); - } - - // Tag and timestamp scans are rejected only when the predicate uses a Global Index. Use the - // latest snapshot here to check whether an applicable index exists. - return snapshot_manager->LatestSnapshot(); -} - } // namespace DataEvolutionBatchScan::DataEvolutionBatchScan( @@ -90,6 +59,13 @@ DataEvolutionBatchScan::DataEvolutionBatchScan( executor_(executor) {} Result> DataEvolutionBatchScan::CreatePlan() { + const bool may_use_global_index = + global_index_result_ || + (core_options_.GlobalIndexEnabled() && batch_scan_->GetNonPartitionPredicate()); + if (may_use_global_index && UsesUnsupportedTimeTravel(core_options_)) { + return Status::NotImplemented("Global index scan does not support time travel"); + } + std::optional global_index_snapshot_id; std::shared_ptr final_global_index_result = global_index_result_; if (!final_global_index_result) { @@ -103,22 +79,15 @@ Result> DataEvolutionBatchScan::CreatePlan() { if (!final_global_index_result) { return batch_scan_->CreatePlan(); } - if (UsesUnsupportedTimeTravel(core_options_)) { - 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) { - const std::shared_ptr& snapshot_manager = - snapshot_reader_->GetSnapshotManager(); - PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, - ResolveGlobalIndexScanSnapshot(core_options_, snapshot_manager)); - if (!snapshot) { - return PlanImpl::EmptyPlan(); - } - global_index_snapshot_id = snapshot->Id(); + if (global_index_snapshot_id) { + return std::make_shared(global_index_snapshot_id, + std::vector>()); } - return std::make_shared(global_index_snapshot_id, + + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_plan, batch_scan_->CreatePlan()); + return std::make_shared(data_plan->SnapshotId(), std::vector>()); } PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index, RowRangeIndex::Create(row_ranges)); @@ -210,8 +179,30 @@ DataEvolutionBatchScan::EvalGlobalIndex() const { auto partition_filter = batch_scan_->GetPartitionPredicate(); const std::shared_ptr& snapshot_manager = snapshot_reader_->GetSnapshotManager(); - PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, - ResolveGlobalIndexScanSnapshot(core_options_, snapshot_manager)); + const StartupMode startup_mode = core_options_.GetStartupMode(); + std::optional snapshot; + if (startup_mode == StartupMode::FromSnapshot() || + startup_mode == StartupMode::FromSnapshotFull()) { + const std::optional& snapshot_id = core_options_.GetScanSnapshotId(); + if (!snapshot_id) { + if (startup_mode == StartupMode::FromSnapshotFull()) { + return Status::Invalid( + "scan.snapshot-id must be set when startup mode is FROM_SNAPSHOT_FULL"); + } + return Status::Invalid( + "scan.snapshot-id or scan.tag-name must be set when startup mode is " + "FROM_SNAPSHOT"); + } + PAIMON_ASSIGN_OR_RAISE(Snapshot loaded_snapshot, + snapshot_manager->LoadSnapshot(snapshot_id.value())); + snapshot = std::move(loaded_snapshot); + } else if (startup_mode == StartupMode::FromTimestamp()) { + return Status::Invalid( + "scan.timestamp-millis or scan.timestamp must be set when startup mode is " + "FROM_TIMESTAMP"); + } else { + PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot()); + } if (!snapshot) { return std::optional(); } diff --git a/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h b/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h index 2ed28fb2e..79ebb79a4 100644 --- a/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h +++ b/src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h @@ -20,7 +20,6 @@ #pragma once #include -#include #include "paimon/core/table/source/snapshot/starting_scanner.h" #include "paimon/logging.h" @@ -37,36 +36,29 @@ class StaticFromSnapshotStartingScanner : public StartingScanner { starting_snapshot_id_ = snapshot_id; } - static Result> ResolveSnapshot( - const std::shared_ptr& snapshot_manager, int64_t snapshot_id) { - PAIMON_ASSIGN_OR_RAISE(std::optional earliest, - snapshot_manager->EarliestSnapshotId()); - PAIMON_ASSIGN_OR_RAISE(std::optional latest, snapshot_manager->LatestSnapshotId()); - if (!earliest || !latest) { - return std::optional(); - } - if (snapshot_id < earliest.value() || snapshot_id > latest.value()) { - return Status::Invalid( - fmt::format("The specified scan snapshotId {} is out of " - "available snapshotId range [{}, {}].", - snapshot_id, earliest.value(), latest.value())); - } - PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager->LoadSnapshot(snapshot_id)); - return std::optional(std::move(snapshot)); - } - Result> Scan( const std::shared_ptr& snapshot_reader) override { - PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, - ResolveSnapshot(snapshot_manager_, starting_snapshot_id_.value())); - if (!snapshot) { + PAIMON_ASSIGN_OR_RAISE(std::optional earliest, + snapshot_manager_->EarliestSnapshotId()); + PAIMON_ASSIGN_OR_RAISE(std::optional latest, + snapshot_manager_->LatestSnapshotId()); + if (earliest == std::nullopt || latest == std::nullopt) { PAIMON_LOG_INFO( logger_, "There is currently no snapshot. Waiting for snapshot generation.%s", ""); return std::make_shared(); } + if (starting_snapshot_id_.value() < earliest.value() || + starting_snapshot_id_.value() > latest.value()) { + return Status::Invalid( + fmt::format("The specified scan snapshotId {} is out of " + "available snapshotId range [{}, {}].", + starting_snapshot_id_.value(), earliest.value(), latest.value())); + } + PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, + snapshot_manager_->LoadSnapshot(starting_snapshot_id_.value())); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr plan, - snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot.value())->Read()); + snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot)->Read()); return std::make_shared(plan); } diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 3d4f2f7c3..17924f496 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -1578,9 +1578,15 @@ TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexSnapshotSelection) { 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> unindexed_time_travel_result = + ScanGlobalIndexAndData(table_path, unindexed_predicate, time_travel_options.back()); + ASSERT_TRUE(unindexed_time_travel_result.status().IsNotImplemented()) + << unindexed_time_travel_result.status().ToString(); + + ASSERT_OK_AND_ASSIGN( + std::shared_ptr no_index_plan, + ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, time_travel_options.back())); + ASSERT_EQ(no_index_plan->SnapshotId(), std::optional(2)); Result> nonexistent_snapshot_result = ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,