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,107 @@
/*
* 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();
for (int32_t col : candidates) {
if (planned_col_to_field[col] == -1) {
return col;
}
if (last_used_[col] < selected_last_used) {
selected_col = col;
selected_last_used = last_used_[col];
}
Comment thread
zjw1111 marked this conversation as resolved.
}
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,49 @@
/*
* 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 current column assignments across rows and evicts least-recently-used
/// columns.
///
/// Keys that are still assigned to physical columns keep those columns when they appear again.
/// New keys, including keys that were previously evicted, use empty columns first; if no empty
/// column is available, the allocator replaces the least-recently-used physical column.
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