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

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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/map_shared_shredding_context.h"

#include <algorithm>
#include <cmath>

namespace paimon {

MapSharedShreddingContext::MapSharedShreddingContext(
    const std::map<std::string, int32_t>& column_to_k_max)
    : column_to_k_max_(column_to_k_max) {}

std::map<std::string, int32_t> MapSharedShreddingContext::ComputeNextK() const {
    std::map<std::string, int32_t> result;
    for (const auto& [field_name, k_max] : column_to_k_max_) {
        auto it = recent_max_row_widths_.find(field_name);
        if (it == recent_max_row_widths_.end() || it->second.empty()) {
            // First file — no history, use K_max.
            result[field_name] = k_max;
        } else {
            int32_t adaptive_width = ComputeAdaptiveWidth(it->second);
            result[field_name] = std::max(1, std::min(adaptive_width, k_max));
        }
    }
    return result;
}

void MapSharedShreddingContext::ReportFileStats(const std::string& field_name,
                                                int32_t max_row_width) {
    auto& window = recent_max_row_widths_[field_name];
    window.push_back(max_row_width);
    if (static_cast<int32_t>(window.size()) > kWindowSize) {
        window.erase(window.begin());
    }
}

std::vector<std::string> MapSharedShreddingContext::GetShreddingColumnNames() const {
    std::vector<std::string> names;
    names.reserve(column_to_k_max_.size());
    for (const auto& [field_name, _] : column_to_k_max_) {
        names.push_back(field_name);
    }
    return names;
}

int32_t MapSharedShreddingContext::ComputeAdaptiveWidth(const std::vector<int32_t>& values) {
    if (values.empty()) {
        return 0;
    }

    std::vector<int32_t> sorted_values(values.begin(), values.end());
    std::sort(sorted_values.begin(), sorted_values.end());

    int32_t max_width = sorted_values.back();
    auto percentile_rank = static_cast<int64_t>(std::ceil(kPercentileRatio * sorted_values.size()));
    percentile_rank = std::clamp<int64_t>(percentile_rank, 1, sorted_values.size());
    int32_t percentile_width = sorted_values[percentile_rank - 1];

    // Use P90 to ignore far outliers, but keep max when it is close enough to normal rows.
    auto relative_close_threshold = static_cast<int32_t>(
        std::ceil(static_cast<double>(percentile_width) * kMaxCloseRelativeRatio));
    if (max_width - percentile_width <= kMaxCloseAbsoluteSlack ||
        max_width <= relative_close_threshold) {
        return max_width;
    }
    return percentile_width;
}

}  // namespace paimon