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

#include <crc32c/crc32c.h>

#include <gen_cpp/Descriptors_types.h>

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/olap_file.pb.h>

#include <gen_cpp/segment_v2.pb.h>

#include <algorithm>

#include <cstring>

#include <memory>

#include <set>

#include <sstream>

#include <utility>

#include "cloud/config.h"

#include "common/config.h"

#include "common/exception.h"

#include "common/logging.h"

#include "common/status.h"

#include "core/column/column.h"

#include "core/data_type/data_type.h"

#include "core/data_type/data_type_factory.hpp"

#include "core/data_type/data_type_nullable.h"

#include "core/data_type/data_type_variant.h"

#include "core/data_type/data_type_variant_v2.h"

#include "core/field.h"

#include "core/string_ref.h"

#include "cpp/sync_point.h"

#include "exec/common/variant_util.h"

#include "exprs/expr_zonemap_filter.h"

#include "exprs/vexpr_context.h"

#include "io/cache/block_file_cache.h"

#include "io/cache/block_file_cache_factory.h"

#include "io/cache/cached_remote_file_reader.h"

#include "io/fs/file_reader.h"

#include "io/fs/file_system.h"

#include "io/io_common.h"

#include "runtime/exec_env.h"

#include "runtime/query_context.h"

#include "runtime/runtime_predicate.h"

#include "runtime/runtime_state.h"

#include "storage/index/index_file_reader.h"

#include "storage/index/indexed_column_reader.h"

#include "storage/index/primary_key_index.h"

#include "storage/index/short_key_index.h"

#include "storage/index/zone_map/zonemap_eval_context.h"

#include "storage/iterator/vgeneric_iterators.h"

#include "storage/iterators.h"

#include "storage/key_coder.h"

#include "storage/olap_common.h"

#include "storage/predicate/block_column_predicate.h"

#include "storage/predicate/column_predicate.h"

#include "storage/rowset/rowset_reader_context.h"

#include "storage/schema.h"

#include "storage/segment/column_meta_accessor.h"

#include "storage/segment/column_reader.h"

#include "storage/segment/column_reader_cache.h"

#include "storage/segment/empty_segment_iterator.h"

#include "storage/segment/page_io.h"

#include "storage/segment/page_pointer.h"

#include "storage/segment/segment_iterator.h"

#include "storage/segment/segment_writer.h" // k_segment_magic_length

#include "storage/segment/stream_reader.h"

#include "storage/segment/variant/variant_column_reader.h"

#include "storage/tablet/tablet_schema.h"

#include "storage/types.h"

#include "storage/utils.h"

#include "util/coding.h"

#include "util/json/path_in_data.h"

#include "util/slice.h" // Slice

namespace doris::segment_v2 {

namespace {

Status build_segment_zonemap_context(Segment* segment, const ReadSchema& schema,

                                     const StorageReadOptions& read_options,

                                     const VExprContextSPtrs& conjuncts, ZoneMapEvalContext* ctx) {

    DORIS_CHECK(segment != nullptr);

    DORIS_CHECK(ctx != nullptr);

    std::set<int> slot_indexes;

    for (const auto& conjunct : conjuncts) {

        DORIS_CHECK(conjunct != nullptr);

        const auto& root = conjunct->root();

        DORIS_CHECK(root != nullptr);

        if (!root->can_evaluate_zonemap_filter()) {

            continue;

        }

        // Segment zone maps have one min/max/null summary per column for the whole segment, so a

        // segment-level context can safely hold every slot referenced by a compound expression.

        // Page zone maps are page-aligned per column and still use single-slot filtering in

        // SegmentIterator.

        root->collect_slot_column_ids(slot_indexes);

    }

    for (const int slot_index : slot_indexes) {

        const auto ordinal = cast_set<ColumnId>(slot_index);

        DORIS_CHECK_LT(ordinal, schema.num_block_columns());

        const auto* tablet_column = schema.column(ordinal);

        if (!segment->can_apply_predicate_safely(

                    slot_index, schema, read_options.target_cast_type_for_variants, read_options)) {

            continue;

        }

        auto data_type =

                segment->get_data_type_of(*tablet_column, schema.data_type(ordinal), read_options);

        if (data_type == nullptr) {

            continue;

        }

        ZoneMapEvalContext::SlotZoneMap slot_zone_map;

        slot_zone_map.data_type = data_type;

        std::shared_ptr<ColumnReader> reader;

        Status st = segment->get_column_reader(*tablet_column, &reader, read_options.stats,

                                               &read_options.io_ctx);

        if (st.is<ErrorCode::NOT_FOUND>()) {

            ctx->slots.emplace(slot_index, std::move(slot_zone_map));

            continue;

        }

        RETURN_IF_ERROR(st);

        if (reader != nullptr && reader->has_zone_map()) {

            ZoneMap zone_map;

            RETURN_IF_ERROR(reader->get_segment_zone_map(&zone_map));

            slot_zone_map.zone_map = std::make_shared<ZoneMap>(std::move(zone_map));

        }

        ctx->slots.emplace(slot_index, std::move(slot_zone_map));

    }

    return Status::OK();

}

void fill_missing_decimal_precision(const TabletColumn& column, ColumnMetaPB* meta) {

    auto meta_type = static_cast<FieldType>(meta->type());

    if (meta_type != column.type()) {

        return;

    }

    if (field_is_decimal_type(meta_type)) {

        if ((!meta->has_precision() || meta->precision() <= 0) && column.precision() > 0) {

            meta->set_precision(column.precision());

        }

        if ((!meta->has_frac() || meta->frac() < 0) && column.frac() >= 0) {

            meta->set_frac(column.frac());

        }

    }

    // Complex column meta may also include storage helper children, such as

    // array offsets. Only schema children have matching TabletColumn subtypes.

    int child_count =

            std::min(meta->children_columns_size(), static_cast<int>(column.get_subtype_count()));

    for (int i = 0; i < child_count; ++i) {

        fill_missing_decimal_precision(column.get_sub_column(i), meta->mutable_children_columns(i));

    }

}

void fill_missing_decimal_precision_from_schema(const TabletSchemaSPtr& tablet_schema,

                                                ColumnMetaPB* meta) {

    if (!meta->has_unique_id()) {

        return;

    }

    int32_t col_idx = tablet_schema->field_index(static_cast<int32_t>(meta->unique_id()));

    if (col_idx < 0) {

        return;

    }

    fill_missing_decimal_precision(tablet_schema->column(col_idx), meta);

}

void fill_footer_missing_decimal_precision(const TabletSchemaSPtr& tablet_schema,

                                           SegmentFooterPB* footer) {

    for (int i = 0; i < footer->columns_size(); ++i) {

        fill_missing_decimal_precision_from_schema(tablet_schema, footer->mutable_columns(i));

    }

}

io::IOContext create_index_io_context(const io::IOContext* source, OlapReaderStatistics* stats) {

    io::IOContext io_ctx;

    if (source != nullptr) {

        io_ctx = *source;

    }

    io_ctx.is_index_data = true;

    io_ctx.is_inverted_index = false;

    io_ctx.file_cache_stats = stats ? &stats->file_cache_stats : nullptr;

    return io_ctx;

}

} // namespace

Status Segment::open(io::FileSystemSPtr fs, const std::string& path, int64_t tablet_id,

                     uint32_t segment_id, RowsetId rowset_id, TabletSchemaSPtr tablet_schema,

                     const io::FileReaderOptions& reader_options, std::shared_ptr<Segment>* output,

                     InvertedIndexFileInfo idx_file_info, OlapReaderStatistics* stats,

                     const io::IOContext* source_io_ctx) {

    // Ensure tablet_id is available in reader_options for CachedRemoteFileReader peer read.

    io::FileReaderOptions opts_with_tablet = reader_options;

    opts_with_tablet.tablet_id = tablet_id;

    auto s = _open(fs, path, segment_id, rowset_id, tablet_schema, opts_with_tablet, output,

                   idx_file_info, stats, source_io_ctx);

    if (s.ok() && output && *output) {

        (*output)->_tablet_id = tablet_id;

    }

    if (!s.ok()) {

        if (!config::is_cloud_mode()) {

            auto res = ExecEnv::get_tablet(tablet_id);

            TabletSharedPtr tablet =

                    res.has_value() ? std::dynamic_pointer_cast<Tablet>(res.value()) : nullptr;

            if (tablet) {

                tablet->report_error(s);

            }

        }

    }

    return s;

}

Status Segment::_open(io::FileSystemSPtr fs, const std::string& path, uint32_t segment_id,

                      RowsetId rowset_id, TabletSchemaSPtr tablet_schema,

                      const io::FileReaderOptions& reader_options, std::shared_ptr<Segment>* output,

                      InvertedIndexFileInfo idx_file_info, OlapReaderStatistics* stats,

                      const io::IOContext* source_io_ctx) {

    io::FileReaderSPtr file_reader;

    auto st = fs->open_file(path, &file_reader, &reader_options);

    TEST_INJECTION_POINT_CALLBACK("Segment::open:corruption", &st);

    std::shared_ptr<Segment> segment(

            new Segment(segment_id, rowset_id, std::move(tablet_schema), idx_file_info));

    segment->_seg_path = path;

    if (st) {

        segment->_fs = fs;

        segment->_file_reader = std::move(file_reader);

        st = segment->_open(stats, source_io_ctx);

    }

    // Three-tier retry for CORRUPTION errors when file cache is enabled.

    // This handles CORRUPTION from both open_file() and _parse_footer() (via _open()).

    if (st.is<ErrorCode::CORRUPTION>() &&

        reader_options.cache_type == io::FileCachePolicy::FILE_BLOCK_CACHE) {

        // Tier 1: Clear file cache and retry with cache support (re-downloads from remote).

        LOG(WARNING) << "bad segment file may be read from file cache, try to read remote source "

                        "file directly, file path: "

                     << path << " cache_key: " << file_cache_key_str(path);

        auto file_key = file_cache_key_from_path(path);

        auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);

        file_cache->remove_if_cached(file_key);

        st = fs->open_file(path, &file_reader, &reader_options);

        if (st) {

            segment->_fs = fs;

            segment->_file_reader = std::move(file_reader);

            st = segment->_open(stats, source_io_ctx);

        }

        TEST_INJECTION_POINT_CALLBACK("Segment::open:corruption1", &st);

        if (st.is<ErrorCode::CORRUPTION>()) { // corrupt again

            // Tier 2: Bypass cache entirely and read directly from remote storage.

            LOG(WARNING) << "failed to try to read remote source file again with cache support,"

                         << " try to read from remote directly, "

                         << " file path: " << path << " cache_key: " << file_cache_key_str(path);

            file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);

            file_cache->remove_if_cached(file_key);

            io::FileReaderOptions opt = reader_options;

            opt.cache_type = io::FileCachePolicy::NO_CACHE; // skip cache

            RETURN_IF_ERROR(fs->open_file(path, &file_reader, &opt));

            segment->_fs = fs;

            segment->_file_reader = std::move(file_reader);

            st = segment->_open(stats, source_io_ctx);

            if (!st.ok()) {

                // Tier 3: Remote source itself is corrupt.

                LOG(WARNING) << "failed to try to read remote source file directly,"

                             << " file path: " << path

                             << " cache_key: " << file_cache_key_str(path);

            }

        }

    }

    RETURN_IF_ERROR(st);

    DCHECK(segment->_fs != nullptr) << "file system is nullptr after segment open";

    *output = std::move(segment);

    return Status::OK();

}

Segment::Segment(uint32_t segment_id, RowsetId rowset_id, TabletSchemaSPtr tablet_schema,

                 InvertedIndexFileInfo idx_file_info)

        : _segment_id(segment_id),

          _meta_mem_usage(0),

          _rowset_id(rowset_id),

          _tablet_schema(std::move(tablet_schema)),

          _idx_file_info(std::move(idx_file_info)) {}

Segment::~Segment() {

    g_segment_estimate_mem_bytes << -_tracked_meta_mem_usage;

    // if failed, fix `_tracked_meta_mem_usage` accuracy

    DCHECK(_tracked_meta_mem_usage == meta_mem_usage());

}

io::UInt128Wrapper Segment::file_cache_key(std::string_view rowset_id, uint32_t seg_id) {

    return io::BlockFileCache::hash(fmt::format("{}_{}.dat", rowset_id, seg_id));

}

int64_t Segment::get_metadata_size() const {

    std::shared_ptr<SegmentFooterPB> footer_pb_shared = _footer_pb.lock();

    return sizeof(Segment) + (_pk_index_meta ? _pk_index_meta->ByteSizeLong() : 0) +

           (footer_pb_shared ? footer_pb_shared->ByteSizeLong() : 0);

}

void Segment::update_metadata_size() {

    MetadataAdder::update_metadata_size();

    g_segment_estimate_mem_bytes << _meta_mem_usage - _tracked_meta_mem_usage;

    _tracked_meta_mem_usage = _meta_mem_usage;

}

Status Segment::_open(OlapReaderStatistics* stats, const io::IOContext* source_io_ctx) {

    std::shared_ptr<SegmentFooterPB> footer_pb_shared;

    RETURN_IF_ERROR(_get_segment_footer(footer_pb_shared, stats, source_io_ctx));

    _pk_index_meta.reset(

            footer_pb_shared->has_primary_key_index_meta()

                    ? new PrimaryKeyIndexMetaPB(footer_pb_shared->primary_key_index_meta())

                    : nullptr);

    // delete_bitmap_calculator_test.cpp

    // DCHECK(footer.has_short_key_index_page());

    _sk_index_page = footer_pb_shared->short_key_index_page();

    _num_rows = footer_pb_shared->num_rows();

    // An estimated memory usage of a segment

    // Footer is seperated to StoragePageCache so we don't need to add it to _meta_mem_usage

    // _meta_mem_usage += footer_pb_shared->ByteSizeLong();

    if (_pk_index_meta != nullptr) {

        _meta_mem_usage += _pk_index_meta->ByteSizeLong();

    }

    _meta_mem_usage += sizeof(*this);

    _meta_mem_usage += std::min(static_cast<int>(_tablet_schema->num_columns()),

                                config::max_segment_partial_column_cache_size) *

                       config::estimated_mem_per_column_reader;

    // 1024 comes from SegmentWriterOptions

    _meta_mem_usage += (_num_rows + 1023) / 1024 * (36 + 4);

    // 0.01 comes from PrimaryKeyIndexBuilder::init

    _meta_mem_usage += BloomFilter::optimal_bit_num(_num_rows, 0.01) / 8;

    update_metadata_size();

    return Status::OK();

}

Status Segment::_open_index_file_reader() {

    // Derive the index path from `_seg_path`, not `_file_reader->path()`: remote FS normalizes the

    // latter to an absolute path that won't match the relative keys in PackedFileSystem's index map.

    _index_file_reader = std::make_shared<IndexFileReader>(

            _fs, std::string {InvertedIndexDescriptor::get_index_file_path_prefix(_seg_path)},

            _tablet_schema->get_inverted_index_storage_format(), _idx_file_info, _tablet_id);

    return Status::OK();

}

bool Segment::is_tso_placeholder_col(int cid, const ReadSchema& schema,

                                     const StorageReadOptions& read_options) const {

    if (read_options.version.first != read_options.version.second) {

        return false;

    }

    if (!read_options.read_row_binlog) {

        return false;

    }

    // tso_ordinal() is -1 for non-binlog schemas, so this returns false there.

    return cid == schema.tso_ordinal();

}

Status Segment::new_iterator(ReadSchemaSPtr schema, const StorageReadOptions& read_options,

                             std::unique_ptr<RowwiseIterator>* iter) {

    if (read_options.runtime_state != nullptr) {

        _be_exec_version = read_options.runtime_state->be_exec_version();

    }

    RETURN_IF_ERROR(_create_column_meta_once(read_options.stats, &read_options.io_ctx));

    read_options.stats->total_segment_number++;

    // trying to prune the current segment by segment-level zone map

    for (const auto& entry : read_options.col_id_to_predicates) {

        // col_id_to_predicates is keyed by read-schema ordinal.

        int32_t column_id = entry.first;

        const TabletColumn& col = *schema->column(column_id);

        std::shared_ptr<ColumnReader> reader;

        // __DORIS_COMMIT_TSO_COL__ on a single-version segment stores a 0 placeholder on disk

        // (replaced with the rowset's real commit_tso at read time). Its on-disk zonemap [0,0]

        // must not drive segment-level pruning, so build a ConstantColumnReader carrying the real

        // commit_tso to prune against the real value instead.

        std::optional<Field> const_value;

        if (read_options.version.first == read_options.version.second &&

            column_id == schema->commit_tso_ordinal() && read_options.commit_tso.end_tso() != -1) {

            const_value = Field::create_field<TYPE_BIGINT>(read_options.commit_tso.end_tso());

        }

        Status st = get_column_reader(col, &reader, read_options.stats, &read_options.io_ctx,

                                      std::move(const_value));

        // not found in this segment, skip

        if (st.is<ErrorCode::NOT_FOUND>()) {

            continue;

        }

        RETURN_IF_ERROR(st);

        // should be OK

        DCHECK(reader != nullptr);

        if (!reader->has_zone_map()) {

            continue;

        }

        // Placeholder tso column on a single-version binlog segment: its zonemap reflects the

        // NULL placeholder (replaced with commit_tso at read time), so skip pruning by

        // zonemap (min == max == commit_tso) and reuse the predicate's own zonemap matching:

        // evaluate_and() returns false iff no value in [min, max] can satisfy the predicates,

        // i.e. commit_tso fails them and the whole segment can be pruned. Predicates that don't

        // support zonemap return true (conservative: not pruned, row-level eval handles them).

        if (read_options.col_id_to_predicates.contains(column_id) &&

            is_tso_placeholder_col(column_id, *schema, read_options)) {

            const Int64 commit_tso =

                    read_options.commit_tso.end_tso() == -1 ? 0 : read_options.commit_tso.end_tso();

            ZoneMap zone_map;

            zone_map.min_value = Field::create_field<TYPE_BIGINT>(commit_tso);

            zone_map.max_value = Field::create_field<TYPE_BIGINT>(commit_tso);

            zone_map.has_not_null = true;

            if (!entry.second->evaluate_and(zone_map)) {

                // any condition not satisfied, return.

                *iter = std::make_unique<EmptySegmentIterator>(*schema);

                read_options.stats->filtered_segment_number++;

                return Status::OK();

            }

            continue;

        }

        if (read_options.col_id_to_predicates.contains(column_id) &&

            can_apply_predicate_safely(column_id, *schema,

                                       read_options.target_cast_type_for_variants, read_options)) {

            bool matched = true;

            RETURN_IF_ERROR(reader->match_condition(entry.second.get(), &matched));

            if (!matched) {

                // any condition not satisfied, return.

                *iter = std::make_unique<EmptySegmentIterator>(*schema);

                read_options.stats->filtered_segment_number++;

                read_options.stats->rows_stats_filtered += num_rows();

                return Status::OK();

            }

        }

    }

    if (expr_zonemap::is_expr_zonemap_filter_enabled(read_options.runtime_state) &&

        !read_options.common_expr_ctxs_push_down.empty()) {

        ZoneMapEvalContext ctx;

        RETURN_IF_ERROR(build_segment_zonemap_context(

                this, *schema, read_options, read_options.common_expr_ctxs_push_down, &ctx));

        const auto result =

                VExprContext::evaluate_zonemap_filter(read_options.common_expr_ctxs_push_down, ctx);

        ctx.stats.accumulate_to(read_options.stats);

        if (result == ZoneMapFilterResult::kNoMatch) {

            *iter = std::make_unique<EmptySegmentIterator>(*schema);

            read_options.stats->filtered_segment_number++;

            read_options.stats->expr_zonemap_filtered_segments++;

            return Status::OK();

        }

    }

    {

        SCOPED_RAW_TIMER(&read_options.stats->segment_load_index_timer_ns);

        RETURN_IF_ERROR(load_index(read_options.stats, &read_options.io_ctx));

    }

    if (read_options.delete_condition_predicates->num_of_column_predicate() == 0 &&

        read_options.push_down_agg_type_opt != TPushAggOp::NONE &&

        read_options.push_down_agg_type_opt != TPushAggOp::COUNT_ON_INDEX) {

        iter->reset(new_vstatistics_iterator(this->shared_from_this(), *schema));

    } else {

        *iter = std::make_unique<SegmentIterator>(this->shared_from_this(), schema);

    }

    // TODO: Valid the opt not only in ReaderType::READER_QUERY

    if (read_options.io_ctx.reader_type == ReaderType::READER_QUERY &&

        !read_options.column_predicates.empty()) {

        auto pruned_predicates = read_options.column_predicates;

        auto pruned = false;

        for (auto& it : _column_reader_cache->get_available_readers(false)) {

            const auto uid = it.first;

            // Column readers are keyed by UID; predicates use read schema ordinals.

            const auto ordinal = schema->ordinal_by_uid(uid);

            if (ordinal < 0) {

                continue;

            }

            bool tmp_pruned = false;

            RETURN_IF_ERROR(it.second->prune_predicates_by_zone_map(pruned_predicates, ordinal,

                                                                    &tmp_pruned));

            pruned |= tmp_pruned;

        }

        if (pruned) {

            auto options_with_pruned_predicates = read_options;

            options_with_pruned_predicates.column_predicates = pruned_predicates;

            //because column_predicates is changed, we need to rebuild col_id_to_predicates so that inverted index will not go through it.

            options_with_pruned_predicates.col_id_to_predicates.clear();

            for (auto pred : options_with_pruned_predicates.column_predicates) {

                if (!options_with_pruned_predicates.col_id_to_predicates.contains(

                            pred->column_id())) {

                    options_with_pruned_predicates.col_id_to_predicates.insert(

                            {pred->column_id(), AndBlockColumnPredicate::create_shared()});

                }

                options_with_pruned_predicates.col_id_to_predicates[pred->column_id()]

                        ->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));

            }

            for (const auto& pred : read_options.column_predicates) {

                const auto pred_cid = pred->column_id();

                // Key columns may still be required by key range seeks even if the segment zone

                // map proves their predicates always true. Only mark non-key columns as safe for

                // the no-need-read path.

                if (!schema->column(pred_cid)->is_key() &&

                    !options_with_pruned_predicates.col_id_to_predicates.contains(pred_cid)) {

                    options_with_pruned_predicates.zonemap_always_true_pred_cols.insert(pred_cid);

                }

            }

            return iter->get()->init(options_with_pruned_predicates);

        }

    }

    return iter->get()->init(read_options);

}

Status Segment::_write_error_file(size_t file_size, size_t offset, size_t bytes_read, char* data,

                                  io::IOContext& io_ctx) {

    if (!config::enbale_dump_error_file || !doris::config::is_cloud_mode()) {

        return Status::OK();

    }

    std::string file_name = _rowset_id.to_string() + "_" + std::to_string(_segment_id) + ".dat";

    std::string dir_path = io::FileCacheFactory::instance()->get_base_paths()[0] + "/error_file/";

    Status create_st = io::global_local_filesystem()->create_directory(dir_path, true);

    if (!create_st.ok() && !create_st.is<ErrorCode::ALREADY_EXIST>()) {

        LOG(WARNING) << "failed to create error file dir: " << create_st.to_string();

        return create_st;

    }

    size_t dir_size = 0;

    RETURN_IF_ERROR(io::global_local_filesystem()->directory_size(dir_path, &dir_size));

    if (dir_size > config::file_cache_error_log_limit_bytes) {

        LOG(WARNING) << "error file dir size is too large: " << dir_size;

        return Status::OK();

    }

    std::string error_part;

    error_part.resize(bytes_read);

    std::string part_path = dir_path + file_name + ".part_offset_" + std::to_string(offset);

    LOG(WARNING) << "writer error part to " << part_path;

    bool is_part_exist = false;

    RETURN_IF_ERROR(io::global_local_filesystem()->exists(part_path, &is_part_exist));

    if (is_part_exist) {

        LOG(WARNING) << "error part already exists: " << part_path;

    } else {

        std::unique_ptr<io::FileWriter> part_writer;

        RETURN_IF_ERROR(io::global_local_filesystem()->create_file(part_path, &part_writer));

        RETURN_IF_ERROR(part_writer->append(Slice(data, bytes_read)));

        RETURN_IF_ERROR(part_writer->close());

    }

    std::string error_file;

    error_file.resize(file_size);

    auto* cached_reader = dynamic_cast<io::CachedRemoteFileReader*>(_file_reader.get());

    if (cached_reader == nullptr) {

        return Status::InternalError("file reader is not CachedRemoteFileReader");

    }

    size_t error_file_bytes_read = 0;

    RETURN_IF_ERROR(cached_reader->get_remote_reader()->read_at(

            0, Slice(error_file.data(), file_size), &error_file_bytes_read, &io_ctx));

    DCHECK(error_file_bytes_read == file_size);

    //std::string file_path = dir_path + std::to_string(cur_time) + "_" + ss.str();

    std::string file_path = dir_path + file_name;

    LOG(WARNING) << "writer error file to " << file_path;

    bool is_file_exist = false;

    RETURN_IF_ERROR(io::global_local_filesystem()->exists(file_path, &is_file_exist));

    if (is_file_exist) {

        LOG(WARNING) << "error file already exists: " << part_path;

    } else {

        std::unique_ptr<io::FileWriter> writer;

        RETURN_IF_ERROR(io::global_local_filesystem()->create_file(file_path, &writer));

        RETURN_IF_ERROR(writer->append(Slice(error_file.data(), file_size)));

        RETURN_IF_ERROR(writer->close());

    }

    return Status::OK(); // already exists

};

Status Segment::_parse_footer(std::shared_ptr<SegmentFooterPB>& footer, OlapReaderStatistics* stats,

                              const io::IOContext* source_io_ctx) {

    // Footer := SegmentFooterPB, FooterPBSize(4), FooterPBChecksum(4), MagicNumber(4)

    auto file_size = _file_reader->size();

    if (file_size < 12) {

        return Status::Corruption("Bad segment file {}: file size {} < 12, cache_key: {}",

                                  _file_reader->path().native(), file_size,

                                  file_cache_key_str(_file_reader->path().native()));

    }

    uint8_t fixed_buf[12];

    size_t bytes_read = 0;

    auto io_ctx = create_index_io_context(source_io_ctx, stats);

    TEST_SYNC_POINT_CALLBACK("Segment::_parse_footer::io_ctx", &io_ctx);

    RETURN_IF_ERROR(

            _file_reader->read_at(file_size - 12, Slice(fixed_buf, 12), &bytes_read, &io_ctx));

    DCHECK_EQ(bytes_read, 12);

    TEST_SYNC_POINT_CALLBACK("Segment::parse_footer:magic_number_corruption", fixed_buf);

    TEST_INJECTION_POINT_CALLBACK("Segment::parse_footer:magic_number_corruption_inj", fixed_buf);

    if (memcmp(fixed_buf + 8, k_segment_magic, k_segment_magic_length) != 0) {

        Status st =

                _write_error_file(file_size, file_size - 12, bytes_read, (char*)fixed_buf, io_ctx);

        if (!st.ok()) {

            LOG(WARNING) << "failed to write error file: " << st.to_string();

        }

        return Status::Corruption(

                "Bad segment file {}: file_size: {}, magic number not match, cache_key: {}",

                _file_reader->path().native(), file_size,

                file_cache_key_str(_file_reader->path().native()));

    }

    // read footer PB

    uint32_t footer_length = decode_fixed32_le(fixed_buf);

    if (file_size < 12 + footer_length) {

        Status st =

                _write_error_file(file_size, file_size - 12, bytes_read, (char*)fixed_buf, io_ctx);

        if (!st.ok()) {

            LOG(WARNING) << "failed to write error file: " << st.to_string();

        }

        return Status::Corruption("Bad segment file {}: file size {} < {}, cache_key: {}",

                                  _file_reader->path().native(), file_size, 12 + footer_length,

                                  file_cache_key_str(_file_reader->path().native()));

    }

    std::string footer_buf;

    footer_buf.resize(footer_length);

    RETURN_IF_ERROR(_file_reader->read_at(file_size - 12 - footer_length, footer_buf, &bytes_read,

                                          &io_ctx));

    DCHECK_EQ(bytes_read, footer_length);

    // validate footer PB's checksum

    uint32_t expect_checksum = decode_fixed32_le(fixed_buf + 4);

    uint32_t actual_checksum = crc32c::Crc32c(footer_buf.data(), footer_buf.size());

    if (actual_checksum != expect_checksum) {

        Status st = _write_error_file(file_size, file_size - 12 - footer_length, bytes_read,

                                      footer_buf.data(), io_ctx);

        if (!st.ok()) {

            LOG(WARNING) << "failed to write error file: " << st.to_string();

        }

        return Status::Corruption(

                "Bad segment file {}: file_size = {}, footer checksum not match, actual={} "

                "vs expect={}, cache_key: {}",

                _file_reader->path().native(), file_size, actual_checksum, expect_checksum,

                file_cache_key_str(_file_reader->path().native()));

    }

    // deserialize footer PB

    footer = std::make_shared<SegmentFooterPB>();

    if (!footer->ParseFromString(footer_buf)) {

        Status st = _write_error_file(file_size, file_size - 12 - footer_length, bytes_read,

                                      footer_buf.data(), io_ctx);

        if (!st.ok()) {

            LOG(WARNING) << "failed to write error file: " << st.to_string();

        }

        return Status::Corruption(

                "Bad segment file {}: file_size = {}, failed to parse SegmentFooterPB, "

                "cache_key: ",

                _file_reader->path().native(), file_size,

                file_cache_key_str(_file_reader->path().native()));

    }

    // Segments written before #26572 do not persist decimal precision/frac in

    // ColumnMetaPB, so recover the logical p/s from TabletSchema before

    // ColumnReader builds DataTypeDecimal.

    fill_footer_missing_decimal_precision(_tablet_schema, footer.get());

    VLOG_DEBUG << fmt::format("Loading segment footer from {} finished",

                              _file_reader->path().native());

    return Status::OK();

}

Status Segment::_load_pk_bloom_filter(OlapReaderStatistics* stats,

                                      const io::IOContext* source_io_ctx) {

#ifdef BE_TEST

    if (_pk_index_meta == nullptr) {

        // for BE UT "segment_cache_test"

        return _load_pk_bf_once.call([this] {

            _meta_mem_usage += 100;

            update_metadata_size();

            return Status::OK();

        });

    }

#endif

    DCHECK(_tablet_schema->keys_type() == UNIQUE_KEYS);

    DCHECK(_pk_index_meta != nullptr);

    DCHECK(_pk_index_reader != nullptr);

    return _load_pk_bf_once.call([this, stats, source_io_ctx] {

        RETURN_IF_ERROR(

                _pk_index_reader->parse_bf(_file_reader, *_pk_index_meta, stats, source_io_ctx));

        // _meta_mem_usage += _pk_index_reader->get_bf_memory_size();

        return Status::OK();

    });

}

Status Segment::load_pk_index_and_bf(OlapReaderStatistics* index_load_stats,

                                     const io::IOContext* source_io_ctx) {

    // `DorisCallOnce` may catch exception in calling stack A and re-throw it in

    // a different calling stack B which doesn't have catch block. So we add catch block here

    // to prevent coreudmp

    RETURN_IF_CATCH_EXCEPTION({

        RETURN_IF_ERROR(load_index(index_load_stats, source_io_ctx));

        RETURN_IF_ERROR(_load_pk_bloom_filter(index_load_stats, source_io_ctx));

    });

    return Status::OK();

}

Status Segment::load_index(OlapReaderStatistics* stats, const io::IOContext* source_io_ctx) {

    return _load_index_once.call([this, stats, source_io_ctx] {

        if (_tablet_schema->keys_type() == UNIQUE_KEYS && _pk_index_meta != nullptr) {

            _pk_index_reader = std::make_unique<PrimaryKeyIndexReader>();

            RETURN_IF_ERROR(_pk_index_reader->parse_index(_file_reader, *_pk_index_meta, stats,

                                                          source_io_ctx));

            // _meta_mem_usage += _pk_index_reader->get_memory_size();

            return Status::OK();

        } else {

            // read and parse short key index page

            OlapReaderStatistics tmp_stats;

            OlapReaderStatistics* stats_ptr = stats != nullptr ? stats : &tmp_stats;

            auto page_io_ctx = create_index_io_context(source_io_ctx, stats_ptr);

            PageReadOptions opts(page_io_ctx);

            opts.use_page_cache = true;

            opts.type = INDEX_PAGE;

            opts.file_reader = _file_reader.get();

            opts.page_pointer = PagePointer(_sk_index_page);

            // short key index page uses NO_COMPRESSION for now

            opts.codec = nullptr;

            opts.stats = &tmp_stats;

            Slice body;

            PageFooterPB footer;

            RETURN_IF_ERROR(

                    PageIO::read_and_decompress_page(opts, &_sk_index_handle, &body, &footer));

            DCHECK_EQ(footer.type(), SHORT_KEY_PAGE);

            DCHECK(footer.has_short_key_page_footer());

            // _meta_mem_usage += body.get_size();

            _sk_index_decoder = std::make_unique<ShortKeyIndexDecoder>();

            return _sk_index_decoder->parse(body, footer.short_key_page_footer());

        }

    });

}

Status Segment::healthy_status() {

    try {

        if (_load_index_once.has_called()) {

            RETURN_IF_ERROR(_load_index_once.stored_result());

        }

        if (_load_pk_bf_once.has_called()) {

            RETURN_IF_ERROR(_load_pk_bf_once.stored_result());

        }

        if (_create_column_meta_once_call.has_called()) {

            RETURN_IF_ERROR(_create_column_meta_once_call.stored_result());

        }

        if (_index_file_reader_open.has_called()) {

            RETURN_IF_ERROR(_index_file_reader_open.stored_result());

        }

        // This status is set by running time, for example, if there is something wrong during read segment iterator.

        return _healthy_status.status();

    } catch (const doris::Exception& e) {

        // If there is an exception during load_xxx, should not throw exception directly because

        // the caller may not exception safe.

        return e.to_status();

    } catch (const std::exception& e) {

        // The exception is not thrown by doris code.

        return Status::InternalError("Unexcepted error during load segment: {}", e.what());

    }

}

DataTypePtr Segment::get_data_type_of(const TabletColumn& read_column, const DataTypePtr& read_type,

                                      const StorageReadOptions& read_options) {

    const PathInDataPtr path = read_column.path_info_ptr();

    // none variant column

    if (path == nullptr || path->empty()) {

        return read_type;

    }

    // Path exists, proceed with variant logic.

    PathInData relative_path = path->copy_pop_front();

    int32_t unique_id =

            read_column.unique_id() >= 0 ? read_column.unique_id() : read_column.parent_unique_id();

    // If this uid does not exist in segment meta, fallback to schema type.

    if (!_column_meta_accessor->has_column_uid(unique_id)) {

        return DataTypeFactory::instance().create_data_type(read_column);

    }

    std::shared_ptr<ColumnReader> v_reader;

    OlapReaderStatistics tmp_stats;

    auto* stats = read_options.stats == nullptr ? &tmp_stats : read_options.stats;

    // Get the parent variant column reader

    // If status is not ok, it will throw exception(data corruption)

    THROW_IF_ERROR(get_column_reader(unique_id, &v_reader, stats, &read_options.io_ctx));

    DCHECK(v_reader != nullptr);

    auto* variant_reader = static_cast<VariantColumnReader*>(v_reader.get());

    // Delegate type inference for variant paths to VariantColumnReader.

    DataTypePtr type;

    THROW_IF_ERROR(variant_reader->infer_data_type_for_path(&type, read_column, read_options,

                                                            _column_reader_cache.get()));

    return type;

}

Status Segment::_create_column_meta_once(OlapReaderStatistics* stats,

                                         const io::IOContext* source_io_ctx) {

    SCOPED_RAW_TIMER(&stats->segment_create_column_readers_timer_ns);

    return _create_column_meta_once_call.call([this, stats, source_io_ctx] {

        std::shared_ptr<SegmentFooterPB> footer_pb_shared;

        RETURN_IF_ERROR(_get_segment_footer(footer_pb_shared, stats, source_io_ctx));

        return _create_column_meta(*footer_pb_shared, stats, source_io_ctx);

    });

}

Status Segment::_create_column_meta(const SegmentFooterPB& footer, OlapReaderStatistics* stats,

                                    const io::IOContext* source_io_ctx) {

    // Initialize column meta accessor which internally maintains uid -> column_ordinal mapping.

    _column_meta_accessor = std::make_unique<ColumnMetaAccessor>();

    RETURN_IF_ERROR(_column_meta_accessor->init(footer, _file_reader));

    if (config::enable_adaptive_batch_size) {

        // Cache raw_data_bytes per column uid for adaptive batch size prediction.

        // This runs under call_once, so no thread-safety concerns.

        auto st = _column_meta_accessor->traverse_metas(

                footer,

                [this](const ColumnMetaPB& meta) {

                    if (meta.has_unique_id() && meta.unique_id() != -1 &&

                        meta.has_raw_data_bytes()) {

                        _column_uid_to_raw_bytes[meta.unique_id()] = meta.raw_data_bytes();

                    }

                },

                stats, source_io_ctx);

        if (!st.ok()) {

            LOG(WARNING) << "Failed to traverse column metas to cache raw_data_bytes, error: "

                         << st.to_string();

        }

    }

    _column_reader_cache = std::make_unique<ColumnReaderCache>(

            _column_meta_accessor.get(), _tablet_schema, _file_reader, _num_rows,

            [this](std::shared_ptr<SegmentFooterPB>& footer_pb, OlapReaderStatistics* stats,

                   const io::IOContext* io_ctx) {

                return _get_segment_footer(footer_pb, stats, io_ctx);

            });

    return Status::OK();

}

Status Segment::new_default_iterator(const TabletColumn& tablet_column,

                                     std::unique_ptr<ColumnIterator>* iter) {

    if (!tablet_column.has_default_value() && !tablet_column.is_nullable()) {

        return Status::InternalError(

                "invalid nonexistent column without default value. column_uid={}, "

                "column_name={}, "

                "column_type={}",

                tablet_column.unique_id(), tablet_column.name(), tablet_column.type());

    }

    std::unique_ptr<DefaultValueColumnIterator> default_value_iter(new DefaultValueColumnIterator(

            tablet_column.has_default_value(), tablet_column.default_value(),

            tablet_column.is_nullable(), tablet_column.type(), tablet_column.precision(),

            tablet_column.frac(), tablet_column.length()));

    ColumnIteratorOptions iter_opts;

    RETURN_IF_ERROR(default_value_iter->init(iter_opts));

    *iter = std::move(default_value_iter);

    return Status::OK();

}

// Not use cid anymore, for example original table schema is colA int, then user do following actions

// 1.add column b

// 2. drop column b

// 3. add column c

// in the new schema column c's cid == 2

// but in the old schema column b's cid == 2

// but they are not the same column

Status Segment::new_column_iterator(const TabletColumn& tablet_column,

                                    std::unique_ptr<ColumnIterator>* iter,

                                    const StorageReadOptions* opt,

                                    const std::unordered_map<int32_t, PathToBinaryColumnCacheUPtr>*

                                            variant_sparse_column_cache) {

    if (opt->runtime_state != nullptr) {

        _be_exec_version = opt->runtime_state->be_exec_version();

    }

    RETURN_IF_ERROR(_create_column_meta_once(opt->stats, &opt->io_ctx));

    // For compability reason unique_id may less than 0 for variant extracted column

    int32_t unique_id = tablet_column.unique_id() >= 0 ? tablet_column.unique_id()

                                                       : tablet_column.parent_unique_id();

    // If column meta for this uid is not found in this segment, use default iterator.

    if (!_column_meta_accessor->has_column_uid(unique_id)) {

        RETURN_IF_ERROR(new_default_iterator(tablet_column, iter));

        return Status::OK();

    }

    // __DORIS_COMMIT_TSO_COL__ on a single-version segment stores a 0 placeholder on disk (its

    // real value is the rowset's commit_tso, filled at read time). Pass the real commit_tso as a

    // const value so the cache returns a ConstantColumnReader, whose iterator yields the real value

    // on every read path (projection / predicate / MIN-MAX zone-map) instead of the placeholder 0.

    // commit_tso == -1 means it is not assigned yet (before publish); keep the on-disk value then.

    // The value is constant per segment (a segment belongs to a single rowset), so caching the

    // ConstantColumnReader does not cross-pollute other queries. Some internal read paths (e.g. MOW

    // partial-update row fetch) build a bare StorageReadOptions without tablet_schema, so guard it.

    std::optional<Field> const_value;

    if (opt->tablet_schema != nullptr && opt->version.first == opt->version.second &&

        opt->commit_tso.end_tso() != -1) {

        int32_t tso_idx = opt->tablet_schema->commit_tso_col_idx();

        if (tso_idx != -1 && opt->tablet_schema->column(tso_idx).unique_id() == unique_id) {

            const_value = Field::create_field<TYPE_BIGINT>(opt->commit_tso.end_tso());

        }

    }

    // init iterator by unique id

    std::shared_ptr<ColumnReader> reader;

    RETURN_IF_ERROR(get_column_reader(unique_id, &reader, opt->stats, &opt->io_ctx,

                                      std::move(const_value)));

    if (reader == nullptr) {

        return Status::InternalError("column reader is nullptr, unique_id={}", unique_id);

    }

    if (reader->get_meta_type() == FieldType::OLAP_FIELD_TYPE_VARIANT) {

        // if sparse_column_cache_ptr is nullptr, means the sparse column cache is not used

        PathToBinaryColumnCache* sparse_column_cache_ptr = nullptr;

        if (variant_sparse_column_cache) {

            auto it = variant_sparse_column_cache->find(unique_id);

            if (it != variant_sparse_column_cache->end()) {

                sparse_column_cache_ptr = it->second.get();

            } else {

                DCHECK(false) << "sparse column cache is not found, unique_id=" << unique_id;

            }

        }

        // use _column_reader_cache to get variant subcolumn(path column) reader

        RETURN_IF_ERROR(assert_cast<VariantColumnReader*>(reader.get())

                                ->new_iterator(iter, &tablet_column, opt,

                                               _column_reader_cache.get(),

                                               sparse_column_cache_ptr));

    } else {

        RETURN_IF_ERROR(reader->new_iterator(iter, &tablet_column, opt));

        if (opt->all_access_paths.contains(unique_id) ||

            opt->predicate_access_paths.contains(unique_id)) {

            const auto& all_access_paths = opt->all_access_paths.contains(unique_id)

                                                   ? opt->all_access_paths.at(unique_id)

                                                   : TColumnAccessPaths {};

            const auto& predicate_access_paths = opt->predicate_access_paths.contains(unique_id)

                                                         ? opt->predicate_access_paths.at(unique_id)

                                                         : TColumnAccessPaths {};

            // set column name to apply access paths.

            (*iter)->set_column_name(tablet_column.name());

            RETURN_IF_ERROR((*iter)->set_access_paths(all_access_paths, predicate_access_paths));

            (*iter)->remove_pruned_sub_iterators();

        }

    }

    if (config::enable_column_type_check && !tablet_column.has_path_info() &&

        !tablet_column.is_agg_state_type() && tablet_column.type() != reader->get_meta_type()) {

        LOG(WARNING) << "different type between schema and column reader,"

                     << " column schema name: " << tablet_column.name()

                     << " column schema type: " << int(tablet_column.type())

                     << " column reader meta type: " << int(reader->get_meta_type());

        return Status::InternalError("different type between schema and column reader");

    }

    return Status::OK();

}

Status Segment::get_column_reader(int32_t col_uid, std::shared_ptr<ColumnReader>* column_reader,

                                  OlapReaderStatistics* stats, const io::IOContext* source_io_ctx,

                                  std::optional<Field> const_value) {

    RETURN_IF_ERROR(_create_column_meta_once(stats, source_io_ctx));

    SCOPED_RAW_TIMER(&stats->segment_create_column_readers_timer_ns);

    // The column is not in this segment, return nullptr

    if (!_tablet_schema->has_column_unique_id(col_uid)) {

        *column_reader = nullptr;

        return Status::Error<ErrorCode::NOT_FOUND, false>("column not found in segment, col_uid={}",

                                                          col_uid);

    }

    return _column_reader_cache->get_column_reader(col_uid, column_reader, stats, source_io_ctx,

                                                   std::move(const_value));

}

Status Segment::traverse_column_meta_pbs(const std::function<void(const ColumnMetaPB&)>& visitor) {

    // Ensure column meta accessor and reader cache are initialized once.

    OlapReaderStatistics dummy_stats;

    RETURN_IF_ERROR(_create_column_meta_once(&dummy_stats));

    std::shared_ptr<SegmentFooterPB> footer_pb_shared;

    RETURN_IF_ERROR(_get_segment_footer(footer_pb_shared, &dummy_stats));

    return _column_meta_accessor->traverse_metas(*footer_pb_shared, visitor, &dummy_stats);

}

Status Segment::get_column_reader(const TabletColumn& col,

                                  std::shared_ptr<ColumnReader>* column_reader,

                                  OlapReaderStatistics* stats, const io::IOContext* source_io_ctx,

                                  std::optional<Field> const_value) {

    RETURN_IF_ERROR(_create_column_meta_once(stats, source_io_ctx));

    SCOPED_RAW_TIMER(&stats->segment_create_column_readers_timer_ns);

    int col_uid = col.unique_id() >= 0 ? col.unique_id() : col.parent_unique_id();