Skip to content

Commit 12cf1af

Browse files
lszskyelszskye
authored
test(compaction): add ut (alibaba#235)
1 parent 393b613 commit 12cf1af

22 files changed

Lines changed: 451 additions & 75 deletions

‎src/paimon/core/deletionvectors/deletion_vectors_index_file_test.cpp‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,8 @@ TEST(DeletionVectorsIndexFileTest, Basic) {
3636
FileSystemFactory::Get("local", dir->Str(), {}));
3737
auto path_factory = std::make_shared<MockIndexPathFactory>(dir->Str());
3838
auto pool = GetDefaultPool();
39-
DeletionVectorsIndexFile index_file(fs, path_factory, /*bitmap64=*/false, pool);
39+
auto index_file =
40+
std::make_shared<DeletionVectorsIndexFile>(fs, path_factory, /*bitmap64=*/false, pool);
4041

4142
std::map<std::string, std::shared_ptr<DeletionVector>> input;
4243
RoaringBitmap32 roaring_1;
@@ -50,15 +51,18 @@ TEST(DeletionVectorsIndexFileTest, Basic) {
5051
}
5152
input["dv2"] = std::make_shared<BitmapDeletionVector>(roaring_2);
5253

53-
ASSERT_FALSE(index_file.Bitmap64());
54-
ASSERT_OK_AND_ASSIGN(auto meta, index_file.WriteSingleFile(input));
54+
ASSERT_FALSE(index_file->Bitmap64());
55+
ASSERT_OK_AND_ASSIGN(auto meta, index_file->WriteSingleFile(input));
5556
ASSERT_GT(meta->FileSize(), 0);
57+
ASSERT_OK_AND_ASSIGN(auto size, index_file->FileSize(meta));
58+
ASSERT_EQ(meta->FileSize(), size);
5659
ASSERT_EQ(meta->IndexType(), DeletionVectorsIndexFile::DELETION_VECTORS_INDEX);
5760
ASSERT_EQ(meta->FileName(), "index-0");
61+
ASSERT_FALSE(index_file->IsExternalPath());
5862
ASSERT_EQ(meta->ExternalPath(), std::nullopt);
5963

6064
// Round trip: write then read all deletion vectors from index file.
61-
ASSERT_OK_AND_ASSIGN(auto read_back, index_file.ReadAllDeletionVectors(meta));
65+
ASSERT_OK_AND_ASSIGN(auto read_back, index_file->ReadAllDeletionVectors(meta));
6266
ASSERT_EQ(read_back.size(), input.size());
6367
ASSERT_EQ(read_back.at("dv1")->GetCardinality(), 10);
6468
ASSERT_EQ(read_back.at("dv2")->GetCardinality(), 10);

‎src/paimon/core/global_index/global_index_write_task.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ Result<std::unique_ptr<BatchReader>> CreateBatchReader(
7979
const std::shared_ptr<MemoryPool>& pool) {
8080
ReadContextBuilder read_context_builder(table_path);
8181
read_context_builder.SetOptions(core_options.ToMap())
82+
.WithFileSystem(core_options.GetFileSystem())
8283
.EnablePrefetch(true)
8384
.WithMemoryPool(pool)
8485
.SetReadSchema({field_name});

‎src/paimon/core/mergetree/compact/aggregate/field_listagg_agg.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ class FieldListaggAgg : public FieldAggregator {
7979
new_result.append(in_str);
8080
result_ = std::move(new_result);
8181
}
82-
return std::string_view(result_);
82+
return std::string_view{result_};
8383
}
8484

8585
private:

‎src/paimon/core/mergetree/compact/lookup_merge_tree_compact_rewriter.cpp‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -65,21 +65,23 @@ LookupMergeTreeCompactRewriter<T>::Create(
6565
auto write_schema = SpecialFields::CompleteSequenceAndValueKindField(data_schema);
6666

6767
// TODO(xinyu.lxy): set executor
68-
// TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
69-
// usage during compaction. Will fix via parquet format refactor.
7068
ReadContextBuilder read_context_builder(path_factory_cache->RootPath());
7169
read_context_builder.SetOptions(options.ToMap())
70+
.WithFileSystem(options.GetFileSystem())
7271
.EnablePrefetch(true)
7372
.SetPrefetchMaxParallelNum(1)
7473
.SetPrefetchBatchCount(3)
75-
.WithMemoryPool(pool)
76-
.AddOption("parquet.read.enable-pre-buffer", "false");
74+
.WithMemoryPool(pool);
7775
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ReadContext> read_context,
7876
read_context_builder.Finish());
79-
80-
PAIMON_ASSIGN_OR_RAISE(
81-
std::shared_ptr<InternalReadContext> internal_context,
82-
InternalReadContext::Create(read_context, table_schema, options.ToMap()));
77+
// TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
78+
// usage during compaction. Will fix via parquet format refactor.
79+
auto new_options = options.ToMap();
80+
if (new_options.find("parquet.read.enable-pre-buffer") == new_options.end()) {
81+
new_options["parquet.read.enable-pre-buffer"] = "false";
82+
}
83+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InternalReadContext> internal_context,
84+
InternalReadContext::Create(read_context, table_schema, new_options));
8385
PAIMON_ASSIGN_OR_RAISE(
8486
std::shared_ptr<FileStorePathFactory> path_factory,
8587
path_factory_cache->GetOrCreatePathFactory(options.GetFileFormat()->Identifier()));

‎src/paimon/core/mergetree/compact/merge_tree_compact_rewriter.cpp‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -71,21 +71,23 @@ Result<std::unique_ptr<MergeTreeCompactRewriter>> MergeTreeCompactRewriter::Crea
7171
auto write_schema = SpecialFields::CompleteSequenceAndValueKindField(data_schema);
7272

7373
// TODO(xinyu.lxy): set executor
74-
// TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
75-
// usage during compaction. Will fix via parquet format refactor.
7674
ReadContextBuilder read_context_builder(path_factory_cache->RootPath());
7775
read_context_builder.SetOptions(options.ToMap())
76+
.WithFileSystem(options.GetFileSystem())
7877
.EnablePrefetch(true)
7978
.SetPrefetchMaxParallelNum(1)
8079
.SetPrefetchBatchCount(3)
81-
.WithMemoryPool(pool)
82-
.AddOption("parquet.read.enable-pre-buffer", "false");
80+
.WithMemoryPool(pool);
8381
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ReadContext> read_context,
8482
read_context_builder.Finish());
85-
86-
PAIMON_ASSIGN_OR_RAISE(
87-
std::shared_ptr<InternalReadContext> internal_context,
88-
InternalReadContext::Create(read_context, table_schema, options.ToMap()));
83+
// TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
84+
// usage during compaction. Will fix via parquet format refactor.
85+
auto new_options = options.ToMap();
86+
if (new_options.find("parquet.read.enable-pre-buffer") == new_options.end()) {
87+
new_options["parquet.read.enable-pre-buffer"] = "false";
88+
}
89+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InternalReadContext> internal_context,
90+
InternalReadContext::Create(read_context, table_schema, new_options));
8991
PAIMON_ASSIGN_OR_RAISE(
9092
std::shared_ptr<FileStorePathFactory> path_factory,
9193
path_factory_cache->GetOrCreatePathFactory(options.GetFileFormat()->Identifier()));

‎src/paimon/core/mergetree/compact/merge_tree_compact_rewriter_test.cpp‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,38 @@ TEST_F(MergeTreeCompactRewriterTest, TestSimple) {
165165
CheckResult(compact_file_name, fs, table_schema, expected_array);
166166
}
167167

168+
TEST_F(MergeTreeCompactRewriterTest, TestCancel) {
169+
std::string origin_table_path = GetDataDir() + "/orc/pk_table_scan_and_read_mor.db/";
170+
auto table_dir = UniqueTestDirectory::Create("local");
171+
ASSERT_TRUE(TestUtil::CopyDirectory(origin_table_path, table_dir->Str()));
172+
std::string table_path = table_dir->Str() + "/pk_table_scan_and_read_mor";
173+
auto fs = table_dir->GetFileSystem();
174+
175+
SchemaManager schema_manager(fs, table_path);
176+
ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
177+
ASSERT_OK_AND_ASSIGN(auto options, CoreOptions::FromMap(table_schema->Options()));
178+
auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
179+
auto dv_factory = [](const std::string&) -> Result<std::shared_ptr<DeletionVector>> {
180+
return std::shared_ptr<DeletionVector>();
181+
};
182+
auto cancellation_controller = std::make_shared<CancellationController>();
183+
auto path_factory_cache =
184+
std::make_shared<FileStorePathFactoryCache>(table_path, table_schema, options, pool_);
185+
ASSERT_OK_AND_ASSIGN(
186+
auto rewriter,
187+
MergeTreeCompactRewriter::Create(
188+
/*bucket=*/1, /*partition=*/BinaryRowGenerator::GenerateRow({10}, pool_.get()),
189+
table_schema, dv_factory, path_factory_cache, options, cancellation_controller, pool_));
190+
191+
ASSERT_OK_AND_ASSIGN(auto runs, GenerateSortedRuns(table_path, table_schema, /*bucket=*/1,
192+
/*partition=*/{{"f1", "10"}}));
193+
194+
// cancel compaction here
195+
cancellation_controller->Cancel();
196+
ASSERT_NOK_WITH_MSG(rewriter->Rewrite(/*output_level=*/5, /*drop_delete=*/true, runs),
197+
"Compaction is cancelled");
198+
}
199+
168200
TEST_F(MergeTreeCompactRewriterTest, TestNotDropDelete) {
169201
std::string origin_table_path = GetDataDir() + "/orc/pk_table_scan_and_read_mor.db/";
170202
auto table_dir = UniqueTestDirectory::Create("local");

‎src/paimon/core/mergetree/compact/merge_tree_compact_task.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ Status MergeTreeCompactTask::Rewrite(std::vector<std::vector<SortedRun>>* candid
9292
if (candidate->size() == 1) {
9393
const auto& section = (*candidate)[0];
9494
if (section.empty()) {
95-
return Status::OK();
95+
return Status::Invalid("invalid section, section cannot be empty in candidate");
9696
}
9797
if (section.size() == 1) {
9898
for (const auto& file : section[0].Files()) {

‎src/paimon/core/mergetree/compact/universal_compaction_test.cpp‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,21 @@ TEST_F(UniversalCompactionTest, TestPick) {
143143
ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()), std::vector<int64_t>({1, 50, 3}));
144144
}
145145

146+
TEST_F(UniversalCompactionTest, TestAllLevelRunsInvolved) {
147+
int64_t current_time = 0;
148+
auto full_compact_trigger = std::make_shared<TestableEarlyFullCompaction>(
149+
/*full_compaction_interval=*/std::nullopt,
150+
/*total_size_threshold=*/std::nullopt,
151+
/*incremental_size_threshold=*/1000l, &current_time);
152+
UniversalCompaction compaction(/*max_size_amp=*/100, /*size_ratio=*/1,
153+
/*num_run_compaction_trigger=*/3, full_compact_trigger, nullptr);
154+
ASSERT_OK_AND_ASSIGN(auto pick, compaction.Pick(/*num_levels=*/3, CreateRunsWithLevelAndSize(
155+
/*levels=*/{0, 0, 0},
156+
/*sizes=*/{1, 1, 3})));
157+
ASSERT_TRUE(pick);
158+
ASSERT_EQ(GetFileSizeVecFromCompactUnit(pick.value()), std::vector<int64_t>({1, 1, 3}));
159+
}
160+
146161
TEST_F(UniversalCompactionTest, TestOptimizedCompactionInterval) {
147162
int64_t current_time = 0;
148163
auto full_compact_trigger = std::make_shared<TestableEarlyFullCompaction>(

‎src/paimon/core/mergetree/lookup_levels.cpp‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -55,20 +55,23 @@ Result<std::unique_ptr<LookupLevels<T>>> LookupLevels<T>::Create(
5555
auto partition_schema = DataField::ConvertDataFieldsToArrowSchema(partition_fields);
5656

5757
// TODO(xinyu.lxy): set executor
58-
// TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
59-
// usage during compaction. Will fix via parquet format refactor.
6058
ReadContextBuilder read_context_builder(path_factory->RootPath());
6159
read_context_builder.SetOptions(options.ToMap())
60+
.WithFileSystem(options.GetFileSystem())
6261
.EnablePrefetch(true)
6362
.SetPrefetchMaxParallelNum(1)
6463
.SetPrefetchBatchCount(3)
65-
.WithMemoryPool(pool)
66-
.AddOption("parquet.read.enable-pre-buffer", "false");
64+
.WithMemoryPool(pool);
6765
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ReadContext> read_context,
6866
read_context_builder.Finish());
69-
PAIMON_ASSIGN_OR_RAISE(
70-
std::shared_ptr<InternalReadContext> internal_read_context,
71-
InternalReadContext::Create(read_context, table_schema, options.ToMap()));
67+
// TODO(xinyu.lxy): temporarily disabled pre-buffer for parquet, which may cause high memory
68+
// usage during compaction. Will fix via parquet format refactor.
69+
auto new_options = options.ToMap();
70+
if (new_options.find("parquet.read.enable-pre-buffer") == new_options.end()) {
71+
new_options["parquet.read.enable-pre-buffer"] = "false";
72+
}
73+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InternalReadContext> internal_read_context,
74+
InternalReadContext::Create(read_context, table_schema, new_options));
7275
auto split_read = std::make_unique<RawFileSplitRead>(path_factory, internal_read_context, pool,
7376
CreateDefaultExecutor());
7477

0 commit comments

Comments
 (0)