Skip to content

Commit 7b3d3f7

Browse files
committed
feat(shredding): add shared-shredding map placement policies
1 parent b631683 commit 7b3d3f7

36 files changed

Lines changed: 1231 additions & 489 deletions

include/paimon/defs.h

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -379,6 +379,12 @@ struct PAIMON_EXPORT Options {
379379
/// map.storage-layout = shared-shredding. Rows with more fields than K_max spill to
380380
/// __overflow. Default value is 256. Each column can have its own max-columns setting.
381381
static const char MAP_SHARED_SHREDDING_MAX_COLUMNS[];
382+
/// "map.shared-shredding.column-placement-policy" - Suffix for per-column shared-shredding
383+
/// physical column placement policy.
384+
/// Used as `fields.<column>.map.shared-shredding.column-placement-policy`.
385+
/// Values: "plain", "sequential" and "lru". Default value is "plain".
386+
/// Only effective when map.storage-layout = shared-shredding.
387+
static const char MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY[];
382388

383389
/// "blob-as-descriptor" - Read blob field using blob descriptor rather than blob
384390
/// bytes. Default value is "false".

src/paimon/CMakeLists.txt

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -142,6 +142,7 @@ set(PAIMON_COMMON_SRCS
142142
common/data/shredding/map_shared_shredding_context.cpp
143143
common/data/shredding/map_shared_shredding_batch_converter.cpp
144144
common/data/shredding/map_shared_shredding_column_allocator.cpp
145+
common/data/shredding/lru_map_shared_shredding_column_allocator.cpp
145146
common/data/shredding/map_shared_shredding_file_reader.cpp
146147
common/utils/delta_varint_compressor.cpp
147148
common/utils/fields_comparator.cpp
@@ -547,7 +548,9 @@ if(PAIMON_BUILD_TESTS)
547548
common/utils/generic_lru_cache_test.cpp
548549
common/data/shredding/map_shared_shredding_utils_test.cpp
549550
common/data/shredding/map_shared_shredding_batch_converter_test.cpp
550-
common/data/shredding/map_shared_shredding_column_allocator_test.cpp
551+
common/data/shredding/lru_map_shared_shredding_column_allocator_test.cpp
552+
common/data/shredding/plain_map_shared_shredding_column_allocator_test.cpp
553+
common/data/shredding/sequential_map_shared_shredding_column_allocator_test.cpp
551554
common/data/shredding/map_shared_shredding_field_dict_test.cpp
552555
common/data/shredding/map_shared_shredding_context_test.cpp
553556
common/data/shredding/map_shared_shredding_file_reader_test.cpp
Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/common/data/shredding/lru_map_shared_shredding_column_allocator.h"
18+
19+
#include <algorithm>
20+
#include <limits>
21+
22+
namespace paimon {
23+
24+
LruMapSharedShreddingColumnAllocator::LruMapSharedShreddingColumnAllocator(int32_t num_columns)
25+
: MapSharedShreddingColumnAllocator(num_columns),
26+
col_field_(num_columns, -1),
27+
last_used_(num_columns, 0) {}
28+
29+
int32_t LruMapSharedShreddingColumnAllocator::SelectColumn(
30+
const std::vector<int32_t>& candidates,
31+
const std::vector<int32_t>& planned_col_to_field) const {
32+
int32_t selected_col = candidates.front();
33+
int64_t selected_last_used = std::numeric_limits<int64_t>::max();
34+
bool selected_empty = false;
35+
for (int32_t col : candidates) {
36+
bool is_empty = planned_col_to_field[col] == -1;
37+
if (is_empty) {
38+
if (!selected_empty || col < selected_col) {
39+
selected_empty = true;
40+
selected_col = col;
41+
}
42+
continue;
43+
}
44+
if (!selected_empty && (last_used_[col] < selected_last_used ||
45+
(last_used_[col] == selected_last_used && col < selected_col))) {
46+
selected_col = col;
47+
selected_last_used = last_used_[col];
48+
}
49+
}
50+
return selected_col;
51+
}
52+
53+
void LruMapSharedShreddingColumnAllocator::UpdateLastUsed(const RowAllocation& allocation) {
54+
bool touched = false;
55+
for (int32_t col = 0; col < num_columns_; ++col) {
56+
if (allocation.col_to_field[col] != -1) {
57+
last_used_[col] = lru_clock_;
58+
touched = true;
59+
}
60+
}
61+
if (touched) {
62+
++lru_clock_;
63+
}
64+
}
65+
66+
RowAllocation LruMapSharedShreddingColumnAllocator::AllocateRow(
67+
const std::vector<int32_t>& field_ids) {
68+
std::vector<int32_t> sorted_field_ids = field_ids;
69+
std::sort(sorted_field_ids.begin(), sorted_field_ids.end());
70+
71+
RowAllocation allocation;
72+
allocation.col_to_field.assign(num_columns_, -1);
73+
std::vector<int32_t> next_col_to_field = col_field_;
74+
std::vector<bool> used_cols(num_columns_, false);
75+
std::vector<int32_t> unassigned;
76+
77+
for (int32_t field_id : sorted_field_ids) {
78+
auto it = std::find(col_field_.begin(), col_field_.end(), field_id);
79+
if (it != col_field_.end()) {
80+
int32_t col = static_cast<int32_t>(it - col_field_.begin());
81+
used_cols[col] = true;
82+
allocation.col_to_field[col] = field_id;
83+
} else {
84+
unassigned.push_back(field_id);
85+
}
86+
}
87+
88+
for (int32_t field_id : unassigned) {
89+
std::vector<int32_t> candidates;
90+
candidates.reserve(num_columns_);
91+
for (int32_t col = 0; col < num_columns_; ++col) {
92+
if (!used_cols[col]) {
93+
candidates.push_back(col);
94+
}
95+
}
96+
97+
if (candidates.empty()) {
98+
allocation.overflow_fields.push_back(field_id);
99+
continue;
100+
}
101+
102+
int32_t col = SelectColumn(candidates, next_col_to_field);
103+
used_cols[col] = true;
104+
allocation.col_to_field[col] = field_id;
105+
next_col_to_field[col] = field_id;
106+
}
107+
108+
UpdateLastUsed(allocation);
109+
col_field_ = next_col_to_field;
110+
CommitRow(allocation, sorted_field_ids);
111+
return allocation;
112+
}
113+
114+
} // namespace paimon
Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#pragma once
18+
19+
#include <cstdint>
20+
#include <vector>
21+
22+
#include "paimon/common/data/shredding/map_shared_shredding_column_allocator.h"
23+
24+
namespace paimon {
25+
26+
/// Allocator that keeps cross-row column state and evicts least-recently-used columns.
27+
class LruMapSharedShreddingColumnAllocator : public MapSharedShreddingColumnAllocator {
28+
public:
29+
/// @param num_columns Number of available physical columns.
30+
explicit LruMapSharedShreddingColumnAllocator(int32_t num_columns);
31+
32+
RowAllocation AllocateRow(const std::vector<int32_t>& field_ids) override;
33+
34+
private:
35+
int32_t SelectColumn(const std::vector<int32_t>& candidates,
36+
const std::vector<int32_t>& planned_col_to_field) const;
37+
void UpdateLastUsed(const RowAllocation& allocation);
38+
39+
int64_t lru_clock_ = 0;
40+
std::vector<int32_t> col_field_;
41+
std::vector<int64_t> last_used_;
42+
};
43+
44+
} // namespace paimon
Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/common/data/shredding/lru_map_shared_shredding_column_allocator.h"
18+
19+
#include <set>
20+
#include <vector>
21+
22+
#include "gtest/gtest.h"
23+
24+
namespace paimon::test {
25+
namespace {
26+
27+
void ExpectAllocation(const RowAllocation& allocation, const std::vector<int32_t>& col_to_field,
28+
const std::vector<int32_t>& overflow_fields) {
29+
ASSERT_EQ(col_to_field, allocation.col_to_field);
30+
ASSERT_EQ(overflow_fields, allocation.overflow_fields);
31+
}
32+
33+
} // namespace
34+
35+
TEST(LruMapSharedShreddingColumnAllocatorTest, AllocatesWithHitRetainEvictAndOverflow) {
36+
LruMapSharedShreddingColumnAllocator allocator(3);
37+
38+
RowAllocation row0 = allocator.AllocateRow({0, 1, 2});
39+
ExpectAllocation(row0, {0, 1, 2}, {});
40+
41+
RowAllocation row1 = allocator.AllocateRow({0, 1});
42+
ExpectAllocation(row1, {0, 1, -1}, {});
43+
44+
RowAllocation row2 = allocator.AllocateRow({3, 4, 5});
45+
ExpectAllocation(row2, {4, 5, 3}, {});
46+
47+
RowAllocation row3 = allocator.AllocateRow({0, 3, 4, 5});
48+
ExpectAllocation(row3, {4, 5, 3}, {0});
49+
50+
ASSERT_EQ(4, allocator.GetMaxRowWidth());
51+
52+
const auto& field_to_columns = allocator.GetFieldToColumns();
53+
ASSERT_EQ((std::set<int32_t>{0}), field_to_columns.at(0));
54+
ASSERT_EQ((std::set<int32_t>{1}), field_to_columns.at(1));
55+
ASSERT_EQ((std::set<int32_t>{2}), field_to_columns.at(2));
56+
ASSERT_EQ((std::set<int32_t>{2}), field_to_columns.at(3));
57+
ASSERT_EQ((std::set<int32_t>{0}), field_to_columns.at(4));
58+
ASSERT_EQ((std::set<int32_t>{1}), field_to_columns.at(5));
59+
ASSERT_EQ((std::set<int32_t>{0}), allocator.GetOverflowFieldSet());
60+
}
61+
62+
TEST(LruMapSharedShreddingColumnAllocatorTest, HandlesEmptyRows) {
63+
LruMapSharedShreddingColumnAllocator allocator(2);
64+
65+
RowAllocation empty_row = allocator.AllocateRow({});
66+
ExpectAllocation(empty_row, {-1, -1}, {});
67+
ASSERT_EQ(0, allocator.GetMaxRowWidth());
68+
69+
RowAllocation row = allocator.AllocateRow({7});
70+
ExpectAllocation(row, {7, -1}, {});
71+
ASSERT_EQ(1, allocator.GetMaxRowWidth());
72+
}
73+
74+
} // namespace paimon::test

0 commit comments

Comments
 (0)