Skip to content

refactor: extract WriteBuffer from MergeTreeWriter #206

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zjw1111 merged 11 commits into from
Apr 3, 2026
Merged

refactor: extract WriteBuffer from MergeTreeWriter #206

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
f3db380
refactor: extract WriteBuffer from MergeTreeWriter
zjw1111 Apr 1, 2026
6643041
refactor: remove WriteBuffer GetBatchCount
zjw1111 Apr 1, 2026
4061117
test: fix arrow struct FieldVector calls
zjw1111 Apr 2, 2026
793c53b
test: fix unstable case in TestOrphanFilesCleanWithIOException
zjw1111 Apr 2, 2026
1176255
Merge branch 'main' into refactor/write-buffer
zjw1111 Apr 2, 2026
ff02c94
fix
zjw1111 Apr 2, 2026
bdf52d4
fix
zjw1111 Apr 2, 2026
a47d52d
fix review
zjw1111 Apr 2, 2026
a4fef57
Merge branch 'main' into refactor/write-buffer
zjw1111 Apr 3, 2026
334df23
fix review
zjw1111 Apr 3, 2026
265e9e0
Merge branch 'refactor/write-buffer' of https://github.com/zjw1111/pa…
zjw1111 Apr 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/paimon/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,7 @@ set(PAIMON_CORE_SRCS
core/mergetree/compact/lookup_merge_tree_compact_rewriter.cpp
core/mergetree/compact/changelog_merge_tree_rewriter.cpp
core/mergetree/merge_tree_writer.cpp
core/mergetree/write_buffer.cpp
core/mergetree/levels.cpp
core/mergetree/lookup_levels.cpp
core/migrate/file_meta_utils.cpp
Expand Down Expand Up @@ -592,6 +593,7 @@ if(PAIMON_BUILD_TESTS)
core/mergetree/lookup/persist_processor_test.cpp
core/mergetree/drop_delete_reader_test.cpp
core/mergetree/merge_tree_writer_test.cpp
core/mergetree/write_buffer_test.cpp
core/mergetree/sorted_run_test.cpp
core/migrate/file_meta_utils_test.cpp
core/operation/metrics/compaction_metrics_test.cpp
Expand Down
6 changes: 3 additions & 3 deletions src/paimon/common/data/internal_row_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ TEST(InternalRowTest, TestCreateFieldGetter) {
};

auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[true, 1, 2, 3, 4, 5.1, 6.12, "abc", "def", "1970-01-02 00:00:01", "1970-01-02 00:00:00.001",
"1970-01-02 00:00:00.000001", "1970-01-02 00:00:00.000000001", "1970-01-02 00:00:02", "1970-01-02 00:00:00.002",
"1970-01-02 00:00:00.000002", "1970-01-02 00:00:00.000000002", "-123456789987654321.45678", 12345,
Expand Down Expand Up @@ -138,7 +138,7 @@ TEST(InternalRowTest, TestCreateFieldGetterWithNull) {
arrow::field("f1", arrow::int8())};

auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[true, null]
])")
.ValueOrDie());
Expand All @@ -161,7 +161,7 @@ TEST(InternalRowTest, TestCreateFieldGetterWithInvalidType) {
arrow::FieldVector fields = {arrow::field("f0", arrow::large_utf8())};

auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
["hello"]
])")
.ValueOrDie());
Expand Down
4 changes: 2 additions & 2 deletions src/paimon/common/data/record_batch_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ TEST(RecordBatchTest, TestAssignAndMove) {
arrow::field("f1", arrow::int8())};
std::map<std::string, std::string> partition = {{"f1", "1"}};
auto old_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[true, 1]
])")
.ValueOrDie());
Expand All @@ -97,7 +97,7 @@ TEST(RecordBatchTest, TestAssignAndMove) {
&old_arrow_array);

auto new_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[false, 1]
])")
.ValueOrDie());
Expand Down
6 changes: 3 additions & 3 deletions src/paimon/common/global_index/complete_index_score_batch_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ TEST_F(CompleteIndexScoreBatchReaderTest, TestSimple) {
arrow::field("_ROW_ID", arrow::int64()),
};

auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
["Alice", 10, null, 0],
["Bob", 11, null, 1],
["Cathy", 12, null, 2]
Expand Down Expand Up @@ -106,7 +106,7 @@ TEST_F(CompleteIndexScoreBatchReaderTest, TestWithBitmap) {
arrow::field("_ROW_ID", arrow::int64()),
};

auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
["Alice", 10, null, 0],
["Bob", 11, null, 1],
["Cathy", 12, null, 2],
Expand Down Expand Up @@ -141,7 +141,7 @@ TEST_F(CompleteIndexScoreBatchReaderTest, TestReadWithNullScores) {
arrow::field("_ROW_ID", arrow::int64()),
};

auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
["Alice", 10, null, 0],
["Bob", 11, null, 1],
["Cathy", 12, null, 2]
Expand Down
2 changes: 1 addition & 1 deletion src/paimon/common/reader/complete_row_kind_batch_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ TEST_F(CompleteRowKindBatchReaderTest, TestNestedType) {
arrow::field("f1", arrow::map(arrow::struct_({field("a", arrow::int64()),
field("b", arrow::boolean())}),
arrow::boolean()))};
auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields}), R"([
auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
[[null, [1, true], null], [[[1, true], true]]],
[[[2, false], null], null],
[[[2, false], [3, true], [4, null]], [[[1, true], true], [[5, false], null]]],
Expand Down
8 changes: 4 additions & 4 deletions src/paimon/core/io/concat_key_value_record_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ TEST_F(ConcatKeyValueRecordReaderTest, TestSimple) {
arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
std::shared_ptr<arrow::Schema> value_schema =
arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4], fields[5], fields[6]}));
std::shared_ptr<arrow::DataType> src_type = arrow::struct_({fields});
std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
[0, 0, 1, 1, 10, 20, 30],
Expand Down Expand Up @@ -130,7 +130,7 @@ TEST_F(ConcatKeyValueRecordReaderTest, TestSingleReaderInConcat) {
arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
std::shared_ptr<arrow::Schema> value_schema =
arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4], fields[5], fields[6]}));
std::shared_ptr<arrow::DataType> src_type = arrow::struct_({fields});
std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
[0, 0, 1, 1, 10, 20, 30],
Expand Down Expand Up @@ -159,7 +159,7 @@ TEST_F(ConcatKeyValueRecordReaderTest, TestEmptyResult) {
arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
std::shared_ptr<arrow::Schema> value_schema =
arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4], fields[5], fields[6]}));
std::shared_ptr<arrow::DataType> src_type = arrow::struct_({fields});
std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
])")
Expand All @@ -180,7 +180,7 @@ TEST_F(ConcatKeyValueRecordReaderTest, TestEmptyReader) {
arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
std::shared_ptr<arrow::Schema> value_schema =
arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4], fields[5], fields[6]}));
std::shared_ptr<arrow::DataType> src_type = arrow::struct_({fields});
std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
])")
Expand Down
21 changes: 10 additions & 11 deletions src/paimon/core/io/field_mapping_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
#include <ostream>
#include <string>
#include <utility>
#include <variant>

#include "arrow/api.h"
#include "arrow/array/array_base.h"
Expand Down Expand Up @@ -418,7 +417,7 @@ TEST_F(FieldMappingReaderTest, TestDictionaryTypeWithSchemaEvolution) {
std::shared_ptr<arrow::Schema> data_schema =
DataField::ConvertDataFieldsToArrowSchema(data_fields);
auto data_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({data_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(data_schema->fields()), R"([
["apple", 4.0, 5.1, 10, 100, 1000, 10000, true],
["banana", 4.1, 6.2, 10, 200, 1000, 20000, null],
[null, 4.2, null, 10, 300, 1000, 30000, true],
Expand All @@ -443,7 +442,7 @@ TEST_F(FieldMappingReaderTest, TestDictionaryTypeWithSchemaEvolution) {
std::vector<std::string> partition_keys = {"f3", "f5"};
BinaryRow partition = BinaryRowGenerator::GenerateRow({10, 1000}, pool_.get());
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({read_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), R"([
[10000, 5.1, "apple", 4.0, 1000, 10, null, true, 100],
[20000, 6.2, "banana", 4.1, 1000, 10, null, null, 200],
[30000, null, null, 4.2, 1000, 10, null, true, 300],
Expand All @@ -464,7 +463,7 @@ TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithModifyType) {
std::shared_ptr<arrow::Schema> data_schema =
DataField::ConvertDataFieldsToArrowSchema(data_fields);
auto data_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({data_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(data_schema->fields()), R"([
["true", 4.0, 5.1, 10],
["False", 4.1, 6.2, 10],
[null, 4.2, null, 10],
Expand All @@ -485,7 +484,7 @@ TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithModifyType) {
std::vector<std::string> partition_keys = {"f3"};
BinaryRow partition = BinaryRowGenerator::GenerateRow({10}, pool_.get());
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({read_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), R"([
[true, "4", 5, 10],
[false, "4.1", 6, 10],
[null, "4.2", null, 10],
Expand All @@ -506,7 +505,7 @@ TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithModifyTypeWithDict) {
std::shared_ptr<arrow::Schema> data_schema =
DataField::ConvertDataFieldsToArrowSchema(data_fields);
auto data_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({data_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(data_schema->fields()), R"([
["true", 4.0, 5.1, 10],
["false", 4.1, 6.2, 10],
[null, 4.2, null, 10],
Expand All @@ -527,7 +526,7 @@ TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithModifyTypeWithDict) {
std::vector<std::string> partition_keys = {"f3"};
BinaryRow partition = BinaryRowGenerator::GenerateRow({10}, pool_.get());
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({read_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), R"([
[true, "4", 5, 10],
[false, "4.1", 6, 10],
[null, "4.2", null, 10],
Expand All @@ -548,7 +547,7 @@ TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithModifyTypeWithPredicate) {
std::shared_ptr<arrow::Schema> data_schema =
DataField::ConvertDataFieldsToArrowSchema(data_fields);
auto data_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({data_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(data_schema->fields()), R"([
["true", 4.0, 5, 10],
["False", 4.1, 6, 10],
[null, 4.2, null, 10],
Expand Down Expand Up @@ -615,7 +614,7 @@ TEST_F(FieldMappingReaderTest, TestReadWithSchemaEvolutionWithRenameAndModifyTyp
/*field_index=*/3, /*field_name=*/"f0", FieldType::STRING,
Literal(FieldType::STRING, literal_str.data(), literal_str.size()));
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({read_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), R"([
[0, "Emily", null, "15.1", 10],
[0, "Bob", null, "12.1", 10],
[0, "Alex", null, "16.1", 10]
Expand All @@ -632,7 +631,7 @@ TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithDictType) {
std::shared_ptr<arrow::Schema> data_schema =
DataField::ConvertDataFieldsToArrowSchema(data_fields);
auto data_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({data_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(data_schema->fields()), R"([
["Bob", 4.0, 5.1, 10],
["Emily", 4.1, 6.2, 10],
["Alice", 4.2, null, 10],
Expand All @@ -653,7 +652,7 @@ TEST_F(FieldMappingReaderTest, TestSchemaEvolutionWithDictType) {
std::vector<std::string> partition_keys = {"f3"};
BinaryRow partition = BinaryRowGenerator::GenerateRow({10}, pool_.get());
auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({read_schema->fields()}), R"([
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), R"([
[10, "4", "Bob", 5],
[10, "4.1", "Emily", 6],
[10, "4.2", "Alice", null],
Expand Down
6 changes: 3 additions & 3 deletions src/paimon/core/io/key_value_data_file_record_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -317,7 +317,7 @@ TEST_F(KeyValueDataFileRecordReaderTest, TestEmptyReader) {
arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
std::shared_ptr<arrow::Schema> value_schema =
arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4], fields[5], fields[6]}));
std::shared_ptr<arrow::DataType> src_type = arrow::struct_({fields});
std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
])")
Expand Down Expand Up @@ -347,7 +347,7 @@ TEST_F(KeyValueDataFileRecordReaderTest, TestInvalidSequenceNumerColumn) {
arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
std::shared_ptr<arrow::Schema> value_schema =
arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4], fields[5], fields[6]}));
std::shared_ptr<arrow::DataType> src_type = arrow::struct_({fields});
std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
["0", 0, 1, 1, 10, 20, 30]
Expand Down Expand Up @@ -376,7 +376,7 @@ TEST_F(KeyValueDataFileRecordReaderTest, TestInvalidValueKindColumn) {
arrow::schema(arrow::FieldVector({fields[2], fields[3]}));
std::shared_ptr<arrow::Schema> value_schema =
arrow::schema(arrow::FieldVector({fields[2], fields[3], fields[4], fields[5], fields[6]}));
std::shared_ptr<arrow::DataType> src_type = arrow::struct_({fields});
std::shared_ptr<arrow::DataType> src_type = arrow::struct_(fields);
auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
[0, 100, 1, 1, 10, 20, 30]
Expand Down
Loading
Loading