feat(shredding): add shared-shredding map placement policies (alibaba#384)

Author: zjw1111

include/paimon/defs.h

@@ -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".

src/paimon/CMakeLists.txt

@@ -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
@@ -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

src/paimon/common/data/shredding/lru_map_shared_shredding_column_allocator.cpp

@@ -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];
+        }
+    }
+    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

src/paimon/common/data/shredding/lru_map_shared_shredding_column_allocator.h

@@ -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

src/paimon/common/data/shredding/lru_map_shared_shredding_column_allocator_test.cpp

@@ -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