// 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 "storage/merger.h"

#include <gen_cpp/olap_file.pb.h>

#include <gen_cpp/types.pb.h>

#include <stddef.h>

#include <unistd.h>

#include <algorithm>

#include <iterator>

#include <memory>

#include <mutex>

#include <numeric>

#include <ostream>

#include <shared_mutex>

#include <string>

#include <unordered_map>

#include <utility>

#include <vector>

#include "cloud/config.h"

#include "common/config.h"

#include "common/logging.h"

#include "common/status.h"

#include "core/block/block.h"

#include "storage/iterator/block_reader.h"

#include "storage/iterator/vertical_block_reader.h"

#include "storage/iterator/vertical_merge_iterator.h"

#include "storage/iterators.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/rowid_conversion.h"

#include "storage/rowset/beta_rowset.h"

#include "storage/rowset/rowset.h"

#include "storage/rowset/rowset_meta.h"

#include "storage/rowset/rowset_writer.h"

#include "storage/segment/segment.h"

#include "storage/segment/segment_writer.h"

#include "storage/storage_engine.h"

#include "storage/tablet/base_tablet.h"

#include "storage/tablet/tablet.h"

#include "storage/tablet/tablet_fwd.h"

#include "storage/tablet/tablet_meta.h"

#include "storage/tablet/tablet_reader.h"

#include "storage/types.h"

#include "storage/utils.h"

#include "util/defer_op.h"

#include "util/slice.h"

namespace doris {

Status Merger::vmerge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type,

                              const TabletSchema& cur_tablet_schema,

                              const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,

                              RowsetWriter* dst_rowset_writer, Statistics* stats_output) {

    if (!cur_tablet_schema.cluster_key_uids().empty()) {

        return Status::InternalError(

                "mow table with cluster keys does not support non vertical compaction");

    }

    BlockReader reader;

    TabletReader::ReaderParams reader_params;

    reader_params.tablet = tablet;

    reader_params.reader_type = reader_type;

    reader_params.read_row_binlog = tablet->is_row_binlog_tablet();

    TabletReadSource read_source;

    read_source.rs_splits.reserve(src_rowset_readers.size());

    for (const RowsetReaderSharedPtr& rs_reader : src_rowset_readers) {

        read_source.rs_splits.emplace_back(rs_reader);

    }

    read_source.fill_delete_predicates();

    reader_params.set_read_source(std::move(read_source));

    reader_params.version = dst_rowset_writer->version();

    TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>();

    merge_tablet_schema->copy_from(cur_tablet_schema);

    // Merge the columns in delete predicate that not in latest schema in to current tablet schema

    for (auto& del_pred_rs : reader_params.delete_predicates) {

        merge_tablet_schema->merge_dropped_columns(*del_pred_rs->tablet_schema());

    }

    reader_params.tablet_schema = merge_tablet_schema;

    if (!tablet->tablet_schema()->cluster_key_uids().empty()) {

        reader_params.delete_bitmap = tablet->tablet_meta()->delete_bitmap_ptr();

    }

    if (reader_params.read_row_binlog) {

        reader_params.delete_bitmap = tablet->tablet_meta()->delete_bitmap_ptr();

    }

    if (stats_output && stats_output->rowid_conversion) {

        reader_params.record_rowids = true;

        reader_params.rowid_conversion = stats_output->rowid_conversion;

        stats_output->rowid_conversion->set_dst_rowset_id(dst_rowset_writer->rowset_id());

    }

    reader_params.return_columns.resize(cur_tablet_schema.num_columns());

    std::iota(reader_params.return_columns.begin(), reader_params.return_columns.end(), 0);

    reader_params.origin_return_columns = &reader_params.return_columns;

    RETURN_IF_ERROR(reader.init(reader_params));

    Block block = cur_tablet_schema.create_block(reader_params.return_columns);

    size_t output_rows = 0;

    bool eof = false;

    while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {

        auto tablet_state = tablet->tablet_state();

        if (tablet_state != TABLET_RUNNING && tablet_state != TABLET_NOTREADY) {

            tablet->clear_cache();

            return Status::Error<INTERNAL_ERROR>("tablet {} is not used any more",

                                                 tablet->tablet_id());

        }

        // Read one block from block reader

        RETURN_NOT_OK_STATUS_WITH_WARN(reader.next_block_with_aggregation(&block, &eof),

                                       "failed to read next block when merging rowsets of tablet " +

                                               std::to_string(tablet->tablet_id()));

        RETURN_NOT_OK_STATUS_WITH_WARN(dst_rowset_writer->add_block(&block),

                                       "failed to write block when merging rowsets of tablet " +

                                               std::to_string(tablet->tablet_id()));

        if (reader_params.record_rowids && block.rows() > 0) {

            std::vector<uint32_t> segment_num_rows;

            RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));

            stats_output->rowid_conversion->add(reader.current_block_row_locations(),

                                                segment_num_rows);

        }

        output_rows += block.rows();

        block.clear_column_data();

    }

    if (ExecEnv::GetInstance()->storage_engine().stopped()) {

        return Status::Error<INTERNAL_ERROR>("tablet {} failed to do compaction, engine stopped",

                                             tablet->tablet_id());

    }

    if (stats_output != nullptr) {

        stats_output->output_rows = output_rows;

        stats_output->merged_rows = reader.merged_rows();

        stats_output->filtered_rows = reader.filtered_rows();

        stats_output->bytes_read_from_local = reader.stats().file_cache_stats.bytes_read_from_local;

        stats_output->bytes_read_from_remote =

                reader.stats().file_cache_stats.bytes_read_from_remote;

        stats_output->cached_bytes_total = reader.stats().file_cache_stats.bytes_write_into_cache;

        if (config::is_cloud_mode()) {

            stats_output->cloud_local_read_time =

                    reader.stats().file_cache_stats.local_io_timer / 1000;

            stats_output->cloud_remote_read_time =

                    reader.stats().file_cache_stats.remote_io_timer / 1000;

        }

    }

    RETURN_NOT_OK_STATUS_WITH_WARN(dst_rowset_writer->flush(),

                                   "failed to flush rowset when merging rowsets of tablet " +

                                           std::to_string(tablet->tablet_id()));

    return Status::OK();

}

// split columns into several groups, make sure all keys in one group

// unique_key should consider sequence&delete column

void Merger::vertical_split_columns(const TabletSchema& tablet_schema,

                                    std::vector<std::vector<uint32_t>>* column_groups,

                                    std::vector<uint32_t>* key_group_cluster_key_idxes,

                                    int32_t num_columns_per_group) {

    size_t num_key_cols = tablet_schema.num_key_columns();

    size_t total_cols = tablet_schema.num_columns();

    std::vector<uint32_t> key_columns;

    for (auto i = 0; i < num_key_cols; ++i) {

        key_columns.emplace_back(i);

    }

    // in unique key, sequence & delete sign column should merge with key columns

    int32_t sequence_col_idx = -1;

    int32_t delete_sign_idx = -1;

    // in key column compaction, seq_col real index is _num_key_columns

    // and delete_sign column is _block->columns() - 1

    if (tablet_schema.keys_type() == KeysType::UNIQUE_KEYS) {

        if (tablet_schema.has_sequence_col()) {

            sequence_col_idx = tablet_schema.sequence_col_idx();

            key_columns.emplace_back(sequence_col_idx);

        }

        delete_sign_idx = tablet_schema.field_index(DELETE_SIGN);

        if (delete_sign_idx != -1) {

            key_columns.emplace_back(delete_sign_idx);

        }

        if (!tablet_schema.cluster_key_uids().empty()) {

            for (const auto& cid : tablet_schema.cluster_key_uids()) {

                auto idx = tablet_schema.field_index(cid);

                DCHECK(idx >= 0) << "could not find cluster key column with unique_id=" << cid

                                 << " in tablet schema, table_id=" << tablet_schema.table_id();

                if (idx >= num_key_cols) {

                    key_columns.emplace_back(idx);

                }

            }

            // tablet schema unique ids: [1, 2, 5, 3, 6, 4], [1 2] is key columns

            // cluster key unique ids: [3, 1, 4]

            // the key_columns should be [0, 1, 3, 5]

            // the key_group_cluster_key_idxes should be [2, 1, 3]

            for (const auto& cid : tablet_schema.cluster_key_uids()) {

                auto idx = tablet_schema.field_index(cid);

                for (auto i = 0; i < key_columns.size(); ++i) {

                    if (idx == key_columns[i]) {

                        key_group_cluster_key_idxes->emplace_back(i);

                        break;

                    }

                }

            }

        }

    }

    VLOG_NOTICE << "sequence_col_idx=" << sequence_col_idx

                << ", delete_sign_idx=" << delete_sign_idx;

    // for duplicate no keys

    if (!key_columns.empty()) {

        column_groups->emplace_back(key_columns);

    }

    std::vector<uint32_t> value_columns;

    for (size_t i = num_key_cols; i < total_cols; ++i) {

        if (i == sequence_col_idx || i == delete_sign_idx ||

            key_columns.end() != std::find(key_columns.begin(), key_columns.end(), i)) {

            continue;

        }

        if (!value_columns.empty() && value_columns.size() % num_columns_per_group == 0) {

            column_groups->push_back(value_columns);

            value_columns.clear();

        }

        value_columns.push_back(cast_set<uint32_t>(i));

    }

    if (!value_columns.empty()) {

        column_groups->push_back(value_columns);

    }

}

Status Merger::vertical_compact_one_group(

        BaseTabletSPtr tablet, ReaderType reader_type, const TabletSchema& tablet_schema,

        bool is_key, const std::vector<uint32_t>& column_group, RowSourcesBuffer* row_source_buf,

        const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,

        RowsetWriter* dst_rowset_writer, uint32_t max_rows_per_segment, Statistics* stats_output,

        std::vector<uint32_t> key_group_cluster_key_idxes, int64_t batch_size,

        CompactionSampleInfo* sample_info, VerticalCompactionContextStats* context_stats,

        bool enable_sparse_optimization) {

    // build tablet reader

    VLOG_NOTICE << "vertical compact one group, max_rows_per_segment=" << max_rows_per_segment;

    VerticalBlockReader reader(row_source_buf, context_stats);

    TabletReader::ReaderParams reader_params;

    reader_params.is_key_column_group = is_key;

    reader_params.key_group_cluster_key_idxes = key_group_cluster_key_idxes;

    reader_params.tablet = tablet;

    reader_params.reader_type = reader_type;

    reader_params.read_row_binlog = tablet->is_row_binlog_tablet();

    reader_params.enable_sparse_optimization = enable_sparse_optimization;

    TabletReadSource read_source;

    read_source.rs_splits.reserve(src_rowset_readers.size());

    for (const RowsetReaderSharedPtr& rs_reader : src_rowset_readers) {

        read_source.rs_splits.emplace_back(rs_reader);

    }

    read_source.fill_delete_predicates();

    reader_params.set_read_source(std::move(read_source));

    reader_params.version = dst_rowset_writer->version();

    TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>();

    merge_tablet_schema->copy_from(tablet_schema);

    for (auto& del_pred_rs : reader_params.delete_predicates) {

        merge_tablet_schema->merge_dropped_columns(*del_pred_rs->tablet_schema());

    }

    reader_params.tablet_schema = merge_tablet_schema;

    bool has_cluster_key = false;

    if (!tablet->tablet_schema()->cluster_key_uids().empty()) {

        reader_params.delete_bitmap = tablet->tablet_meta()->delete_bitmap_ptr();

        has_cluster_key = true;

    }

    if (reader_params.read_row_binlog) {

        reader_params.delete_bitmap = tablet->tablet_meta()->delete_bitmap_ptr();

    }

    if (is_key && stats_output && stats_output->rowid_conversion) {

        reader_params.record_rowids = true;

        reader_params.rowid_conversion = stats_output->rowid_conversion;

        stats_output->rowid_conversion->set_dst_rowset_id(dst_rowset_writer->rowset_id());

    }

    reader_params.return_columns = column_group;

    reader_params.origin_return_columns = &reader_params.return_columns;

    reader_params.batch_size = batch_size;

    RETURN_IF_ERROR(reader.init(reader_params, sample_info));

    Block block = tablet_schema.create_block(reader_params.return_columns);

    size_t output_rows = 0;

    bool eof = false;

    while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {

        auto tablet_state = tablet->tablet_state();

        if (tablet_state != TABLET_RUNNING && tablet_state != TABLET_NOTREADY) {

            tablet->clear_cache();

            return Status::Error<INTERNAL_ERROR>("tablet {} is not used any more",

                                                 tablet->tablet_id());

        }

        // Read one block from block reader

        RETURN_NOT_OK_STATUS_WITH_WARN(reader.next_block_with_aggregation(&block, &eof),

                                       "failed to read next block when merging rowsets of tablet " +

                                               std::to_string(tablet->tablet_id()));

        RETURN_NOT_OK_STATUS_WITH_WARN(

                dst_rowset_writer->add_columns(&block, column_group, is_key, max_rows_per_segment,

                                               has_cluster_key),

                "failed to write block when merging rowsets of tablet " +

                        std::to_string(tablet->tablet_id()));

        if (is_key && reader_params.record_rowids && block.rows() > 0) {

            std::vector<uint32_t> segment_num_rows;

            RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows));

            stats_output->rowid_conversion->add(reader.current_block_row_locations(),

                                                segment_num_rows);

        }

        output_rows += block.rows();

        block.clear_column_data();

    }

    if (ExecEnv::GetInstance()->storage_engine().stopped()) {

        return Status::Error<INTERNAL_ERROR>("tablet {} failed to do compaction, engine stopped",

                                             tablet->tablet_id());

    }

    if (stats_output != nullptr) {

        if (is_key) {

            stats_output->output_rows = output_rows;

            stats_output->merged_rows = reader.merged_rows();

            stats_output->filtered_rows = reader.filtered_rows();

        }

        stats_output->bytes_read_from_local = reader.stats().file_cache_stats.bytes_read_from_local;

        stats_output->bytes_read_from_remote =

                reader.stats().file_cache_stats.bytes_read_from_remote;

        stats_output->cached_bytes_total = reader.stats().file_cache_stats.bytes_write_into_cache;

        if (config::is_cloud_mode()) {

            stats_output->cloud_local_read_time =

                    reader.stats().file_cache_stats.local_io_timer / 1000;

            stats_output->cloud_remote_read_time =

                    reader.stats().file_cache_stats.remote_io_timer / 1000;

        }

    }

    RETURN_IF_ERROR(dst_rowset_writer->flush_columns(is_key));

    return Status::OK();

}

// for segcompaction

Status Merger::vertical_compact_one_group(

        int64_t tablet_id, ReaderType reader_type, const TabletSchema& tablet_schema, bool is_key,

        const std::vector<uint32_t>& column_group, RowSourcesBuffer* row_source_buf,

        VerticalBlockReader& src_block_reader, segment_v2::SegmentWriter& dst_segment_writer,

        Statistics* stats_output, uint64_t* index_size, KeyBoundsPB& key_bounds,

        SimpleRowIdConversion* rowid_conversion) {

    // TODO: record_rowids

    Block block = tablet_schema.create_block(column_group);

    size_t output_rows = 0;

    bool eof = false;

    while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) {

        // Read one block from block reader

        RETURN_NOT_OK_STATUS_WITH_WARN(src_block_reader.next_block_with_aggregation(&block, &eof),

                                       "failed to read next block when merging rowsets of tablet " +

                                               std::to_string(tablet_id));

        if (!block.rows()) {

            break;

        }

        RETURN_NOT_OK_STATUS_WITH_WARN(dst_segment_writer.append_block(&block, 0, block.rows()),

                                       "failed to write block when merging rowsets of tablet " +

                                               std::to_string(tablet_id));

        if (is_key && rowid_conversion != nullptr) {

            rowid_conversion->add(src_block_reader.current_block_row_locations());

        }

        output_rows += block.rows();

        block.clear_column_data();

    }

    if (ExecEnv::GetInstance()->storage_engine().stopped()) {

        return Status::Error<INTERNAL_ERROR>("tablet {} failed to do compaction, engine stopped",

                                             tablet_id);

    }

    if (stats_output != nullptr) {

        if (is_key) {

            stats_output->output_rows = output_rows;

            stats_output->merged_rows = src_block_reader.merged_rows();

            stats_output->filtered_rows = src_block_reader.filtered_rows();

        }

        stats_output->bytes_read_from_local =

                src_block_reader.stats().file_cache_stats.bytes_read_from_local;

        stats_output->bytes_read_from_remote =

                src_block_reader.stats().file_cache_stats.bytes_read_from_remote;

        stats_output->cached_bytes_total =

                src_block_reader.stats().file_cache_stats.bytes_write_into_cache;

    }

    // segcompaction produce only one segment at once

    RETURN_IF_ERROR(dst_segment_writer.finalize_columns_data());

    RETURN_IF_ERROR(dst_segment_writer.finalize_columns_index(index_size));

    if (is_key) {

        Slice min_key = dst_segment_writer.min_encoded_key();

        Slice max_key = dst_segment_writer.max_encoded_key();

        DCHECK_LE(min_key.compare(max_key), 0);

        key_bounds.set_min_key(min_key.to_string());

        key_bounds.set_max_key(max_key.to_string());

    }

    return Status::OK();

}

int64_t estimate_batch_size(int group_index, BaseTabletSPtr tablet, int64_t way_cnt,

                            ReaderType reader_type, int64_t group_per_row_from_footer,

                            bool footer_fallback) {

    auto& sample_info_lock = tablet->get_sample_info_lock(reader_type);

    auto& sample_infos = tablet->get_sample_infos(reader_type);

    std::unique_lock<std::mutex> lock(sample_info_lock);

    CompactionSampleInfo info = sample_infos[group_index];

    if (way_cnt <= 0) {

        LOG(INFO) << "estimate batch size for vertical compaction, tablet id: "

                  << tablet->tablet_id() << " way cnt: " << way_cnt;

        return 4096 - 32;

    }

    int64_t block_mem_limit = config::compaction_memory_bytes_limit / way_cnt;

    if (tablet->last_compaction_status.is<ErrorCode::MEM_LIMIT_EXCEEDED>()) {

        block_mem_limit /= 4;

    }

    int64_t group_data_size = 0;

    if (info.group_data_size > 0 && info.bytes > 0 && info.rows > 0) {

        double smoothing_factor = 0.5;

        group_data_size =

                int64_t((cast_set<double>(info.group_data_size) * (1 - smoothing_factor)) +

                        (cast_set<double>(info.bytes / info.rows) * smoothing_factor));

        sample_infos[group_index].group_data_size = group_data_size;

    } else if (info.group_data_size > 0 && (info.bytes <= 0 || info.rows <= 0)) {

        group_data_size = info.group_data_size;

    } else if (info.group_data_size <= 0 && info.bytes > 0 && info.rows > 0) {

        group_data_size = info.bytes / info.rows;

        sample_infos[group_index].group_data_size = group_data_size;

    } else {

        // No historical sampling data available.

        // Try to use raw_data_bytes from segment footer for a better estimate.

        if (!footer_fallback && group_per_row_from_footer > 0) {

            int64_t batch_size = block_mem_limit / group_per_row_from_footer;

            int64_t res = std::max(std::min(batch_size, int64_t(4096 - 32)), int64_t(32L));

            LOG(INFO) << "estimate batch size from footer for vertical compaction, tablet id: "

                      << tablet->tablet_id()

                      << " group_per_row_from_footer: " << group_per_row_from_footer

                      << " way cnt: " << way_cnt << " batch size: " << res;

            return res;

        }

        LOG(INFO) << "estimate batch size for vertical compaction, tablet id: "

                  << tablet->tablet_id() << " group data size: " << info.group_data_size

                  << " row num: " << info.rows << " consume bytes: " << info.bytes

                  << " footer_fallback: " << footer_fallback;

        return 1024 - 32;

    }

    if (group_data_size <= 0) {

        LOG(WARNING) << "estimate batch size for vertical compaction, tablet id: "

                     << tablet->tablet_id() << " unexpected group data size: " << group_data_size;

        return 4096 - 32;

    }

    sample_infos[group_index].bytes = 0;

    sample_infos[group_index].rows = 0;

    int64_t batch_size = block_mem_limit / group_data_size;

    int64_t res = std::max(std::min(batch_size, int64_t(4096 - 32)), int64_t(32L));

    LOG(INFO) << "estimate batch size for vertical compaction, tablet id: " << tablet->tablet_id()

              << " group data size: " << info.group_data_size << " row num: " << info.rows

              << " consume bytes: " << info.bytes << " way cnt: " << way_cnt

              << " batch size: " << res;

    return res;

}

// steps to do vertical merge:

// 1. split columns into column groups

// 2. compact groups one by one, generate a row_source_buf when compact key group

// and use this row_source_buf to compact value column groups

// 3. build output rowset

Status Merger::vertical_merge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type,

                                      const TabletSchema& tablet_schema,

                                      const std::vector<RowsetReaderSharedPtr>& src_rowset_readers,

                                      RowsetWriter* dst_rowset_writer,

                                      uint32_t max_rows_per_segment, int64_t merge_way_num,

                                      Statistics* stats_output,

                                      VerticalCompactionProgressCallback progress_cb) {

    LOG(INFO) << "Start to do vertical compaction, tablet_id: " << tablet->tablet_id();

    VerticalCompactionContextStats context_stats;

    Defer log_context_stats {[&] {

        DCHECK_EQ(context_stats.active_segment_contexts, 0);

        LOG(INFO) << "Vertical compaction segment context statistics, tablet_id: "

                  << tablet->tablet_id() << ", vertical_compaction_active_segment_contexts_peak: "

                  << context_stats.active_segment_contexts_peak;

    }};

    std::vector<std::vector<uint32_t>> column_groups;

    std::vector<uint32_t> key_group_cluster_key_idxes;

    // If BE config vertical_compaction_num_columns_per_group has been modified from

    // its default value (5), use the BE config; otherwise use the tablet meta value.

    constexpr int32_t default_num_columns_per_group = 5;

    int32_t num_columns_per_group =

            config::vertical_compaction_num_columns_per_group != default_num_columns_per_group

                    ? config::vertical_compaction_num_columns_per_group

                    : tablet->tablet_meta()->vertical_compaction_num_columns_per_group();

    DBUG_EXECUTE_IF("Merger.vertical_merge_rowsets.check_num_columns_per_group", {

        auto expected_value = DebugPoints::instance()->get_debug_param_or_default<int32_t>(

                "Merger.vertical_merge_rowsets.check_num_columns_per_group", "expected_value", -1);

        auto expected_tablet_id = DebugPoints::instance()->get_debug_param_or_default<int64_t>(

                "Merger.vertical_merge_rowsets.check_num_columns_per_group", "tablet_id", -1);

        if (expected_tablet_id != -1 && expected_tablet_id == tablet->tablet_id()) {

            if (expected_value != -1 && expected_value != num_columns_per_group) {

                LOG(FATAL) << "DEBUG_POINT CHECK FAILED: expected num_columns_per_group="

                           << expected_value << " but got " << num_columns_per_group

                           << " for tablet_id=" << tablet->tablet_id();

            } else {

                LOG(INFO) << "DEBUG_POINT CHECK PASSED: num_columns_per_group="

                          << num_columns_per_group << ", tablet_id=" << tablet->tablet_id();

            }

        }

    });

    vertical_split_columns(tablet_schema, &column_groups, &key_group_cluster_key_idxes,

                           num_columns_per_group);

    if (progress_cb) {

        progress_cb(column_groups.size(), 0);

    }

    // Calculate total rows for density calculation after compaction

    int64_t total_rows = 0;

    for (const auto& rs_reader : src_rowset_readers) {

        total_rows += rs_reader->rowset()->rowset_meta()->num_rows();

    }

    // Use historical density for sparse wide table optimization

    // density = (total_cells - null_cells) / total_cells, smaller means more sparse

    // When density <= threshold, enable sparse optimization

    // threshold = 0 means disable, 1 means always enable (default)

    bool enable_sparse_optimization = false;

    if (config::sparse_column_compaction_threshold_percent > 0 &&

        tablet->keys_type() == KeysType::UNIQUE_KEYS) {

        double density = tablet->compaction_density.load();

        enable_sparse_optimization = density <= config::sparse_column_compaction_threshold_percent;

        LOG(INFO) << "Vertical compaction sparse optimization check: tablet_id="

                  << tablet->tablet_id() << ", density=" << density

                  << ", threshold=" << config::sparse_column_compaction_threshold_percent

                  << ", total_rows=" << total_rows

                  << ", num_columns=" << tablet_schema.num_columns()

                  << ", total_cells=" << total_rows * tablet_schema.num_columns()

                  << ", enable_sparse_optimization=" << enable_sparse_optimization;

    }

    RowSourcesBuffer row_sources_buf(tablet->tablet_id(), dst_rowset_writer->context().tablet_path,

                                     reader_type);

    Merger::Statistics total_stats;

    if (stats_output != nullptr) {

        total_stats.rowid_conversion = stats_output->rowid_conversion;

    }

    auto& sample_info_lock = tablet->get_sample_info_lock(reader_type);

    auto& sample_infos = tablet->get_sample_infos(reader_type);

    {

        std::unique_lock<std::mutex> lock(sample_info_lock);

        sample_infos.resize(column_groups.size());

    }

    // Collect per-column raw_data_bytes from segment footer for first-time batch size estimation.

    // raw_data_bytes is the original data size before encoding, close to runtime Block::bytes().

    // Only collect when needed: skip if manual batch_size override is set, or if ALL groups

    // already have historical sampling data. Use per-group granularity so that schema evolution

    // (new groups without history) still gets footer-based estimation.

    struct ColumnRawSizeInfo {

        int64_t total_raw_bytes = 0;

        int64_t rows_with_data = 0;

    };

    std::unordered_map<int32_t, ColumnRawSizeInfo> column_raw_sizes;

    bool need_footer_collection = false;

    if (config::compaction_batch_size == -1) {

        std::unique_lock<std::mutex> lock(sample_info_lock);

        for (const auto& info : sample_infos) {

            if (info.group_data_size <= 0 && info.bytes <= 0 && info.rows <= 0) {

                need_footer_collection = true;

                break;

            }

        }

    }

    if (need_footer_collection) {

        for (const auto& rs_reader : src_rowset_readers) {

            auto beta_rowset = std::dynamic_pointer_cast<BetaRowset>(rs_reader->rowset());

            if (!beta_rowset) {

                continue;

            }

            std::vector<segment_v2::SegmentSharedPtr> segments;

            auto st = beta_rowset->load_segments(&segments);

            if (!st.ok()) {

                LOG(WARNING) << "Failed to load segments for footer raw_data_bytes collection"

                             << ", tablet_id: " << tablet->tablet_id()

                             << ", rowset_id: " << beta_rowset->rowset_id() << ", status: " << st;

                continue;

            }

            for (const auto& segment : segments) {

                int64_t row_count = segment->num_rows();

                auto collect_st = segment->traverse_column_meta_pbs(

                        [&](const segment_v2::ColumnMetaPB& meta) {

                            int32_t uid = meta.unique_id();

                            if (uid >= 0 && meta.has_raw_data_bytes()) {

                                auto& info = column_raw_sizes[uid];

                                info.total_raw_bytes += meta.raw_data_bytes();

                                info.rows_with_data += row_count;

                            }

                        });

                if (!collect_st.ok()) {

                    LOG(WARNING) << "Failed to traverse column meta for footer collection"

                                 << ", tablet_id: " << tablet->tablet_id()

                                 << ", status: " << collect_st;

                }

            }

        }

    }

    // Pre-compute per-row estimate for each column group from footer data.

    std::vector<int64_t> group_per_row_from_footer(column_groups.size(), 0);

    std::vector<bool> group_footer_fallback(column_groups.size(), false);

    for (size_t i = 0; i < column_groups.size(); ++i) {

        int64_t group_per_row = 0;

        bool need_fallback = false;

        for (uint32_t col_ordinal : column_groups[i]) {

            const auto& col = tablet_schema.column(col_ordinal);

            int32_t uid = col.unique_id();

            // Variant columns (root or subcolumn): raw_data_bytes is 0 (TODO in writer),

            // cannot estimate from footer, fallback to default for the entire group.

            if (uid < 0 || col.is_variant_type()) {

                need_fallback = true;

                break;

            }

            // Any column without footer data (e.g. legacy segments written before

            // raw_data_bytes existed) makes the group sample partial and unreliable.

            // Fall back to the default for the whole group instead of summing only

            // the columns we measured.

            auto it = column_raw_sizes.find(uid);

            if (it == column_raw_sizes.end() || it->second.rows_with_data <= 0) {

                need_fallback = true;

                break;

            }

            int64_t raw_per_row = it->second.total_raw_bytes / it->second.rows_with_data;

            int64_t col_per_row = 0;

            if (col.type() == FieldType::OLAP_FIELD_TYPE_ARRAY ||

                col.type() == FieldType::OLAP_FIELD_TYPE_MAP ||

                col.type() == FieldType::OLAP_FIELD_TYPE_STRUCT) {

                // Complex types: raw_data_bytes recursively aggregates sub-writers.

                col_per_row = raw_per_row;

            } else if (col.is_length_variable_type()) {

                // Variable-length scalar (VARCHAR/STRING/HLL/BITMAP/...): raw_per_row

                // is the average char payload across all rows; reader still pays an

                // 8-byte offset entry per row regardless of null-ness.

                col_per_row = raw_per_row + 8;

                if (col.is_nullable()) {

                    col_per_row += 1; // null map

                }

            } else {

                // Fixed-width scalar (INT/BIGINT/DOUBLE/DATE/...).

                // raw_data_bytes only counts non-null payload (append_nulls() does

                // not advance the page builder), but FileColumnIterator::next_batch

                // still calls ColumnNullable::insert_many_defaults() for null runs,

                // which grows the nested PODArray by N * type_size. So the runtime

                // per-row footprint is at least type_size, no matter how sparse.

                int64_t type_size = field_type_size(col.type());

                col_per_row = std::max(raw_per_row, type_size);

                if (col.is_nullable()) {

                    col_per_row += 1; // null map

                }

            }

            group_per_row += col_per_row;

        }

        group_per_row_from_footer[i] = group_per_row;

        group_footer_fallback[i] = need_fallback;

    }

    // compact group one by one

    for (auto i = 0; i < column_groups.size(); ++i) {

        VLOG_NOTICE << "row source size: " << row_sources_buf.total_size();

        bool is_key = (i == 0);

        int64_t batch_size = config::compaction_batch_size != -1

                                     ? config::compaction_batch_size

                                     : estimate_batch_size(i, tablet, merge_way_num, reader_type,

                                                           group_per_row_from_footer[i],

                                                           group_footer_fallback[i]);

        CompactionSampleInfo sample_info;

        Merger::Statistics group_stats;

        group_stats.rowid_conversion = total_stats.rowid_conversion;

        Merger::Statistics* group_stats_ptr = stats_output != nullptr ? &group_stats : nullptr;

        Status st = vertical_compact_one_group(

                tablet, reader_type, tablet_schema, is_key, column_groups[i], &row_sources_buf,

                src_rowset_readers, dst_rowset_writer, max_rows_per_segment, group_stats_ptr,

                key_group_cluster_key_idxes, batch_size, &sample_info, &context_stats,

                enable_sparse_optimization);

        {

            std::unique_lock<std::mutex> lock(sample_info_lock);

            sample_infos[i] = sample_info;

        }

        RETURN_IF_ERROR(st);

        if (stats_output != nullptr) {

            total_stats.bytes_read_from_local += group_stats.bytes_read_from_local;

            total_stats.bytes_read_from_remote += group_stats.bytes_read_from_remote;

            total_stats.cached_bytes_total += group_stats.cached_bytes_total;

            total_stats.cloud_local_read_time += group_stats.cloud_local_read_time;

            total_stats.cloud_remote_read_time += group_stats.cloud_remote_read_time;

            if (is_key) {

                total_stats.output_rows = group_stats.output_rows;

                total_stats.merged_rows = group_stats.merged_rows;

                total_stats.filtered_rows = group_stats.filtered_rows;

                total_stats.rowid_conversion = group_stats.rowid_conversion;

            }

        }

        if (progress_cb) {

            progress_cb(column_groups.size(), i + 1);

        }

        if (is_key) {

            RETURN_IF_ERROR(row_sources_buf.flush());

        }

        RETURN_IF_ERROR(row_sources_buf.seek_to_begin());

    }

    // Calculate and store density for next compaction's sparse optimization threshold

    // density = (total_cells - total_null_count) / total_cells

    // Smaller density means more sparse

    {

        std::unique_lock<std::mutex> lock(sample_info_lock);

        int64_t total_null_count = 0;

        for (const auto& info : sample_infos) {

            total_null_count += info.null_count;

        }

        int64_t total_cells = total_rows * tablet_schema.num_columns();

        if (total_cells > 0) {

            double density = static_cast<double>(total_cells - total_null_count) /

                             static_cast<double>(total_cells);

            tablet->compaction_density.store(density);

            LOG(INFO) << "Vertical compaction density update: tablet_id=" << tablet->tablet_id()

                      << ", total_cells=" << total_cells

                      << ", total_null_count=" << total_null_count << ", density=" << density;

        }

    }

    // finish compact, build output rowset

    VLOG_NOTICE << "finish compact groups";

    RETURN_IF_ERROR(dst_rowset_writer->final_flush());

    if (stats_output != nullptr) {

        *stats_output = total_stats;

    }

    return Status::OK();

}

} // namespace doris