SteNicholas commented on code in PR #191:
URL: https://github.com/apache/paimon-cpp/pull/191#discussion_r3768421168
##########
src/paimon/core/operation/commit/conflict_detection_test.cpp:
##########
@@ -572,41 +572,58 @@ TEST_F(ConflictDetectionTest,
TestBucketKeepSameCacheEviction) {
ASSERT_EQ(1U, evicted_partition_buckets.size());
}
-TEST_F(ConflictDetectionTest,
TestDeletionVectorsNotSupportedWithBucketUnawareMode) {
- ASSERT_OK_AND_ASSIGN(
- std::shared_ptr<TableSchema> table_schema,
- TableSchema::Create(/*schema_id=*/0, arrow::schema(fields_),
/*partition_keys=*/{"f1"},
- /*primary_keys=*/{}, /*options=*/{}));
- ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
- CoreOptions::FromMap({{Options::BUCKET, "0"},
-
{Options::DELETION_VECTORS_ENABLED, "true"}}));
- ConflictDetection detection(table_schema, core_options,
/*snapshot_manager=*/nullptr,
- /*manifest_list=*/nullptr,
/*manifest_file=*/nullptr,
- /*commit_scanner=*/nullptr, "test_user",
"test_table",
- /*path_factory=*/nullptr);
+TEST_F(ConflictDetectionTest, TestDeletionVectorsAllowedWithBucketUnawareMode)
{
+ // an unaware bucket table, however its bucket resolves, may carry
deletion vectors, told
+ // apart by index file name rather than by bucket. Adding files commits;
dropping one needs
+ // the pairing this class does not build and is refused
+ for (const char* bucket : {"0", "-1"}) {
+ ASSERT_OK_AND_ASSIGN(
Review Comment:
You are right — `bucket = 0` cannot be persisted: `SchemaValidation` rejects
`bucket < 1` for non-postpone tables, and Paimon Java has no `0 ->
BUCKET_UNAWARE` mapping either (`AppendOnlyFileStore#bucketMode` only maps
`-1`; the `BucketMode.UNAWARE_BUCKET = 0` constant is the bucket directory
number an unaware table writes to, not an option value). Removed the `"0"` leg
from this test. Dropping the `bucket == 0` branch from `ResolveBucketMode`
itself is left to a follow-up to keep this change focused, since the branch
predates this PR — removing it would also align the hypothetical `bucket = 0`
with Java, which resolves it to `HASH_FIXED`.
##########
src/paimon/core/operation/data_evolution_split_read.cpp:
##########
@@ -48,9 +50,84 @@
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/range_helper.h"
#include "paimon/core/core_options.h"
+#include "paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h"
#include "paimon/core/global_index/indexed_split_impl.h"
#include "paimon/core/utils/blob_view_lookup.h"
+#include "paimon/core/utils/data_evolution_utils.h"
+
namespace paimon {
+namespace {
+/// A read-only view over one file's window of a row range group's deletion
vector. The reader
+/// probes with positions local to its file; IsDeleted(p) forwards them as
IsDeleted(p + offset),
+/// the anchor-relative positions the group deletion vector is indexed by.
+///
+/// The shift rides on a full DeletionVector because that is what
DeletionVector::Factory hands
+/// to ApplyIndexAndDvReaderIfNeeded; narrowing it to Paimon Java's read-only
DeletionVectorJudger
+/// would reach every caller of the factory. The mutating and serializing
halves reject instead.
+class PositionShiftedDeletionVector : public DeletionVector {
+ public:
+ PositionShiftedDeletionVector(const std::shared_ptr<DeletionVector>&
inner, int64_t offset,
+ int64_t length)
+ : inner_(inner), offset_(offset), length_(length) {}
+
+ Result<bool> IsDeleted(int64_t position) const override {
+ if (position < 0 || position >= length_) {
+ return Status::Invalid(
+ fmt::format("PositionShiftedDeletionVector position {} out of
window [0, {})",
+ position, length_));
+ }
+ return inner_->IsDeleted(position + offset_);
+ }
+
+ /// Conservative: false only means the group deletion vector holds
deletions somewhere, not
+ /// necessarily inside this window. Over-reporting just makes the wrapping
filter a no-op.
+ bool IsEmpty() const override {
+ return inner_->IsEmpty();
+ }
+
+ /// Deleted positions inside this window. Off the read path, which only
calls IsDeleted and
+ /// IsEmpty; the row counting that does consult a vector's cardinality
builds its factory
+ /// from the split's deletion files, never from this view.
+ ///
+ /// The interface cannot report a failure, so a failed IsValid is answered
with "everything
+ /// is deleted". That is the safe direction for a row count, which
subtracts it, but it is a
+ /// guess: the assert keeps it from passing silently where asserts are on.
+ int64_t GetCardinality() const override {
+ Result<RoaringBitmap32> valid = inner_->IsValid(offset_, length_);
+ if (!valid.ok()) {
+ assert(false);
+ return length_;
+ }
+ return length_ - valid.value().Cardinality();
Review Comment:
Done — `DeletionVector::GetCardinality` now returns `Result<int64_t>`, and a
failed `IsValid` propagates instead of asserting and returning a guess. The
ripple stays small because the interface is internal: `BitmapDeletionVector`
and the test fake return their infallible values unchanged, and the two
production callers (`DeletionFileWriter::Write`,
`DataSplitImpl::RawMergedRowCount`) propagate with `PAIMON_ASSIGN_OR_RAISE`.
##########
test/inte/data_evolution_table_test.cpp:
##########
@@ -2108,6 +2353,618 @@ TEST_P(DataEvolutionTableTest, TestWithRowIds) {
}
}
+TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectors) {
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto schema = arrow::schema(fields_);
+
+ // full-row write assigns row ids 0-3, producing the anchor file of the
row range group
+ auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [2, "b", "y"],
+ [3, "c", "z"],
+ [4, "d", "w"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
+ WriteArray(table_path, schema->field_names(),
src_array));
+ SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
+ ASSERT_OK(Commit(table_path, commit_msgs0));
+
+ // partial write of f2 over the same row range: the group merges columns
from two files
+ arrow::FieldVector f2_fields = {fields_[2]};
+ auto update_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(f2_fields),
R"([
+ ["x2"],
+ ["y2"],
+ ["z2"],
+ ["w2"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {"f2"},
update_array));
+ SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs1);
+ ASSERT_OK(Commit(table_path, commit_msgs1));
+
+ auto expected_all = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x2"],
+ [2, "b", "y2"],
+ [3, "c", "z2"],
+ [4, "d", "w2"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_all));
+
+ ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
PlannedAnchorFileName(table_path));
+ ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
+ {{anchor_file_name,
/*deleted_positions=*/{1, 3}}})
+ .status());
+
+ // both files of the group must drop the same rows to keep the column
merge aligned
+ auto expected_deleted = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x2"],
+ [3, "c", "z2"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(),
expected_deleted));
+
+ auto expected_with_row_id = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({fields_[0], fields_[1], fields_[2],
SpecialFields::RowId().field_}),
+ R"([
+ [1, "a", "x2", 0],
+ [3, "c", "z2", 2]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "_ROW_ID"},
expected_with_row_id));
+
+ // a row-range selection composes with the deletion vector: rows {1, 2}
minus deleted {1}
+ auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [3, "c", "z2"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
+ /*predicate=*/nullptr, /*row_ranges=*/{Range(1,
2)}));
+}
+
+TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsAcrossReadBatches) {
+ // the 12 rows below span several read batches, so the deletion vector
empties a whole
+ // batch of every file of the group: each file reader then skips that
batch entirely and
+ // the column merge has to stay aligned on the surviving row count alone
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
{{Options::READ_BATCH_SIZE, "4"}});
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto schema = arrow::schema(fields_);
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>>
group_msgs,
+ WriteAndCommitGroup(table_path, /*first_row_id=*/0,
+ /*f0_values=*/{0, 1, 2, 3, 4, 5,
6, 7, 8, 9, 10, 11}));
+
+ // positions 4-7 cover a whole read batch, position 9 only part of the
next one
+ ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
PlannedAnchorFileName(table_path));
+ ASSERT_OK(CommitDeletionVectors(table_path, group_msgs[0],
+ {{anchor_file_name,
/*deleted_positions=*/{4, 5, 6, 7, 9}}})
+ .status());
+
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [0, "a0", "y0"],
+ [1, "a1", "y1"],
+ [2, "a2", "y2"],
+ [3, "a3", "y3"],
+ [8, "a8", "y8"],
+ [10, "a10", "y10"],
+ [11, "a11", "y11"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+
+ arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
+ auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
+ [0], [1], [2], [3], [8], [10], [11]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));
+
+ // a row-range selection that spans the fully deleted batch composes with
it
+ auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [3, "a3", "y3"],
+ [8, "a8", "y8"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
+ /*predicate=*/nullptr, /*row_ranges=*/{Range(3,
8)}));
+}
+
+TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsOnPartOfRowRangeGroups) {
+ // one split per row range group, so a group's deletion file must not
reach the other
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
+ {{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto schema = arrow::schema(fields_);
+
+ auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [2, "b", "y"],
+ [3, "c", "z"],
+ [4, "d", "w"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs0,
+ WriteArray(table_path, schema->field_names(),
src_array0));
+ SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs0);
+ ASSERT_OK(Commit(table_path, commit_msgs0));
+
+ auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [5, "e", "v"],
+ [6, "f", "u"],
+ [7, "g", "t"],
+ [8, "h", "s"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
+ WriteArray(table_path, schema->field_names(),
src_array1));
+ SetFirstRowId(/*reset_first_row_id=*/4, commit_msgs1);
+ ASSERT_OK(Commit(table_path, commit_msgs1));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+ PlannedAnchorFileNames(table_path));
+ ASSERT_EQ(anchor_file_names.size(), 2);
+ ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs0[0],
+ {{anchor_file_names[0],
/*deleted_positions=*/{1, 3}}})
+ .status());
+
+ // the deletion vector applies to its own group only, the other group
keeps every row
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [3, "c", "z"],
+ [5, "e", "v"],
+ [6, "f", "u"],
+ [7, "g", "t"],
+ [8, "h", "s"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+
+ // a projection-only read drops the same rows
+ arrow::FieldVector f0_fields = {fields_[0]};
+ auto expected_f0 = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(f0_fields),
R"([
+ [1], [3], [5], [6], [7], [8]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f0"}, expected_f0));
+
+ arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
+ auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
+ [0], [2], [4], [5], [6], [7]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));
+}
+
+TEST_P(DataEvolutionTableTest,
TestReadWithDeletionVectorsOnEveryRowRangeGroup) {
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto schema = arrow::schema(fields_);
+
+ // one split (the default target size keeps both groups together), so a
single split
+ // deletion vector factory serves two groups anchored at different row ids
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
+ WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0,
1, 2, 3}));
+ ASSERT_OK(
+ WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4,
5, 6, 7}).status());
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+ PlannedAnchorFileNames(table_path));
+ ASSERT_EQ(anchor_file_names.size(), 2);
+
+ // Positions are anchor-relative, so the groups deliberately delete
different ones: group 0
+ // drops {1, 3} of row ids 0-3, group 1 drops {0, 2} of row ids 4-7.
Reading a group with the
+ // other group's vector, or with its anchor range as the shift base,
cannot match below.
+ ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
+ {{anchor_file_names[0],
/*deleted_positions=*/{1, 3}},
+ {anchor_file_names[1],
/*deleted_positions=*/{0, 2}}})
+ .status());
+
+ // The read looks a group's deletion vector up by its anchor file name, so
the scan has to
+ // hand it exactly that. Asserting it separates a scan-side mix-up from a
read-side one.
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> planned_splits,
+ PlanSplits(table_path));
+ ASSERT_EQ(planned_splits.size(), 1);
+ auto planned_split_impl =
std::dynamic_pointer_cast<DataSplitImpl>(planned_splits[0]);
+ ASSERT_TRUE(planned_split_impl);
+
+ ASSERT_OK_AND_ASSIGN(DeletionCardinalityMap cardinality_by_file,
+ DeletionCardinalityByDataFile(planned_splits[0]));
+ DeletionCardinalityMap expected_cardinalities = {{anchor_file_names[0], 2},
+ {anchor_file_names[1],
2}};
+ ASSERT_EQ(cardinality_by_file, expected_cardinalities);
+
+ // the same split reports the surviving row count the limit push down
prunes on: the two
+ // groups hold 4 rows each and each deletion vector drops 2 of them
+ ASSERT_OK_AND_ASSIGN(std::optional<int64_t> merged_row_count,
+ planned_split_impl->MergedRowCount());
+ ASSERT_EQ(std::optional<int64_t>(4), merged_row_count);
+
+ // a count query answers from that metadata alone, never reading a row, so
the deletion
+ // vectors have to reach it too: without them it reports the 8 rows the
files hold
+ ReadContextBuilder count_context_builder(table_path);
+ count_context_builder.SetReadFieldNames(schema->field_names());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> count_context,
+ count_context_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> count_table_read,
+ TableRead::Create(std::move(count_context)));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CountReader> count_reader,
+ count_table_read->CreateCountReader(planned_splits));
+ ASSERT_OK_AND_ASSIGN(int64_t counted_rows, count_reader->CountRows());
+ ASSERT_EQ(counted_rows, 4);
+
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [0, "a0", "y0"],
+ [2, "a2", "y2"],
+ [5, "a1", "y1"],
+ [7, "a3", "y3"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+
+ arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
+ auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
+ [0], [2], [5], [7]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));
+
+ // a row-range selection straddling the group boundary composes with both
deletion vectors:
+ // row ids {2, 3, 4, 5} minus the deleted {3, 4}
+ auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [2, "a2", "y2"],
+ [5, "a1", "y1"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
+ /*predicate=*/nullptr, /*row_ranges=*/{Range(2,
5)}));
+}
+
+TEST_P(DataEvolutionTableTest, TestReadWithFullyDeletedRowRangeGroup) {
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto schema = arrow::schema(fields_);
+
+ // one split (the default target size keeps both groups together), so the
emptied group's
+ // readers are concatenated with the surviving group's
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
+ WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0,
1, 2, 3}));
+ ASSERT_OK(
+ WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4,
5, 6, 7}).status());
+
+ // both file readers of the first group then yield nothing, and its column
merge must
+ // produce no rows at all instead of misaligning
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+ PlannedAnchorFileNames(table_path));
+ ASSERT_EQ(anchor_file_names.size(), 2);
+ ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
+ {{anchor_file_names[0],
/*deleted_positions=*/{0, 1, 2, 3}}})
+ .status());
+
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [4, "a0", "y0"],
+ [5, "a1", "y1"],
+ [6, "a2", "y2"],
+ [7, "a3", "y3"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+
+ arrow::FieldVector row_id_fields = {SpecialFields::RowId().field_};
+ auto expected_row_ids = std::dynamic_pointer_cast<arrow::StructArray>(
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(row_id_fields), R"([
+ [4], [5], [6], [7]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"_ROW_ID"}, expected_row_ids));
+
+ // a row-range selection spanning both groups keeps only what survives in
the second one
+ auto expected_selected = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [4, "a0", "y0"],
+ [5, "a1", "y1"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_selected,
+ /*predicate=*/nullptr, /*row_ranges=*/{Range(0,
5)}));
+
+ // a selection covering only deleted rows returns nothing. The plan is
asserted non-empty
+ // too: the scan cannot prune the split on row ids alone, so the emptiness
comes from the
+ // deletion vector rather than from a plan with nothing to read.
+ ASSERT_OK_AND_ASSIGN(LimitScanResult only_deleted,
+ ScanAndReadWithLimit(table_path,
schema->field_names(), /*limit=*/100,
+ /*predicate=*/nullptr,
+ /*row_ranges=*/{Range(0, 3)}));
+ ASSERT_FALSE(only_deleted.splits.empty());
+ ASSERT_FALSE(only_deleted.rows);
+}
+
+TEST_P(DataEvolutionTableTest, TestReadAfterUpdatingDeletionVectors) {
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+ auto schema = arrow::schema(fields_);
+
+ auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [2, "b", "y"],
+ [3, "c", "z"],
+ [4, "d", "w"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+ WriteArray(table_path, schema->field_names(),
src_array));
+ SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
+ ASSERT_OK(Commit(table_path, commit_msgs));
+
+ ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
PlannedAnchorFileName(table_path));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitMessage> first_dv_msg,
+ CommitDeletionVectors(table_path, commit_msgs[0],
+ {{anchor_file_name,
/*deleted_positions=*/{1}}}));
+
+ auto expected_after_first = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [3, "c", "z"],
+ [4, "d", "w"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(),
expected_after_first));
+
+ // a second deletion vector replaces the first one instead of both staying
live
+ ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs[0],
+ {{anchor_file_name,
/*deleted_positions=*/{1, 3}}},
+ /*replaced_commit_msg=*/first_dv_msg)
+ .status());
+
+ auto expected_after_update = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [3, "c", "z"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, schema->field_names(),
expected_after_update));
+}
+
+TEST_P(DataEvolutionTableTest, TestReadWithDeletionVectorsAfterAddingColumn) {
+ if (FileFormat() == "avro") {
+ GTEST_SKIP() << "Avro has no stats, which the added column's scan
pruning relies on";
+ }
+ std::map<std::string, std::string> options =
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+ auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+ [1, "a", "x"],
+ [2, "b", "y"],
+ [3, "c", "z"],
+ [4, "d", "w"]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+ WriteArray(table_path,
arrow::schema(fields_)->field_names(), src_array));
+ SetFirstRowId(/*reset_first_row_id=*/0, commit_msgs);
+ ASSERT_OK(Commit(table_path, commit_msgs));
+
+ // add column f3, then fill it for the same row range: the group merges
columns from two
+ // files written under different schema ids
+ auto f3 = arrow::field("f3", arrow::int64());
+ ASSERT_OK(TestHelper::WriteNextSchema(dir_->GetFileSystem(), table_path,
+ {DataField(0, fields_[0]),
DataField(1, fields_[1]),
+ DataField(2, fields_[2]),
DataField(3, f3)},
+ /*highest_field_id=*/3, options));
+
+ auto f3_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({f3}), R"([
+ [10], [20], [30], [40]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(auto f3_commit_msgs, WriteArray(table_path, {"f3"},
f3_array));
+ SetFirstRowId(/*reset_first_row_id=*/0, f3_commit_msgs);
+ ASSERT_OK(Commit(table_path, f3_commit_msgs));
+
+ // the deletion vector is still anchored on the oldest normal file,
written before the
+ // column was added
+ ASSERT_OK_AND_ASSIGN(std::string anchor_file_name,
PlannedAnchorFileName(table_path));
+ ASSERT_OK(CommitDeletionVectors(table_path, commit_msgs[0],
+ {{anchor_file_name,
/*deleted_positions=*/{1, 3}}})
+ .status());
+
+ arrow::FieldVector evolved_fields = {fields_[0], fields_[1], fields_[2],
f3};
+ auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(evolved_fields), R"([
+ [1, "a", "x", 10],
+ [3, "c", "z", 30]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f0", "f1", "f2", "f3"},
expected_array));
+
+ // projecting only the added column keeps the same surviving rows
+ auto expected_f3 = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({f3}), R"([
+ [10], [30]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK(ScanAndRead(table_path, {"f3"}, expected_f3));
+}
+
+TEST_P(DataEvolutionTableTest,
TestLimitPushDownWithHeavilyDeletedFirstRowRangeGroup) {
+ // one split per row range group, so the limit has to span both to be
satisfied
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true,
+ {{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
+ WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0,
1, 2, 3}));
+ ASSERT_OK(
+ WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4,
5, 6, 7}).status());
+
+ // the first group keeps a single surviving row
+ ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+ PlannedAnchorFileNames(table_path));
+ ASSERT_EQ(anchor_file_names.size(), 2);
+ ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
+ {{anchor_file_names[0],
/*deleted_positions=*/{0, 1, 2}}})
+ .status());
+
+ // the first split alone satisfies a limit of 1: it still holds the one
row that survived
+ // the deletion vector
+ ASSERT_OK_AND_ASSIGN(LimitScanResult limit_1,
+ ScanAndReadWithLimit(table_path, {"f0"},
/*limit=*/1));
+ ASSERT_EQ(limit_1.splits.size(), 1);
+ ASSERT_OK_AND_ASSIGN(std::vector<int32_t> limit_1_values,
CollectF0Values(limit_1.rows));
+ ASSERT_EQ(limit_1_values, (std::vector<int32_t>{3}));
+
+ // the first split contributes only one surviving row, so a limit of 3
needs the second one
+ ASSERT_OK_AND_ASSIGN(LimitScanResult limit_3,
+ ScanAndReadWithLimit(table_path, {"f0"},
/*limit=*/3));
+ ASSERT_EQ(limit_3.splits.size(), 2);
+ ASSERT_OK_AND_ASSIGN(std::vector<int32_t> limit_3_values,
CollectF0Values(limit_3.rows));
+ ASSERT_EQ(limit_3_values, (std::vector<int32_t>{3, 4, 5, 6, 7}));
+}
+
+TEST_P(DataEvolutionTableTest, TestLimitPushDownDisabledByNonPartitionFilter) {
+ // one split per row range group, so the plan can drop the group holding
the matches
+ CreateDataEvolutionTable(/*deletion_vectors_enabled=*/false,
+ {{Options::SOURCE_SPLIT_TARGET_SIZE, "1"}});
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+ // First row range group: no f0 value lies in [100, 200], but the 300
keeps the group's stats
+ // range straddling the filter so the scan cannot prune it. It therefore
reaches the plan
+ // reporting four rows, and contributes none of them to the result.
+ ASSERT_OK(
+ WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0,
1, 2, 300}).status());
+ // second row range group: every f0 value matches
+ ASSERT_OK(WriteAndCommitGroup(table_path, /*first_row_id=*/4,
+ /*f0_values=*/{100, 101, 102, 103})
+ .status());
Review Comment:
Adjusted — dropped the redundant `.status()` in all 23 places this PR had
added it, across `data_evolution_table_test.cpp` and `blob_table_inte_test.cpp`
(`ASSERT_OK` goes through `GenericToStatus`, which has `Result<T>` overloads).
##########
src/paimon/core/utils/data_evolution_utils_test.cpp:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/utils/data_evolution_utils.h"
+
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+std::shared_ptr<DataFileMeta> CreateFile(const std::string& file_name,
+ int64_t max_sequence_number) {
+ return DataFileMeta::ForAppend(file_name, /*file_size=*/1, /*row_count=*/1,
+ /*row_stats=*/SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0,
max_sequence_number,
+ /*schema_id=*/0, FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt,
/*first_row_id=*/0,
+ /*write_cols=*/std::nullopt)
+ .value();
+}
+} // namespace
+
+TEST(DataEvolutionUtilsTest,
TestRetrieveAnchorFileSkipsBlobAndVectorStoreFiles) {
+ auto blob_file = CreateFile("blob-file.blob", 1);
+ auto vector_store_file = CreateFile("vector-store.vector.parquet", 2);
+ auto oldest_normal_file = CreateFile("oldest-normal.parquet", 3);
+ auto newest_normal_file = CreateFile("newest-normal.parquet", 4);
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<DataFileMeta> anchor,
+ DataEvolutionUtils::RetrieveAnchorFile(
+ {blob_file, vector_store_file, newest_normal_file,
oldest_normal_file}));
+ ASSERT_EQ(anchor, oldest_normal_file);
+
+ // the same rule a caller applies to tell whether a group can be anchored
at all
+
ASSERT_TRUE(DataEvolutionUtils::IsNormalFile(oldest_normal_file->file_name));
+ ASSERT_FALSE(DataEvolutionUtils::IsNormalFile(blob_file->file_name));
+
ASSERT_FALSE(DataEvolutionUtils::IsNormalFile(vector_store_file->file_name));
+}
+
+TEST(DataEvolutionUtilsTest, TestRetrieveAnchorFileFailsWithoutNormalFile) {
+ auto blob_file_1 = CreateFile("blob-file-1.blob", 1);
+ auto blob_file_2 = CreateFile("blob-file-2.blob", 2);
+ auto vector_store_file = CreateFile("vector-store.vector.parquet", 3);
+
+ ASSERT_NOK(
Review Comment:
Done — both `ASSERT_NOK`s here use `ASSERT_NOK_WITH_MSG` now. The pinned
message is byte-identical to the `checkState` message in Java's
`DataEvolutionUtils#retrieveAnchorFile`, so the assertion also guards the port
staying aligned.
--
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]
