Skip to content

Commit 7205637

Browse files
zjw1111zjw1111
committed
fix review
1 parent 7f14cf7 commit 7205637

8 files changed

Lines changed: 35 additions & 34 deletions

File tree

‎src/paimon/CMakeLists.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -251,7 +251,7 @@ set(PAIMON_CORE_SRCS
251251
core/mergetree/compact/lookup_merge_tree_compact_rewriter.cpp
252252
core/mergetree/compact/changelog_merge_tree_rewriter.cpp
253253
core/mergetree/merge_tree_writer.cpp
254-
core/mergetree/binary_in_memory_sort_buffer.cpp
254+
core/mergetree/in_memory_sort_buffer.cpp
255255
core/mergetree/write_buffer.cpp
256256
core/mergetree/levels.cpp
257257
core/mergetree/lookup_file.cpp

‎src/paimon/core/append/append_only_writer.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,8 @@ class AppendOnlyWriter : public BatchWriter {
7070
}
7171

7272
uint64_t GetMemoryUsage() const override {
73+
// used for spill, AppendOnlyWriter do not support spill, so return 0 to avoid triggering
74+
// spill
7375
return 0;
7476
}
7577
Status FlushMemory() override {

src/paimon/core/mergetree/binary_in_memory_sort_buffer.cpp renamed to src/paimon/core/mergetree/in_memory_sort_buffer.cpp

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
* limitations under the License.
1515
*/
1616

17-
#include "paimon/core/mergetree/binary_in_memory_sort_buffer.h"
17+
#include "paimon/core/mergetree/in_memory_sort_buffer.h"
1818

1919
#include <cassert>
2020
#include <utility>
@@ -33,12 +33,14 @@
3333

3434
namespace paimon {
3535

36-
BinaryInMemorySortBuffer::BinaryInMemorySortBuffer(
37-
int64_t last_sequence_number, const std::shared_ptr<arrow::DataType>& value_type,
38-
const std::vector<std::string>& trimmed_primary_keys,
39-
const std::vector<std::string>& user_defined_sequence_fields, bool sequence_fields_ascending,
40-
const std::shared_ptr<FieldsComparator>& key_comparator, uint64_t write_buffer_size,
41-
const std::shared_ptr<MemoryPool>& pool)
36+
InMemorySortBuffer::InMemorySortBuffer(int64_t last_sequence_number,
37+
const std::shared_ptr<arrow::DataType>& value_type,
38+
const std::vector<std::string>& trimmed_primary_keys,
39+
const std::vector<std::string>& user_defined_sequence_fields,
40+
bool sequence_fields_ascending,
41+
const std::shared_ptr<FieldsComparator>& key_comparator,
42+
uint64_t write_buffer_size,
43+
const std::shared_ptr<MemoryPool>& pool)
4244
: pool_(pool),
4345
value_type_(value_type),
4446
trimmed_primary_keys_(trimmed_primary_keys),
@@ -48,20 +50,20 @@ BinaryInMemorySortBuffer::BinaryInMemorySortBuffer(
4850
write_buffer_size_(write_buffer_size),
4951
next_sequence_number_(last_sequence_number + 1) {}
5052

51-
void BinaryInMemorySortBuffer::Clear() {
53+
void InMemorySortBuffer::Clear() {
5254
buffered_batches_.clear();
5355
current_memory_in_bytes_ = 0;
5456
}
5557

56-
uint64_t BinaryInMemorySortBuffer::GetMemorySize() const {
58+
uint64_t InMemorySortBuffer::GetMemorySize() const {
5759
return current_memory_in_bytes_;
5860
}
5961

60-
Result<bool> BinaryInMemorySortBuffer::FlushMemory() {
62+
Result<bool> InMemorySortBuffer::FlushMemory() {
6163
return false;
6264
}
6365

64-
Result<bool> BinaryInMemorySortBuffer::Write(std::unique_ptr<RecordBatch>&& moved_batch) {
66+
Result<bool> InMemorySortBuffer::Write(std::unique_ptr<RecordBatch>&& moved_batch) {
6567
if (ArrowArrayIsReleased(moved_batch->GetData())) {
6668
return Status::Invalid("invalid batch: data is released");
6769
}
@@ -84,8 +86,7 @@ Result<bool> BinaryInMemorySortBuffer::Write(std::unique_ptr<RecordBatch>&& move
8486
return current_memory_in_bytes_ < write_buffer_size_;
8587
}
8688

87-
Result<std::vector<std::unique_ptr<KeyValueRecordReader>>>
88-
BinaryInMemorySortBuffer::CreateReaders() {
89+
Result<std::vector<std::unique_ptr<KeyValueRecordReader>>> InMemorySortBuffer::CreateReaders() {
8990
std::vector<std::unique_ptr<KeyValueRecordReader>> readers;
9091
if (buffered_batches_.empty()) {
9192
return readers;
@@ -102,14 +103,13 @@ BinaryInMemorySortBuffer::CreateReaders() {
102103
return readers;
103104
}
104105

105-
bool BinaryInMemorySortBuffer::HasData() const {
106+
bool InMemorySortBuffer::HasData() const {
106107
return !buffered_batches_.empty();
107108
}
108109

109110
// TODO(jinli.zjw): Consider making the memory estimation more accurate.
110111
// https://github.com/alibaba/paimon-cpp/pull/206#discussion_r3021325389
111-
Result<int64_t> BinaryInMemorySortBuffer::EstimateMemoryUse(
112-
const std::shared_ptr<arrow::Array>& array) {
112+
Result<int64_t> InMemorySortBuffer::EstimateMemoryUse(const std::shared_ptr<arrow::Array>& array) {
113113
arrow::Type::type type = array->type()->id();
114114
int64_t null_bits_size_in_bytes = (array->length() + 7) / 8;
115115
switch (type) {

src/paimon/core/mergetree/binary_in_memory_sort_buffer.h renamed to src/paimon/core/mergetree/in_memory_sort_buffer.h

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -47,15 +47,15 @@ struct BufferedWriteBatch {
4747

4848
/// Pure in-memory SortBuffer: buffers RecordBatches and exposes them as sorted
4949
/// KeyValueRecordReaders. Does not support spill to disk.
50-
class BinaryInMemorySortBuffer : public SortBuffer {
50+
class InMemorySortBuffer : public SortBuffer {
5151
public:
52-
BinaryInMemorySortBuffer(int64_t last_sequence_number,
53-
const std::shared_ptr<arrow::DataType>& value_type,
54-
const std::vector<std::string>& trimmed_primary_keys,
55-
const std::vector<std::string>& user_defined_sequence_fields,
56-
bool sequence_fields_ascending,
57-
const std::shared_ptr<FieldsComparator>& key_comparator,
58-
uint64_t write_buffer_size, const std::shared_ptr<MemoryPool>& pool);
52+
InMemorySortBuffer(int64_t last_sequence_number,
53+
const std::shared_ptr<arrow::DataType>& value_type,
54+
const std::vector<std::string>& trimmed_primary_keys,
55+
const std::vector<std::string>& user_defined_sequence_fields,
56+
bool sequence_fields_ascending,
57+
const std::shared_ptr<FieldsComparator>& key_comparator,
58+
uint64_t write_buffer_size, const std::shared_ptr<MemoryPool>& pool);
5959

6060
void Clear() override;
6161
uint64_t GetMemorySize() const override;

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
#include "arrow/type.h"
2424
#include "paimon/core/io/key_value_record_reader.h"
2525
#include "paimon/core/io/merged_key_value_record_reader.h"
26-
#include "paimon/core/mergetree/binary_in_memory_sort_buffer.h"
26+
#include "paimon/core/mergetree/in_memory_sort_buffer.h"
2727

2828
namespace paimon {
2929

@@ -36,7 +36,7 @@ WriteBuffer::WriteBuffer(
3636
const std::shared_ptr<MemoryPool>& pool)
3737
: key_comparator_(key_comparator), merge_function_wrapper_(merge_function_wrapper) {
3838
// TODO(jinli.zjw): input sequence_fields_ascending as parameter
39-
sort_buffer_ = std::make_unique<BinaryInMemorySortBuffer>(
39+
sort_buffer_ = std::make_unique<InMemorySortBuffer>(
4040
last_sequence_number, value_type, trimmed_primary_keys, user_defined_sequence_fields,
4141
/*sequence_fields_ascending=*/true, key_comparator,
4242
/*write_buffer_size=*/std::numeric_limits<int64_t>::max(), pool);

‎src/paimon/core/mergetree/write_buffer.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
#include <vector>
2323

2424
#include "arrow/type_fwd.h"
25-
#include "paimon/core/mergetree/binary_in_memory_sort_buffer.h"
25+
#include "paimon/core/mergetree/in_memory_sort_buffer.h"
2626
#include "paimon/core/mergetree/sort_buffer.h"
2727
#include "paimon/record_batch.h"
2828
#include "paimon/result.h"

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

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -179,8 +179,7 @@ TEST_F(WriteBufferTest, TestEstimateMemoryUse) {
179179
["Alice", 10, 0, 13.1]
180180
])")
181181
.ValueOrDie();
182-
ASSERT_OK_AND_ASSIGN(int64_t memory_use,
183-
BinaryInMemorySortBuffer::EstimateMemoryUse(array));
182+
ASSERT_OK_AND_ASSIGN(int64_t memory_use, InMemorySortBuffer::EstimateMemoryUse(array));
184183
int64_t expected_memory_use =
185184
1 + (13 + 3 * 4 + 1) + (3 * 4 + 1) + (3 * 4 + 1) + (3 * 8 + 1);
186185
ASSERT_EQ(memory_use, expected_memory_use);
@@ -207,8 +206,7 @@ TEST_F(WriteBufferTest, TestEstimateMemoryUse) {
207206
[true, 0, 0, 0, 0, 1.4E-45, 4.9E-324, 0, 0, "0.00000000000000000000", "Alice", "wood"]
208207
])")
209208
.ValueOrDie());
210-
ASSERT_OK_AND_ASSIGN(int64_t memory_use,
211-
BinaryInMemorySortBuffer::EstimateMemoryUse(array));
209+
ASSERT_OK_AND_ASSIGN(int64_t memory_use, InMemorySortBuffer::EstimateMemoryUse(array));
212210
int64_t expected_memory_use = 1 + (4 + 1) + (4 + 1) + (2 * 4 + 1) + (4 * 4 + 1) +
213211
(8 * 4 + 1) + (4 * 4 + 1) + (8 * 4 + 1) + (4 * 4 + 1) +
214212
(8 * 4 + 1) + (4 * 16 + 1) + (25 + 4 * 4 + 1) +
@@ -230,8 +228,7 @@ TEST_F(WriteBufferTest, TestEstimateMemoryUse) {
230228
[[6], [["elephant", 7], ["fox", 8]], [null, 30.1, true]]
231229
])")
232230
.ValueOrDie());
233-
ASSERT_OK_AND_ASSIGN(int64_t memory_use,
234-
BinaryInMemorySortBuffer::EstimateMemoryUse(array));
231+
ASSERT_OK_AND_ASSIGN(int64_t memory_use, InMemorySortBuffer::EstimateMemoryUse(array));
235232
int64_t list_mem = 1 + (4 * 6 + 1);
236233
int64_t map_mem = 1 + (33 + 4 * 7 + 1) + (8 * 7 + 1);
237234
int64_t struct_mem = 1 + (8 * 3 + 1) + (8 * 3 + 1) + (1 * 3 + 1);

‎src/paimon/core/postpone/postpone_bucket_writer.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,8 @@ class PostponeBucketWriter : public BatchWriter {
7373
}
7474

7575
uint64_t GetMemoryUsage() const override {
76+
// used for spill, PostponeBucketWriter do not support spill, so return 0 to avoid
77+
// triggering spill
7678
return 0;
7779
}
7880

0 commit comments

Comments
 (0)