lxy-9602 commented on code in PR #368:
URL: https://github.com/apache/paimon-cpp/pull/368#discussion_r4043018900
##########
test/inte/data_evolution_table_test.cpp:
##########
@@ -402,6 +404,130 @@ class DataEvolutionTableTest : public ::testing::Test,
return file_store_commit->Commit(commit_msgs);
}
+ void CheckGlobalIndexScoresWithFiltering(bool merge_files) const {
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
+ {{Options::READ_BATCH_SIZE, "4"}});
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ constexpr int64_t first_row_id = 100;
+ auto base_array = PrepareBulkData(
+ 12, [](int32_t i) { return fmt::format(R"({}, "a{}", "x{}")", i,
i, i); }, fields_);
+ ASSERT_OK_AND_ASSIGN(auto base_msgs,
+ WriteArray(table_path, {"f0", "f1", "f2"},
base_array));
+ SetFirstRowId(first_row_id, base_msgs);
+ ASSERT_OK(Commit(table_path, base_msgs));
+ if (merge_files) {
+ auto update_array = PrepareBulkData(
+ 12, [](int32_t i) { return fmt::format(R"("y{}")", i); },
{fields_[2]});
+ ASSERT_OK_AND_ASSIGN(auto update_msgs, WriteArray(table_path,
{"f2"}, update_array));
+ SetFirstRowId(first_row_id, update_msgs);
+ ASSERT_OK(Commit(table_path, update_msgs));
+ }
+ auto not_two = PredicateBuilder::NotEqual(0, "f0", FieldType::INT,
Literal(2));
+ auto below_ten = PredicateBuilder::LessThan(0, "f0", FieldType::INT,
Literal(10));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
+ PredicateBuilder::And({not_two, below_ten}));
+ auto reject_all = PredicateBuilder::GreaterOrEqual(0, "f0",
FieldType::INT, Literal(100));
+ const std::vector<std::shared_ptr<Predicate>> predicates = {nullptr,
predicate, reject_all};
+ const std::vector<int64_t> candidates = {100, 101, 102, 104, 105, 106,
+ 107, 108, 109, 110, 111};
+ const std::vector<int64_t> deleted = {1, 4, 5, 6, 7, 11};
+ for (bool with_dv : {false, true}) {
+ if (with_dv) {
+ ASSERT_OK_AND_ASSIGN(std::string anchor,
PlannedAnchorFileName(table_path));
+ ASSERT_OK(CommitDeletionVectors(table_path, base_msgs[0],
{{anchor, deleted}}));
+ }
+ std::vector<float> scores;
+ for (int64_t row_id : candidates) {
+ scores.push_back(static_cast<float>(row_id) + 0.5f);
+ }
+ ScanContextBuilder scan_builder(table_path);
+
scan_builder.SetGlobalIndexResult(std::make_shared<BitmapScoredGlobalIndexResult>(
+ RoaringBitmap64::From(candidates), std::move(scores)));
+ ASSERT_OK_AND_ASSIGN(auto scan_context,
FinishScanContext(scan_builder));
+ ASSERT_OK_AND_ASSIGN(auto scan,
TableScan::Create(std::move(scan_context)));
+ ASSERT_OK_AND_ASSIGN(auto plan, scan->CreatePlan());
+ ASSERT_EQ(plan->Splits().size(), 1);
+ auto scored_split =
std::dynamic_pointer_cast<IndexedSplitImpl>(plan->Splits()[0]);
+ ASSERT_TRUE(scored_split);
+ ASSERT_EQ(scored_split->RowRanges(),
+ std::vector<Range>({Range(100, 102), Range(104, 111)}));
+ ASSERT_EQ(scored_split->Scores().size(), candidates.size());
+ auto data_split =
+
std::dynamic_pointer_cast<DataSplitImpl>(scored_split->GetDataSplit());
+ ASSERT_TRUE(data_split);
+ ASSERT_EQ(data_split->DataFiles().size(), merge_files ? 2 : 1);
+ auto unscored_split =
+ std::make_shared<IndexedSplitImpl>(data_split,
scored_split->RowRanges());
+ for (const auto& read_predicate : predicates) {
+ for (bool project_row_id : {false, true}) {
+ std::vector<std::string> read_fields = {"_INDEX_SCORE",
"f2", "f0"};
+ arrow::FieldVector expected_fields =
{SpecialFields::ValueKind().field_,
+
SpecialFields::IndexScore().field_,
+ fields_[2],
fields_[0]};
+ if (project_row_id) {
+ read_fields.insert(read_fields.begin(), "_ROW_ID");
+ expected_fields.insert(expected_fields.begin() + 1,
+ SpecialFields::RowId().field_);
+ }
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate>
bound_predicate,
+
PredicateUtils::CreatePickedFieldFilter(
+ read_predicate, {{"f0",
project_row_id ? 3 : 2}}));
+ ReadContextBuilder read_builder(table_path);
+ read_builder.SetReadFieldNames(read_fields)
+ .SetPredicate(bound_predicate)
+ .EnablePredicateFilter(true);
+ ASSERT_OK_AND_ASSIGN(auto read_context,
read_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(auto read,
TableRead::Create(std::move(read_context)));
+ // Reuse the same TableRead with and without scores: the
internal row-id
+ // projection must not affect a subsequent split's output
schema.
+ for (bool with_scores : {true, false}) {
+ SCOPED_TRACE(fmt::format(
+ "merge={}, dv={}, predicate={}, row_id={},
scores={}", merge_files,
+ with_dv, read_predicate ?
read_predicate->ToString() : "none",
+ project_row_id, with_scores));
+ ASSERT_OK_AND_ASSIGN(
+ auto reader,
+ read->CreateReader(with_scores ? scored_split :
unscored_split));
+ ASSERT_OK_AND_ASSIGN(auto result,
+
ReadResultCollector::CollectResult(std::move(reader)));
+ if (read_predicate == reject_all) {
+ ASSERT_FALSE(result);
+ continue;
+ }
+ ASSERT_TRUE(result);
+ std::vector<std::string> expected_rows;
+ for (int64_t row_id : candidates) {
+ int64_t value = row_id - first_row_id;
+ if (with_dv &&
+ std::find(deleted.begin(), deleted.end(),
value) != deleted.end()) {
+ continue;
+ }
+ if (read_predicate && (value == 2 || value >= 10))
{
+ continue;
+ }
+ std::string row_id_json =
+ project_row_id ? fmt::format("{}, ", row_id) :
"";
+ std::string score_json =
+ with_scores ? fmt::format("{}", row_id + 0.5f)
: "null";
+ expected_rows.push_back(
+ fmt::format(R"([0, {}{}, "{}{}", {}])",
row_id_json, score_json,
+ merge_files ? "y" : "x", value,
value));
+ }
+ std::shared_ptr<arrow::ChunkedArray> expected;
+
ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(
+ arrow::struct_(expected_fields),
+ {fmt::format("[{}]",
fmt::join(expected_rows, ","))},
+ &expected)
+ .ok());
+ ASSERT_TRUE(expected->Equals(*result))
+ << "actual=" << result->ToString()
+ << "\nexpected=" << expected->ToString();
Review Comment:
I find the current test a bit hard to follow. Could we at least simplify the
correspondence between the inputs and the expected results, for example by
specifying it more explicitly?
##########
src/paimon/core/global_index/indexed_split_impl.h:
##########
@@ -98,7 +100,22 @@ class IndexedSplitImpl : public IndexedSplit {
}
if (!scores_.empty()) {
size_t row_count = 0;
- for (const auto& range : row_ranges_) {
+ auto sorted_ranges = row_ranges_;
+ std::sort(sorted_ranges.begin(), sorted_ranges.end());
+ std::optional<int64_t> previous_end;
+ for (const auto& range : sorted_ranges) {
+ if (range.from < 0 || range.from > range.to) {
+ return Status::Invalid("Invalid row id range in scored
indexed split.");
+ }
+ if (range.to == std::numeric_limits<int64_t>::max()) {
+ return Status::Invalid(
+ "Row id range upper bound must be less than INT64_MAX
in scored indexed "
+ "split.");
+ }
+ if (previous_end && range.from <= previous_end.value()) {
+ return Status::Invalid("Duplicate row id in scored indexed
split.");
+ }
+ previous_end = range.to;
Review Comment:
Could you clarify the purpose of this added part? I couldn’t find similar
logic in Java `IndexedSplit`.
##########
src/paimon/common/global_index/complete_index_score_batch_reader_test.cpp:
##########
@@ -150,14 +154,106 @@ TEST_F(CompleteIndexScoreBatchReaderTest,
TestReadWithNullScores) {
])")
.ValueOrDie();
- // scores is empty, indicates all null score
- auto reader = PrepareCompleteIndexScoreBatchReader(src_array,
/*scores=*/{},
- /*batch_size=*/1);
+ // Unscored splits bypass this reader. A returned row without a score is
invalid here.
+ auto reader = PrepareCompleteIndexScoreBatchReader(src_array,
/*scores=*/{}, /*batch_size=*/1);
- ASSERT_OK_AND_ASSIGN(auto result_array,
ReadResultCollector::CollectResult(std::move(reader)));
+ ASSERT_NOK_WITH_MSG(reader->NextBatch(), "Missing global index score for
row id 0");
+}
+
+TEST_F(CompleteIndexScoreBatchReaderTest, TestGlobalScoresFollowRowIds) {
+ arrow::FieldVector fields = {
+ arrow::field("_INDEX_SCORE", arrow::float32()),
+ arrow::field("f0", arrow::int32()),
+ arrow::field("_ROW_ID", arrow::int64()),
+ };
+ // Includes compacted-out candidates, a bitmap-filtered row without a
score, and
+ // deliberately non-monotonic output ids. Scores must depend only on the
row id.
+ auto src_array = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_(fields),
+ R"([[null, 12, 102], [null, 99, null], [null, 10,
100], [null, 18, 108]])")
+ .ValueOrDie();
+ for (bool remove_row_id : {false, true}) {
+ auto expected_fields = fields;
+ if (remove_row_id) {
+ expected_fields.pop_back();
+ }
+ std::string expected_json = remove_row_id ? R"([[3, 12], [1, 10], [9,
18]])"
+ : R"([[3, 12, 102], [1, 10,
100], [9, 18, 108]])";
+ std::shared_ptr<arrow::ChunkedArray> expected_array;
+ ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(
+ arrow::struct_(expected_fields), {expected_json},
&expected_array)
+ .ok());
+ for (int32_t batch_size : {1, 3, 10}) {
+ SCOPED_TRACE(fmt::format("remove_row_id={}, batch_size={}",
remove_row_id, batch_size));
+ auto inner = std::make_unique<MockFileBatchReader>(
+ src_array, src_array->type(), RoaringBitmap32::From({0, 2,
3}), batch_size);
+ inner->EnableRandomizeBatchSize(false);
+ auto reader = std::make_unique<CompleteIndexScoreBatchReader>(
+ std::move(inner),
+ std::unordered_map<int64_t, float>{
+ {100, 1.0f}, {101, 2.0f}, {102, 3.0f}, {108, 9.0f}, {109,
10.0f}},
Review Comment:
Could we move the map before `expected_json`? Otherwise, `expected_json` is
a bit hard to understand on its own.
##########
src/paimon/core/operation/data_evolution_split_read.cpp:
##########
@@ -215,11 +215,32 @@ Result<std::unique_ptr<BatchReader>>
DataEvolutionSplitRead::CreateReader(
if (auto indexed_split =
std::dynamic_pointer_cast<IndexedSplitImpl>(split)) {
PAIMON_RETURN_NOT_OK(indexed_split->Validate());
const auto& data_split = indexed_split->GetDataSplit();
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> batch_reader,
- InnerCreateReader(data_split,
indexed_split->RowRanges()));
- if (HasIndexScoreField(raw_read_schema_)) {
+ auto read_schema = raw_read_schema_;
+ bool complete_scores = HasIndexScoreField(read_schema) &&
!indexed_split->Scores().empty();
+ bool remove_row_id = false;
+ std::unordered_map<int64_t, float> scores_by_row_id;
+ if (complete_scores) {
+ scores_by_row_id.reserve(indexed_split->Scores().size());
+ size_t score_idx = 0;
+ for (const auto& range : indexed_split->RowRanges()) {
+ for (int64_t row_id = range.from; row_id <= range.to;
++row_id) {
+ scores_by_row_id.emplace(row_id,
indexed_split->Scores()[score_idx++]);
+ }
+ }
+ remove_row_id =
read_schema->GetFieldIndex(SpecialFields::RowId().Name()) < 0;
+ if (remove_row_id) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ read_schema,
read_schema->AddField(read_schema->num_fields(),
+
DataField::ConvertDataFieldToArrowField(
+
SpecialFields::RowId())));
+ }
+ }
Review Comment:
I’m wondering whether the mapping from id to score, as well as whether to
remove `row_id`, could be handled in `CompleteIndexScoreBatchReader` instead.
We may have similar needs later for the global PK index, and doing it there
could reduce the amount of split-read-specific implementation. Also, could
`CompleteIndexScoreBatchReader` be placed in `FileBatchReader`? In that case,
`SetReadSchema` could naturally take care of adding or removing the `row_id`
field as needed.
##########
src/paimon/core/global_index/indexed_split_impl.h:
##########
@@ -98,7 +100,22 @@ class IndexedSplitImpl : public IndexedSplit {
}
if (!scores_.empty()) {
size_t row_count = 0;
- for (const auto& range : row_ranges_) {
+ auto sorted_ranges = row_ranges_;
+ std::sort(sorted_ranges.begin(), sorted_ranges.end());
+ std::optional<int64_t> previous_end;
+ for (const auto& range : sorted_ranges) {
+ if (range.from < 0 || range.from > range.to) {
+ return Status::Invalid("Invalid row id range in scored
indexed split.");
+ }
+ if (range.to == std::numeric_limits<int64_t>::max()) {
+ return Status::Invalid(
+ "Row id range upper bound must be less than INT64_MAX
in scored indexed "
+ "split.");
+ }
+ if (previous_end && range.from <= previous_end.value()) {
+ return Status::Invalid("Duplicate row id in scored indexed
split.");
+ }
+ previous_end = range.to;
Review Comment:
The `rowRanges` produced by the Java generation path should already have
strictly increasing start positions and no overlaps. Given that, is sorting
here actually necessary?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
