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 2f86769a..1b97bfd8 100644 --- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp @@ -24,11 +24,26 @@ #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/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(); +} + +} // namespace + DataEvolutionBatchScan::DataEvolutionBatchScan( const std::string& table_path, const std::shared_ptr& snapshot_reader, std::unique_ptr&& batch_scan, @@ -44,27 +59,43 @@ DataEvolutionBatchScan::DataEvolutionBatchScan( executor_(executor) {} Result> DataEvolutionBatchScan::CreatePlan() { - std::optional> row_ranges; + 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) { - 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) { + 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(); + PAIMON_ASSIGN_OR_RAISE(std::vector row_ranges, final_global_index_result->ToRanges()); + if (row_ranges.empty()) { + if (global_index_snapshot_id) { + return std::make_shared(global_index_snapshot_id, + std::vector>()); + } + + 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.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()); + 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)) { @@ -136,35 +167,52 @@ 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 - std::optional snapshot; const std::shared_ptr& snapshot_manager = snapshot_reader_->GetSnapshotManager(); - if (const std::optional& snapshot_id = core_options_.GetScanSnapshotId()) { + 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 Status::Invalid("not found latest snapshot"); + return std::optional(); } PAIMON_ASSIGN_OR_RAISE( std::unique_ptr index_scan, GlobalIndexScanImpl::Create(table_path_, table_schema_, snapshot.value(), partition_filter, core_options_, executor_, pool_)); - return index_scan->Scan(predicate); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr result, index_scan->Scan(predicate)); + return std::optional(EvaluatedGlobalIndex{result, snapshot->Id()}); } } // 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 546ae101..5f0c5a91 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 @@ -51,7 +52,12 @@ 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; private: std::shared_ptr pool_; diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index e05dc4b2..17924f49 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" @@ -1509,6 +1512,89 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScan) { } } +TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexSnapshotSelection) { + 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))); + + 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(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_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())}}}; + 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)); + 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, + {{Options::SCAN_SNAPSHOT_ID, "999"}}, empty_index_result); + ASSERT_TRUE(nonexistent_snapshot_result.status().IsInvalid()) + << nonexistent_snapshot_result.status().ToString(); +} + TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithOnlyOnePartitionHasIndex) { CreateTable(/*partition_keys=*/{"f1"}); std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");