Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions include/paimon/defs.h
Original file line number Diff line number Diff line change
Expand Up @@ -379,6 +379,12 @@ struct PAIMON_EXPORT Options {
/// map.storage-layout = shared-shredding. Rows with more fields than K_max spill to
/// __overflow. Default value is 256. Each column can have its own max-columns setting.
static const char MAP_SHARED_SHREDDING_MAX_COLUMNS[];
/// "map.shared-shredding.column-placement-policy" - Suffix for per-column shared-shredding
/// physical column placement policy.
/// Used as `fields.<column>.map.shared-shredding.column-placement-policy`.
/// Values: "plain", "sequential" and "lru". Default value is "lru".
/// Only effective when map.storage-layout = shared-shredding.
static const char MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY[];

/// "blob-as-descriptor" - Read blob field using blob descriptor rather than blob
/// bytes. Default value is "false".
Expand Down
5 changes: 4 additions & 1 deletion src/paimon/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,7 @@ set(PAIMON_COMMON_SRCS
common/data/shredding/map_shared_shredding_context.cpp
common/data/shredding/map_shared_shredding_batch_converter.cpp
common/data/shredding/map_shared_shredding_column_allocator.cpp
common/data/shredding/lru_map_shared_shredding_column_allocator.cpp
common/data/shredding/map_shared_shredding_file_reader.cpp
common/utils/delta_varint_compressor.cpp
common/utils/fields_comparator.cpp
Expand Down Expand Up @@ -547,7 +548,9 @@ if(PAIMON_BUILD_TESTS)
common/utils/generic_lru_cache_test.cpp
common/data/shredding/map_shared_shredding_utils_test.cpp
common/data/shredding/map_shared_shredding_batch_converter_test.cpp
common/data/shredding/map_shared_shredding_column_allocator_test.cpp
common/data/shredding/lru_map_shared_shredding_column_allocator_test.cpp
common/data/shredding/plain_map_shared_shredding_column_allocator_test.cpp
common/data/shredding/sequential_map_shared_shredding_column_allocator_test.cpp
common/data/shredding/map_shared_shredding_field_dict_test.cpp
common/data/shredding/map_shared_shredding_context_test.cpp
common/data/shredding/map_shared_shredding_file_reader_test.cpp
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
/*
* Copyright 2026-present Alibaba Inc.
*
* Licensed 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/common/data/shredding/lru_map_shared_shredding_column_allocator.h"

#include <algorithm>
#include <limits>

namespace paimon {

LruMapSharedShreddingColumnAllocator::LruMapSharedShreddingColumnAllocator(int32_t num_columns)
: MapSharedShreddingColumnAllocator(num_columns),
col_field_(num_columns, -1),
last_used_(num_columns, 0) {}

int32_t LruMapSharedShreddingColumnAllocator::SelectColumn(
const std::vector<int32_t>& candidates,
const std::vector<int32_t>& planned_col_to_field) const {
int32_t selected_col = candidates.front();
int64_t selected_last_used = std::numeric_limits<int64_t>::max();
bool selected_empty = false;
for (int32_t col : candidates) {
bool is_empty = planned_col_to_field[col] == -1;
if (is_empty) {
if (!selected_empty || col < selected_col) {
selected_empty = true;
selected_col = col;

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’m not quite following the purpose of the defensive check col < selected_col here. At the current call site, candidates is constructed in ascending column order, so could we just return as soon as we find a suitable empty one?

}
continue;
}
if (!selected_empty && (last_used_[col] < selected_last_used ||
(last_used_[col] == selected_last_used && col < selected_col))) {
selected_col = col;
selected_last_used = last_used_[col];
}

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add comments for the strategy.

}
return selected_col;
}

void LruMapSharedShreddingColumnAllocator::UpdateLastUsed(const RowAllocation& allocation) {
bool touched = false;
for (int32_t col = 0; col < num_columns_; ++col) {
if (allocation.col_to_field[col] != -1) {
last_used_[col] = lru_clock_;
touched = true;
}
}
if (touched) {
++lru_clock_;
}
}

RowAllocation LruMapSharedShreddingColumnAllocator::AllocateRow(
const std::vector<int32_t>& field_ids) {
std::vector<int32_t> sorted_field_ids = field_ids;
std::sort(sorted_field_ids.begin(), sorted_field_ids.end());

RowAllocation allocation;
allocation.col_to_field.assign(num_columns_, -1);
std::vector<int32_t> next_col_to_field = col_field_;
std::vector<bool> used_cols(num_columns_, false);
std::vector<int32_t> unassigned;

for (int32_t field_id : sorted_field_ids) {
auto it = std::find(col_field_.begin(), col_field_.end(), field_id);
if (it != col_field_.end()) {
int32_t col = static_cast<int32_t>(it - col_field_.begin());
used_cols[col] = true;
allocation.col_to_field[col] = field_id;
} else {
unassigned.push_back(field_id);
}
}

for (int32_t field_id : unassigned) {
std::vector<int32_t> candidates;
candidates.reserve(num_columns_);
for (int32_t col = 0; col < num_columns_; ++col) {
if (!used_cols[col]) {
candidates.push_back(col);
}
}

if (candidates.empty()) {
allocation.overflow_fields.push_back(field_id);
continue;
}

int32_t col = SelectColumn(candidates, next_col_to_field);
used_cols[col] = true;
allocation.col_to_field[col] = field_id;
next_col_to_field[col] = field_id;
}

UpdateLastUsed(allocation);
col_field_ = next_col_to_field;
CommitRow(allocation, sorted_field_ids);
return allocation;
}

} // namespace paimon
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/*
* Copyright 2026-present Alibaba Inc.
*
* Licensed 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 <cstdint>
#include <vector>

#include "paimon/common/data/shredding/map_shared_shredding_column_allocator.h"

namespace paimon {

/// Allocator that keeps cross-row column state and evicts least-recently-used columns.
class LruMapSharedShreddingColumnAllocator : public MapSharedShreddingColumnAllocator {
public:
/// @param num_columns Number of available physical columns.
explicit LruMapSharedShreddingColumnAllocator(int32_t num_columns);

RowAllocation AllocateRow(const std::vector<int32_t>& field_ids) override;

private:
int32_t SelectColumn(const std::vector<int32_t>& candidates,
const std::vector<int32_t>& planned_col_to_field) const;
void UpdateLastUsed(const RowAllocation& allocation);

int64_t lru_clock_ = 0;
std::vector<int32_t> col_field_;
std::vector<int64_t> last_used_;
};

} // namespace paimon
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
/*
* Copyright 2026-present Alibaba Inc.
*
* Licensed 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/common/data/shredding/lru_map_shared_shredding_column_allocator.h"

#include <set>
#include <vector>

#include "gtest/gtest.h"

namespace paimon::test {
namespace {

void ExpectAllocation(const RowAllocation& allocation, const std::vector<int32_t>& col_to_field,
const std::vector<int32_t>& overflow_fields) {
ASSERT_EQ(col_to_field, allocation.col_to_field);
ASSERT_EQ(overflow_fields, allocation.overflow_fields);
}

} // namespace

TEST(LruMapSharedShreddingColumnAllocatorTest, AllocatesWithHitRetainEvictAndOverflow) {
LruMapSharedShreddingColumnAllocator allocator(3);

RowAllocation row0 = allocator.AllocateRow({0, 1, 2});
ExpectAllocation(row0, {0, 1, 2}, {});

RowAllocation row1 = allocator.AllocateRow({0, 1});
ExpectAllocation(row1, {0, 1, -1}, {});

RowAllocation row2 = allocator.AllocateRow({3, 4, 5});
ExpectAllocation(row2, {4, 5, 3}, {});

RowAllocation row3 = allocator.AllocateRow({0, 3, 4, 5});
ExpectAllocation(row3, {4, 5, 3}, {0});

ASSERT_EQ(4, allocator.GetMaxRowWidth());

const auto& field_to_columns = allocator.GetFieldToColumns();
ASSERT_EQ((std::set<int32_t>{0}), field_to_columns.at(0));
ASSERT_EQ((std::set<int32_t>{1}), field_to_columns.at(1));
ASSERT_EQ((std::set<int32_t>{2}), field_to_columns.at(2));
ASSERT_EQ((std::set<int32_t>{2}), field_to_columns.at(3));
ASSERT_EQ((std::set<int32_t>{0}), field_to_columns.at(4));
ASSERT_EQ((std::set<int32_t>{1}), field_to_columns.at(5));
ASSERT_EQ((std::set<int32_t>{0}), allocator.GetOverflowFieldSet());
}

TEST(LruMapSharedShreddingColumnAllocatorTest, HandlesEmptyRows) {
LruMapSharedShreddingColumnAllocator allocator(2);

RowAllocation empty_row = allocator.AllocateRow({});
ExpectAllocation(empty_row, {-1, -1}, {});
ASSERT_EQ(0, allocator.GetMaxRowWidth());

RowAllocation row = allocator.AllocateRow({7});
ExpectAllocation(row, {7, -1}, {});
ASSERT_EQ(1, allocator.GetMaxRowWidth());
}

} // namespace paimon::test
Loading
Loading