lxy-9602 commented on code in PR #191:
URL: https://github.com/apache/paimon-cpp/pull/191#discussion_r3755752353
##########
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:
If I understand correctly, bucket cannot be set to 0 in practice, because
table creation validation would fail. In that case, it seems
`ResolveBucketMode` might not need to support the `bucket == 0` branch.
##########
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:
Would it make sense to change the base class to return `Result` instead, so
we can avoid `assert(false)`?
##########
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:
It looks like we could replace `ASSERT_OK(Func.status())` with
`ASSERT_OK(Func())` here as well. Could you please adjust it?
--
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]
