This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 6389e4a3 fix(scan): preserve snapshot id for empty data evolution
index plans (#258)
6389e4a3 is described below
commit 6389e4a359f1776f7e0202987b6d848f25157ecf
Author: wangyong9999 <[email protected]>
AuthorDate: Fri Sep 4 08:47:19 2026 +0800
fix(scan): preserve snapshot id for empty data evolution index plans (#258)
---
.../table/source/data_evolution_batch_scan.cpp | 81 +++++++++++++-----
.../core/table/source/data_evolution_batch_scan.h | 8 +-
test/inte/global_index_test.cpp | 96 ++++++++++++++++++++++
3 files changed, 164 insertions(+), 21 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 2f86769a..8bbc8986 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<SnapshotReader>&
snapshot_reader,
std::unique_ptr<DataTableBatchScan>&& batch_scan,
@@ -44,27 +59,33 @@ DataEvolutionBatchScan::DataEvolutionBatchScan(
executor_(executor) {}
Result<std::shared_ptr<Plan>> DataEvolutionBatchScan::CreatePlan() {
- std::optional<std::vector<Range>> row_ranges;
+ std::optional<int64_t> global_index_snapshot_id;
std::shared_ptr<GlobalIndexResult> final_global_index_result =
global_index_result_;
if (!final_global_index_result) {
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult>
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<EvaluatedGlobalIndex>
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<Range> row_ranges,
final_global_index_result->ToRanges());
+ if (row_ranges.empty()) {
+ if (!global_index_snapshot_id) {
+ return PlanImpl::EmptyPlan();
+ }
+ return std::make_shared<PlanImpl>(global_index_snapshot_id,
+
std::vector<std::shared_ptr<Split>>());
}
- 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<Plan> 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<int64_t, float> id_to_score;
if (auto scored_result =
std::dynamic_pointer_cast<ScoredGlobalIndexResult>(final_global_index_result)) {
@@ -136,35 +157,55 @@ 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>();
+ }
+ if (UsesUnsupportedTimeTravel(core_options_)) {
+ return Status::NotImplemented("Global index scan does not support time
travel");
}
auto partition_filter = batch_scan_->GetPartitionPredicate();
- // TODO(lisizhuo.lsz): support time travel
- std::optional<Snapshot> snapshot;
const std::shared_ptr<SnapshotManager>& snapshot_manager =
snapshot_reader_->GetSnapshotManager();
- if (const std::optional<int64_t>& snapshot_id =
core_options_.GetScanSnapshotId()) {
+ const StartupMode startup_mode = core_options_.GetStartupMode();
+ std::optional<Snapshot> snapshot;
+ if (startup_mode == StartupMode::FromSnapshot() ||
+ startup_mode == StartupMode::FromSnapshotFull()) {
+ const std::optional<int64_t>& 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<EvaluatedGlobalIndex>();
}
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<GlobalIndexScanImpl> 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<GlobalIndexResult> result,
index_scan->Scan(predicate));
+ return std::optional<EvaluatedGlobalIndex>(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 <cstdint>
#include <map>
#include <memory>
+#include <optional>
#include <utility>
#include <vector>
@@ -51,7 +52,12 @@ class DataEvolutionBatchScan : public AbstractTableScan {
const std::map<int64_t, float>& id_to_score);
private:
- Result<std::shared_ptr<GlobalIndexResult>> EvalGlobalIndex() const;
+ struct EvaluatedGlobalIndex {
+ std::shared_ptr<GlobalIndexResult> result;
+ int64_t snapshot_id;
+ };
+
+ Result<std::optional<EvaluatedGlobalIndex>> EvalGlobalIndex() const;
private:
std::shared_ptr<MemoryPool> pool_;
diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp
index 2e410513..517d55ac 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 <limits>
+
#include "arrow/type.h"
#include "gtest/gtest.h"
#include "paimon/common/factories/io_hook.h"
@@ -1504,6 +1507,99 @@ 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<std::string> 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<int64_t>(2));
+
+ const std::map<std::string, std::string> 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<int64_t>(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_TRUE(supplied_latest_plan->Splits().empty());
+ ASSERT_FALSE(supplied_latest_plan->SnapshotId());
+
+ 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_FALSE(supplied_explicit_plan->SnapshotId());
+
+ ASSERT_NOK_WITH_MSG(
+ ScanGlobalIndexAndData(table_path, predicate, {{Options::SCAN_MODE,
"from-snapshot"}}),
+ "scan.snapshot-id or scan.tag-name must be set when startup mode is
FROM_SNAPSHOT");
+
+ std::vector<std::map<std::string, std::string>> time_travel_options = {
+ {{Options::SCAN_TAG_NAME, "tag"}},
+ {{Options::SCAN_TIMESTAMP_MILLIS,
std::to_string(std::numeric_limits<int64_t>::max())}}};
+ for (const auto& options : time_travel_options) {
+ Result<std::shared_ptr<Plan>> result =
+ ScanGlobalIndexAndData(table_path, predicate, options);
+ ASSERT_TRUE(result.status().IsNotImplemented()) <<
result.status().ToString();
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Plan> supplied_plan,
+ ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, options,
empty_index_result));
+ ASSERT_TRUE(supplied_plan->Splits().empty());
+ ASSERT_FALSE(supplied_plan->SnapshotId());
+ }
+
+ auto unindexed_predicate = PredicateBuilder::Equal(/*field_index=*/3,
/*field_name=*/"f3",
+ FieldType::DOUBLE,
Literal(99.9));
+ Result<std::shared_ptr<Plan>> 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<Plan> no_index_plan,
+ ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
time_travel_options.back()));
+ ASSERT_EQ(no_index_plan->SnapshotId(), std::optional<int64_t>(2));
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Plan> supplied_nonexistent_snapshot_plan,
+ ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+ {{Options::SCAN_SNAPSHOT_ID, "999"}},
empty_index_result));
+ ASSERT_TRUE(supplied_nonexistent_snapshot_plan->Splits().empty());
+ ASSERT_FALSE(supplied_nonexistent_snapshot_plan->SnapshotId());
+
+ ASSERT_NOK_WITH_MSG(
+ ScanGlobalIndexAndData(table_path, predicate,
{{Options::SCAN_SNAPSHOT_ID, "999"}}),
+ "snapshot-999' not exists");
+}
+
TEST_P(GlobalIndexTest,
TestDataEvolutionBatchScanWithOnlyOnePartitionHasIndex) {
CreateTable(/*partition_keys=*/{"f1"});
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
