Copilot commented on code in PR #104: URL: https://github.com/apache/paimon-cpp/pull/104#discussion_r3456649661
########## src/paimon/core/mergetree/levels.h: ########## @@ -0,0 +1,105 @@ +/* + * 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. + */ + + +#pragma once +#include "paimon/common/utils/fields_comparator.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/mergetree/level_sorted_run.h" +#include "paimon/core/mergetree/sorted_run.h" +#include "paimon/result.h" +namespace paimon { Review Comment: This header uses `std::set`, `std::map`, and `std::string` but doesn’t include `<set>`, `<map>`, or `<string>`. Relying on transitive includes will break compilation for TUs that include `levels.h` first. ########## src/paimon/core/mergetree/levels.cpp: ########## @@ -0,0 +1,238 @@ +/* + * 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/mergetree/levels.h" + +#include <algorithm> +namespace paimon { +bool Levels::Level0Comparator::operator()(const std::shared_ptr<DataFileMeta>& a, + const std::shared_ptr<DataFileMeta>& b) const { + if (a->max_sequence_number != b->max_sequence_number) { + // file with larger sequence number should be in front + return a->max_sequence_number > b->max_sequence_number; + } else { + // When two or more jobs are writing the same merge tree, it is + // possible that multiple files have the same maxSequenceNumber. In + // this case we have to compare their file names so that files with + // same maxSequenceNumber won't be "de-duplicated" by the tree set. + int64_t min_seq_a = a->min_sequence_number; + int64_t min_seq_b = b->min_sequence_number; + if (min_seq_a != min_seq_b) { + return min_seq_a < min_seq_b; + } + // If minSequenceNumber is also the same, use creation time + Timestamp time_a = a->creation_time; + Timestamp time_b = b->creation_time; + if (time_a != time_b) { + return time_a < time_b; + } + // Final fallback: filename (to ensure uniqueness in set) + return a->file_name < b->file_name; + } +} + +Result<std::unique_ptr<Levels>> Levels::Create( + const std::shared_ptr<FieldsComparator>& key_comparator, + const std::vector<std::shared_ptr<DataFileMeta>>& input_files, int32_t num_levels) { + // in case the num of levels is not specified explicitly + int32_t restored_num_levels = -1; + for (const auto& file : input_files) { + if (file->level > restored_num_levels) { + restored_num_levels = file->level; + } + } + restored_num_levels = std::max(restored_num_levels + 1, num_levels); + if (restored_num_levels <= 1) { + return Status::Invalid("Number of levels must be at least 2."); + } + + std::set<std::shared_ptr<DataFileMeta>, Levels::Level0Comparator> level0; + std::vector<SortedRun> levels; + levels.reserve(restored_num_levels - 1); + for (int32_t i = 1; i < restored_num_levels; ++i) { + levels.push_back(SortedRun::Empty()); + } + auto level_map = GroupByLevel(input_files); + for (auto& [level, files] : level_map) { + PAIMON_RETURN_NOT_OK( + UpdateLevel(level, /*before=*/{}, /*after=*/files, key_comparator, &levels, &level0)); + } + + size_t total_file_num = level0.size(); + for (const auto& run : levels) { + total_file_num += run.Files().size(); + } + if (total_file_num != input_files.size()) { + return Status::Invalid( + "Number of files stored in Levels does not equal to the size of inputFiles. This " + "is unexpected."); + } + return std::unique_ptr<Levels>(new Levels(key_comparator, level0, levels)); +} + +int32_t Levels::NumberOfSortedRuns() const { + int32_t number_of_runs = level0_.size(); + for (const auto& run : levels_) { + if (!run.IsEmpty()) { + number_of_runs++; + } + } + return number_of_runs; +} + +void Levels::AddDropFileCallback(DropFileCallback* callback) { + drop_file_callbacks_.push_back(callback); +} + +void Levels::RemoveDropFileCallback(DropFileCallback* callback) { + drop_file_callbacks_.erase( + std::remove(drop_file_callbacks_.begin(), drop_file_callbacks_.end(), callback), + drop_file_callbacks_.end()); +} + +Status Levels::AddLevel0File(const std::shared_ptr<DataFileMeta>& file) { + if (file->level != 0) { + return Status::Invalid("must add level0 file in AddLevel0File"); + } + level0_.insert(file); + return Status::OK(); +} + +int32_t Levels::NonEmptyHighestLevel() const { + for (int32_t i = levels_.size() - 1; i >= 0; i--) { + if (!levels_[i].IsEmpty()) { + return i + 1; + } + } + return level0_.empty() ? -1 : 0; +} Review Comment: `NonEmptyHighestLevel()` initializes the loop counter with `levels_.size() - 1` where `levels_.size()` is `size_t`. When `levels_` is empty this underflows before the cast, which is easy to miss and can trigger warnings. Prefer a `size_t` countdown loop. ########## src/paimon/core/mergetree/sorted_run_test.cpp: ########## @@ -0,0 +1,143 @@ +/* + * 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/mergetree/sorted_run.h" + +#include <optional> +#include <string> +#include <variant> + +#include "arrow/type_fwd.h" +#include "gtest/gtest.h" +#include "paimon/common/types/data_field.h" +#include "paimon/core/manifest/file_source.h" +#include "paimon/core/mergetree/level_sorted_run.h" +#include "paimon/core/stats/simple_stats.h" +#include "paimon/data/timestamp.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/testing/utils/binary_row_generator.h" +#include "paimon/testing/utils/testharness.h" +namespace paimon::test { +class SortedRunTest : public testing::Test { + public: + std::shared_ptr<DataFileMeta> CreateDataFileMeta(int32_t min_key, int32_t max_key) { + auto pool = GetDefaultPool(); + return std::make_shared<DataFileMeta>( + "fake.orc", /*file_size=*/1165, /*row_count=*/1, + /*min_key=*/BinaryRowGenerator::GenerateRow({min_key}, pool.get()), /*max_key=*/ + BinaryRowGenerator::GenerateRow({max_key}, pool.get()), + /*key_stats=*/ + SimpleStats::EmptyStats(), + /*value_stats=*/ + SimpleStats::EmptyStats(), + /*min_sequence_number=*/0, /*max_sequence_number=*/6, /*schema_id=*/0, + /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(), + /*creation_time=*/Timestamp(0ll, 0), + /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(), + /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt, + /*first_row_id=*/std::nullopt, + /*write_cols=*/std::nullopt); + } +}; + +TEST_F(SortedRunTest, TestSortedRunIsValid) { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<FieldsComparator> comparator, + FieldsComparator::Create({DataField(0, arrow::field("test", arrow::int32()))}, + /*is_ascending_order=*/true)); + + // m1 [10, 20] + auto m1 = CreateDataFileMeta(10, 20); + + // m2 [30, 40] + auto m2 = CreateDataFileMeta(30, 40); + + // m3 [15, 35] + auto m3 = CreateDataFileMeta(15, 35); + { + auto sorted_run = SortedRun::FromSingle(m1); + ASSERT_TRUE(sorted_run.IsValid(comparator)); + } + { + auto sorted_run = SortedRun::FromSorted({m1, m2}); + ASSERT_TRUE(sorted_run.IsValid(comparator)); + } + { + auto sorted_run = SortedRun::FromSorted({m1, m3}); + ASSERT_FALSE(sorted_run.IsValid(comparator)); + } + { + auto sorted_run = SortedRun::FromSorted({m3, m2}); + ASSERT_FALSE(sorted_run.IsValid(comparator)); + } +} + +TEST_F(SortedRunTest, TestFromUnsorted) { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<FieldsComparator> comparator, + FieldsComparator::Create({DataField(0, arrow::field("test", arrow::int32()))}, + /*is_ascending_order=*/true)); + + // m1 [10, 20] + auto m1 = CreateDataFileMeta(10, 20); + + // m2 [30, 40] + auto m2 = CreateDataFileMeta(30, 40); + + // m3 [15, 35] + auto m3 = CreateDataFileMeta(15, 35); + + ASSERT_OK_AND_ASSIGN(auto run1, SortedRun::FromUnsorted({m2, m1}, comparator)); + auto run2 = SortedRun::FromSorted({m1, m2}); + ASSERT_EQ(run1, run2); + + ASSERT_NOK_WITH_MSG(SortedRun::FromUnsorted({m2, m1, m3}, comparator), + "from unsorted validate failed"); +} + +TEST_F(SortedRunTest, TestEqual) { + auto empty = SortedRun::Empty(); + auto m1 = CreateDataFileMeta(10, 20); + auto run1 = SortedRun::FromSingle({m1}); + auto other_run1 = SortedRun::FromSingle({m1}); + + auto m2 = CreateDataFileMeta(100, 200); + auto run2 = SortedRun::FromSingle({m2}); + Review Comment: Same issue here: `SortedRun::FromSingle({m2})` won’t compile because `{m2}` is an initializer_list, not a `std::shared_ptr<DataFileMeta>`. ########## src/paimon/core/mergetree/levels_test.cpp: ########## @@ -0,0 +1,293 @@ +/* + * 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/mergetree/levels.h" + +#include "arrow/api.h" +#include "gtest/gtest.h" +#include "paimon/common/utils/uuid.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/testing/utils/binary_row_generator.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +class LevelsTest : public testing::Test { + public: + std::shared_ptr<DataFileMeta> CreateDataFileMeta(int32_t level, int64_t min_sequence_number, + int64_t max_sequence_number, + int64_t ts_second) const { + std::string uuid; + EXPECT_TRUE(UUID::Generate(&uuid)); + return std::make_shared<DataFileMeta>( + /*file_name=*/uuid, /*file_size=*/1, + /*row_count=*/max_sequence_number - min_sequence_number + 1, + BinaryRowGenerator::GenerateRow({min_sequence_number}, pool_.get()), + BinaryRowGenerator::GenerateRow({max_sequence_number}, pool_.get()), + SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), min_sequence_number, + max_sequence_number, + /*schema_id=*/0, level, std::vector<std::optional<std::string>>(), + Timestamp(ts_second, 0l), std::nullopt, nullptr, FileSource::Append(), std::nullopt, + std::nullopt, std::nullopt, std::nullopt); + } + + std::shared_ptr<FieldsComparator> CreateComparator() const { + std::vector<DataField> data_fields; + data_fields.emplace_back(/*id=*/0, arrow::field("f0", arrow::int32())); + EXPECT_OK_AND_ASSIGN(auto cmp, + FieldsComparator::Create(data_fields, /*is_ascending_order=*/true)); + return cmp; + } + + private: + std::shared_ptr<MemoryPool> pool_ = GetDefaultPool(); +}; + +TEST_F(LevelsTest, TestNonEmptyHighestLevelNo) { + std::vector<std::shared_ptr<DataFileMeta>> input_files; + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + ASSERT_EQ(levels->NonEmptyHighestLevel(), -1); +} + +TEST_F(LevelsTest, TestInvalidNumberLevels) { + std::vector<std::shared_ptr<DataFileMeta>> input_files; + ASSERT_NOK_WITH_MSG(Levels::Create(CreateComparator(), input_files, /*num_levels=*/1), + "Number of levels must be at least 2."); +} + +TEST_F(LevelsTest, TestAddLevel0FileInvalid) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = {CreateDataFileMeta(0, 0, 1, 0), + CreateDataFileMeta(0, 2, 3, 0)}; + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + std::vector<std::shared_ptr<DataFileMeta>> new_files = {CreateDataFileMeta(1, 0, 1, 0)}; + ASSERT_NOK_WITH_MSG(levels->AddLevel0File(new_files[0]), + "must add level0 file in AddLevel0File"); +} + +TEST_F(LevelsTest, TestNonEmptyHighestLevel0) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = {CreateDataFileMeta(0, 0, 1, 0), + CreateDataFileMeta(0, 2, 3, 0)}; + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + ASSERT_EQ(levels->NonEmptyHighestLevel(), 0); +} + +TEST_F(LevelsTest, TestNonEmptyHighestLevel1) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = {CreateDataFileMeta(0, 0, 1, 0), + CreateDataFileMeta(1, 2, 3, 0)}; + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + ASSERT_EQ(levels->NonEmptyHighestLevel(), 1); + ASSERT_EQ(levels->NumberOfLevels(), 3); + ASSERT_EQ(levels->MaxLevel(), 2); +} + +TEST_F(LevelsTest, TestNonEmptyHighestLevel2) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = { + CreateDataFileMeta(0, 0, 100, 0), CreateDataFileMeta(0, 100, 200, 0), + CreateDataFileMeta(0, 0, 200, 0), CreateDataFileMeta(0, 0, 200, 10), + CreateDataFileMeta(1, 0, 500, 0), CreateDataFileMeta(2, 0, 1000, 0)}; + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + ASSERT_EQ(levels->NonEmptyHighestLevel(), 2); + ASSERT_EQ(levels->TotalFileSize(), 6); + + std::vector<LevelSortedRun> expected_sorted_run = { + LevelSortedRun(0, SortedRun::FromSingle(input_files[2])), + LevelSortedRun(0, SortedRun::FromSingle(input_files[3])), + LevelSortedRun(0, SortedRun::FromSingle(input_files[1])), + LevelSortedRun(0, SortedRun::FromSingle(input_files[0])), + LevelSortedRun(1, SortedRun::FromSingle(input_files[4])), + LevelSortedRun(2, SortedRun::FromSingle(input_files[5])), + }; + + ASSERT_EQ(levels->LevelSortedRuns(), expected_sorted_run); + ASSERT_EQ(levels->NumberOfSortedRuns(), 6); + + std::vector<std::shared_ptr<DataFileMeta>> expected_all_files = { + input_files[2], input_files[3], input_files[1], + input_files[0], input_files[4], input_files[5]}; + ASSERT_EQ(levels->AllFiles(), expected_all_files); +} + +TEST_F(LevelsTest, TestAddLevel0File) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = { + CreateDataFileMeta(0, 100, 200, 0), CreateDataFileMeta(0, 0, 200, 0), + CreateDataFileMeta(0, 0, 200, 10), CreateDataFileMeta(1, 0, 500, 0), + CreateDataFileMeta(2, 0, 1000, 0)}; + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + ASSERT_EQ(levels->TotalFileSize(), 5); + + auto new_level0 = CreateDataFileMeta(0, 0, 100, 0); + ASSERT_OK(levels->AddLevel0File(new_level0)); + ASSERT_EQ(levels->TotalFileSize(), 6); + std::vector<LevelSortedRun> expected_sorted_run = { + LevelSortedRun(0, SortedRun::FromSingle(input_files[1])), + LevelSortedRun(0, SortedRun::FromSingle(input_files[2])), + LevelSortedRun(0, SortedRun::FromSingle(input_files[0])), + LevelSortedRun(0, SortedRun::FromSingle(new_level0)), + LevelSortedRun(1, SortedRun::FromSingle(input_files[3])), + LevelSortedRun(2, SortedRun::FromSingle(input_files[4])), + }; + + ASSERT_EQ(levels->LevelSortedRuns(), expected_sorted_run); + ASSERT_EQ(levels->NumberOfSortedRuns(), 6); +} + +TEST_F(LevelsTest, TestUpdate) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = { + CreateDataFileMeta(0, 100, 200, 0), CreateDataFileMeta(0, 0, 200, 0), + CreateDataFileMeta(0, 0, 200, 10), CreateDataFileMeta(1, 0, 500, 0), + CreateDataFileMeta(1, 600, 1000, 0)}; + + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + ASSERT_EQ(levels->TotalFileSize(), 5); + ASSERT_EQ(levels->NumberOfSortedRuns(), 4); + + std::vector<std::shared_ptr<DataFileMeta>> before = { + input_files[1], + input_files[3], + }; + + std::vector<std::shared_ptr<DataFileMeta>> after = {CreateDataFileMeta(0, 0, 100, 0), + CreateDataFileMeta(1, 0, 550, 0)}; + + ASSERT_OK(levels->Update(before, after)); + + std::vector<LevelSortedRun> expected_sorted_run = { + LevelSortedRun(0, SortedRun::FromSingle(input_files[2])), + LevelSortedRun(0, SortedRun::FromSingle(input_files[0])), + LevelSortedRun(0, SortedRun::FromSingle(after[0])), + LevelSortedRun(1, SortedRun::FromSorted({after[1], input_files[4]})), + }; + + ASSERT_EQ(levels->LevelSortedRuns(), expected_sorted_run); + ASSERT_EQ(levels->NumberOfSortedRuns(), 4); +} + +TEST_F(LevelsTest, TestRunOfLevelInvalidLevel) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = {CreateDataFileMeta(2, 0, 1, 0), + CreateDataFileMeta(1, 2, 3, 1)}; + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + + // Test invalid level 0 + auto result = Levels::RunOfLevel(0, levels->GetLevels()); + ASSERT_NOK_WITH_MSG(result, "Level0 does not have one single sorted run."); + + // Test invalid negative level + auto result_neg = Levels::RunOfLevel(-1, levels->GetLevels()); + ASSERT_NOK_WITH_MSG(result_neg, "Level0 does not have one single sorted run."); +} + +/// A simple test callback that records dropped file names. +class TestDropFileCallback : public Levels::DropFileCallback { + public: + void NotifyDropFile(const std::string& file) override { + dropped_files.push_back(file); + } + std::vector<std::string> dropped_files; +}; + +TEST_F(LevelsTest, TestUpdateDropFileCallback) { + std::vector<std::shared_ptr<DataFileMeta>> input_files = { + CreateDataFileMeta(0, 100, 200, 0), CreateDataFileMeta(0, 0, 200, 0), + CreateDataFileMeta(1, 0, 500, 0), CreateDataFileMeta(1, 600, 1000, 0)}; + + ASSERT_OK_AND_ASSIGN(auto levels, + Levels::Create(CreateComparator(), input_files, /*num_levels=*/3)); + + TestDropFileCallback callback; + levels->AddDropFileCallback(&callback); + + // Remove input_files[0] from level0 and input_files[2] from level1, + // add new files to level0 and level1. + std::vector<std::shared_ptr<DataFileMeta>> before = {input_files[0], input_files[2]}; + std::vector<std::shared_ptr<DataFileMeta>> after = {CreateDataFileMeta(0, 0, 100, 0), + CreateDataFileMeta(1, 0, 550, 0)}; + ASSERT_OK(levels->Update(before, after)); + + // Both files in before are replaced by new files, so both should be dropped. + ASSERT_EQ(callback.dropped_files.size(), 2); + std::set<std::string> dropped_set(callback.dropped_files.begin(), callback.dropped_files.end()); + ASSERT_TRUE(dropped_set.count(input_files[0]->file_name)); + ASSERT_TRUE(dropped_set.count(input_files[2]->file_name)); + + // Remove callback + ASSERT_EQ(levels->drop_file_callbacks_.size(), 1); + levels->RemoveDropFileCallback(nullptr); + ASSERT_EQ(levels->drop_file_callbacks_.size(), 1); + levels->RemoveDropFileCallback(&callback); + ASSERT_TRUE(levels->drop_file_callbacks_.empty()); Review Comment: The test directly accesses `levels->drop_file_callbacks_`, which is a private member of `Levels`; this will not compile. The test can validate callback removal behavior via the public API by removing the callback and ensuring subsequent `Update()` calls no longer record drops. ########## src/paimon/core/mergetree/levels.cpp: ########## @@ -0,0 +1,238 @@ +/* + * 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/mergetree/levels.h" + +#include <algorithm> +namespace paimon { +bool Levels::Level0Comparator::operator()(const std::shared_ptr<DataFileMeta>& a, + const std::shared_ptr<DataFileMeta>& b) const { + if (a->max_sequence_number != b->max_sequence_number) { + // file with larger sequence number should be in front + return a->max_sequence_number > b->max_sequence_number; + } else { + // When two or more jobs are writing the same merge tree, it is + // possible that multiple files have the same maxSequenceNumber. In + // this case we have to compare their file names so that files with + // same maxSequenceNumber won't be "de-duplicated" by the tree set. + int64_t min_seq_a = a->min_sequence_number; + int64_t min_seq_b = b->min_sequence_number; + if (min_seq_a != min_seq_b) { + return min_seq_a < min_seq_b; + } + // If minSequenceNumber is also the same, use creation time + Timestamp time_a = a->creation_time; + Timestamp time_b = b->creation_time; + if (time_a != time_b) { + return time_a < time_b; + } + // Final fallback: filename (to ensure uniqueness in set) + return a->file_name < b->file_name; + } +} + +Result<std::unique_ptr<Levels>> Levels::Create( + const std::shared_ptr<FieldsComparator>& key_comparator, + const std::vector<std::shared_ptr<DataFileMeta>>& input_files, int32_t num_levels) { + // in case the num of levels is not specified explicitly + int32_t restored_num_levels = -1; + for (const auto& file : input_files) { + if (file->level > restored_num_levels) { + restored_num_levels = file->level; + } + } + restored_num_levels = std::max(restored_num_levels + 1, num_levels); + if (restored_num_levels <= 1) { + return Status::Invalid("Number of levels must be at least 2."); + } + + std::set<std::shared_ptr<DataFileMeta>, Levels::Level0Comparator> level0; + std::vector<SortedRun> levels; + levels.reserve(restored_num_levels - 1); + for (int32_t i = 1; i < restored_num_levels; ++i) { + levels.push_back(SortedRun::Empty()); + } + auto level_map = GroupByLevel(input_files); + for (auto& [level, files] : level_map) { + PAIMON_RETURN_NOT_OK( + UpdateLevel(level, /*before=*/{}, /*after=*/files, key_comparator, &levels, &level0)); + } + + size_t total_file_num = level0.size(); + for (const auto& run : levels) { + total_file_num += run.Files().size(); + } + if (total_file_num != input_files.size()) { + return Status::Invalid( + "Number of files stored in Levels does not equal to the size of inputFiles. This " + "is unexpected."); + } + return std::unique_ptr<Levels>(new Levels(key_comparator, level0, levels)); +} + +int32_t Levels::NumberOfSortedRuns() const { + int32_t number_of_runs = level0_.size(); + for (const auto& run : levels_) { + if (!run.IsEmpty()) { + number_of_runs++; + } + } + return number_of_runs; +} + +void Levels::AddDropFileCallback(DropFileCallback* callback) { + drop_file_callbacks_.push_back(callback); +} + +void Levels::RemoveDropFileCallback(DropFileCallback* callback) { + drop_file_callbacks_.erase( + std::remove(drop_file_callbacks_.begin(), drop_file_callbacks_.end(), callback), + drop_file_callbacks_.end()); +} + +Status Levels::AddLevel0File(const std::shared_ptr<DataFileMeta>& file) { + if (file->level != 0) { + return Status::Invalid("must add level0 file in AddLevel0File"); + } + level0_.insert(file); + return Status::OK(); +} + +int32_t Levels::NonEmptyHighestLevel() const { + for (int32_t i = levels_.size() - 1; i >= 0; i--) { + if (!levels_[i].IsEmpty()) { + return i + 1; + } + } + return level0_.empty() ? -1 : 0; +} + +int64_t Levels::TotalFileSize() const { + int64_t total_size = 0; + for (const auto& file : level0_) { + total_size += file->file_size; + } + for (const auto& run : levels_) { + total_size += run.TotalSize(); + } + return total_size; +} + +std::vector<std::shared_ptr<DataFileMeta>> Levels::AllFiles() const { + std::vector<std::shared_ptr<DataFileMeta>> all_files; + auto runs = LevelSortedRuns(); + for (const auto& run : runs) { + all_files.insert(all_files.end(), run.run.Files().begin(), run.run.Files().end()); + } + return all_files; +} + +std::vector<LevelSortedRun> Levels::LevelSortedRuns() const { + std::vector<LevelSortedRun> runs; + for (const auto& file : level0_) { + runs.emplace_back(/*level=*/0, SortedRun::FromSingle(file)); + } + for (int32_t i = 0; i < static_cast<int32_t>(levels_.size()); i++) { + const auto& run = levels_[i]; + if (!run.IsEmpty()) { + runs.emplace_back(/*level=*/i + 1, run); + } + } + return runs; +} + +Status Levels::Update(const std::vector<std::shared_ptr<DataFileMeta>>& before, + const std::vector<std::shared_ptr<DataFileMeta>>& after) { + auto grouped_before = GroupByLevel(before); + auto grouped_after = GroupByLevel(after); + int32_t number_of_levels = NumberOfLevels(); + for (int32_t i = 0; i < number_of_levels; i++) { + PAIMON_RETURN_NOT_OK(UpdateLevel(i, grouped_before[i], grouped_after[i], key_comparator_, + &levels_, &level0_)); + } + if (!drop_file_callbacks_.empty()) { + std::set<std::string> dropped_files; + for (const auto& file : before) { + dropped_files.insert(file->file_name); + } + // exclude upgrade files + for (const auto& file : after) { + dropped_files.erase(file->file_name); + } + for (auto* callback : drop_file_callbacks_) { + if (!callback) { + continue; + } + for (const auto& file_name : dropped_files) { + callback->NotifyDropFile(file_name); + } + } + } + return Status::OK(); +} + +Status Levels::UpdateLevel(int32_t level, const std::vector<std::shared_ptr<DataFileMeta>>& before, + const std::vector<std::shared_ptr<DataFileMeta>>& after, + const std::shared_ptr<FieldsComparator>& key_comparator, + std::vector<SortedRun>* levels, + std::set<std::shared_ptr<DataFileMeta>, Level0Comparator>* level0) { + if (before.empty() && after.empty()) { + return Status::OK(); + } + if (level == 0) { + for (const auto& file : before) { + level0->erase(file); + } + for (const auto& file : after) { + level0->insert(file); + } + } else { + PAIMON_ASSIGN_OR_RAISE(SortedRun run, RunOfLevel(level, *levels)); + std::vector<std::shared_ptr<DataFileMeta>> files = run.Files(); + for (const auto& before_file : before) { + auto iter = std::find_if(files.begin(), files.end(), [&before_file](const auto& cur) { + return before_file->file_name == cur->file_name; + }); + if (iter != files.end()) { + files.erase(iter); + } + } + files.insert(files.end(), after.begin(), after.end()); + PAIMON_ASSIGN_OR_RAISE((*levels)[level - 1], + SortedRun::FromUnsorted(files, key_comparator)); + } + return Status::OK(); +} + +Result<SortedRun> Levels::RunOfLevel(int32_t level, const std::vector<SortedRun>& levels) { + if (level <= 0) { + return Status::Invalid("Level0 does not have one single sorted run."); + } + return levels[level - 1]; +} Review Comment: `RunOfLevel` validates `level <= 0` but not `level > levels.size()`. Passing an out-of-range level will cause an out-of-bounds access (`levels[level - 1]`). Since this is a public helper used outside this file, it should guard the upper bound too. ########## src/paimon/core/mergetree/sorted_run_test.cpp: ########## @@ -0,0 +1,143 @@ +/* + * 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/mergetree/sorted_run.h" + +#include <optional> +#include <string> +#include <variant> + +#include "arrow/type_fwd.h" +#include "gtest/gtest.h" +#include "paimon/common/types/data_field.h" +#include "paimon/core/manifest/file_source.h" +#include "paimon/core/mergetree/level_sorted_run.h" +#include "paimon/core/stats/simple_stats.h" +#include "paimon/data/timestamp.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/testing/utils/binary_row_generator.h" +#include "paimon/testing/utils/testharness.h" +namespace paimon::test { +class SortedRunTest : public testing::Test { + public: + std::shared_ptr<DataFileMeta> CreateDataFileMeta(int32_t min_key, int32_t max_key) { + auto pool = GetDefaultPool(); + return std::make_shared<DataFileMeta>( + "fake.orc", /*file_size=*/1165, /*row_count=*/1, + /*min_key=*/BinaryRowGenerator::GenerateRow({min_key}, pool.get()), /*max_key=*/ + BinaryRowGenerator::GenerateRow({max_key}, pool.get()), + /*key_stats=*/ + SimpleStats::EmptyStats(), + /*value_stats=*/ + SimpleStats::EmptyStats(), + /*min_sequence_number=*/0, /*max_sequence_number=*/6, /*schema_id=*/0, + /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(), + /*creation_time=*/Timestamp(0ll, 0), + /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(), + /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt, + /*first_row_id=*/std::nullopt, + /*write_cols=*/std::nullopt); + } +}; + +TEST_F(SortedRunTest, TestSortedRunIsValid) { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<FieldsComparator> comparator, + FieldsComparator::Create({DataField(0, arrow::field("test", arrow::int32()))}, + /*is_ascending_order=*/true)); + + // m1 [10, 20] + auto m1 = CreateDataFileMeta(10, 20); + + // m2 [30, 40] + auto m2 = CreateDataFileMeta(30, 40); + + // m3 [15, 35] + auto m3 = CreateDataFileMeta(15, 35); + { + auto sorted_run = SortedRun::FromSingle(m1); + ASSERT_TRUE(sorted_run.IsValid(comparator)); + } + { + auto sorted_run = SortedRun::FromSorted({m1, m2}); + ASSERT_TRUE(sorted_run.IsValid(comparator)); + } + { + auto sorted_run = SortedRun::FromSorted({m1, m3}); + ASSERT_FALSE(sorted_run.IsValid(comparator)); + } + { + auto sorted_run = SortedRun::FromSorted({m3, m2}); + ASSERT_FALSE(sorted_run.IsValid(comparator)); + } +} + +TEST_F(SortedRunTest, TestFromUnsorted) { + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<FieldsComparator> comparator, + FieldsComparator::Create({DataField(0, arrow::field("test", arrow::int32()))}, + /*is_ascending_order=*/true)); + + // m1 [10, 20] + auto m1 = CreateDataFileMeta(10, 20); + + // m2 [30, 40] + auto m2 = CreateDataFileMeta(30, 40); + + // m3 [15, 35] + auto m3 = CreateDataFileMeta(15, 35); + + ASSERT_OK_AND_ASSIGN(auto run1, SortedRun::FromUnsorted({m2, m1}, comparator)); + auto run2 = SortedRun::FromSorted({m1, m2}); + ASSERT_EQ(run1, run2); + + ASSERT_NOK_WITH_MSG(SortedRun::FromUnsorted({m2, m1, m3}, comparator), + "from unsorted validate failed"); +} + +TEST_F(SortedRunTest, TestEqual) { + auto empty = SortedRun::Empty(); + auto m1 = CreateDataFileMeta(10, 20); + auto run1 = SortedRun::FromSingle({m1}); + auto other_run1 = SortedRun::FromSingle({m1}); + Review Comment: `SortedRun::FromSingle` takes a `std::shared_ptr<DataFileMeta>`; the braced initializer (`{m1}`) doesn’t match any overload and will not compile. Use `FromSingle(m1)` instead. -- 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]
