// 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.

#pragma once

#include <cstdint>

// clang20 + -O1 causes warnings about pass-failed

// error: loop not unrolled: the optimizer was unable to perform the requested transformation; the transformation might be disabled or specified as part of an unsupported transformation ordering [-Werror,-Wpass-failed=transform-warning]

#if defined(__clang__)

#pragma clang diagnostic push

#pragma clang diagnostic ignored "-Wpass-failed"

#endif

#include <limits>

#include <memory>

#if defined(__clang__)

#pragma clang diagnostic pop

#endif

#include <optional>

#include <string>

#include <utility>

#include <vector>

#include "cloud/cloud_tablet.h"

#include "common/status.h"

#include "io/io_common.h"

#include "runtime/runtime_profile.h"

#include "storage/compaction_task_tracker.h"

#include "storage/merger.h"

#include "storage/olap_common.h"

#include "storage/rowid_conversion.h"

#include "storage/rowset/pending_rowset_helper.h"

#include "storage/rowset/rowset_fwd.h"

#include "storage/tablet/tablet_fwd.h"

namespace doris {

class MemTrackerLimiter;

class RowsetWriter;

struct RowsetWriterContext;

class StorageEngine;

class CloudStorageEngine;

static constexpr int COMPACTION_DELETE_BITMAP_LOCK_ID = -1;

static constexpr int64_t INVALID_COMPACTION_INITIATOR_ID = -100;

// This class is a base class for compaction.

// The entrance of this class is compact()

// Any compaction should go through four procedures.

//  1. pick rowsets satisfied to compact

//  2. do compaction

//  3. modify rowsets

//  4. gc output rowset if failed

class Compaction {

public:

    Compaction(BaseTabletSPtr tablet, const std::string& label);

    virtual ~Compaction();

    // Pick input rowsets satisfied to compact

    virtual Status prepare_compact() = 0;

    // Merge input rowsets to output rowset and modify tablet meta

    virtual Status execute_compact() = 0;

    RuntimeProfile* runtime_profile() const { return _profile.get(); }

    virtual ReaderType compaction_type() const = 0;

    virtual std::string_view compaction_name() const = 0;

    // Returns compaction profile type for task tracking.

    // Default returns std::nullopt (not tracked). Only base/cumulative/full override.

    virtual std::optional<CompactionProfileType> profile_type() const { return std::nullopt; }

    // Accessors for tracker registration

    int64_t compaction_id() const { return _compaction_id; }

    int64_t input_rowsets_data_size() const { return _input_rowsets_data_size; }

    int64_t input_rowsets_index_size() const { return _input_rowsets_index_size; }

    int64_t input_rowsets_total_size() const { return _input_rowsets_total_size; }

    int64_t input_row_num_value() const { return _input_row_num; }

    int64_t input_rowsets_count() const { return static_cast<int64_t>(_input_rowsets.size()); }

    virtual int64_t input_segments_num_value() const { return _input_num_segments; }

    bool is_vertical() const { return _is_vertical; }

    std::string input_version_range_str() const;

    // the difference between index change compmaction and other compaction.

    // 1. delete predicate should be kept when input is cumu rowset.

    // 2. inverted compaction should be skipped.

    // 3. compute level should not be changed.

    virtual bool is_index_change_compaction() { return false; }

private:

    void set_delete_predicate_for_output_rowset();

protected:

    struct MergeInputRowsetsResult {

        bool is_segment_grouped = false;

        int64_t segment_group_size = 0;

        std::vector<int32_t> output_segment_group_sizes;

    };

    struct MergeInputRowsetsContext {

        MergeInputRowsetsResult result;

        std::vector<RowsetReaderSharedPtr> input_rs_readers;

    };

    Status merge_input_rowsets();

    Status prepare_merge_input_rowsets_execution(MergeInputRowsetsContext* context);

    Status execute_merge_input_rowsets(MergeInputRowsetsContext* context);

    Status finish_merge_input_rowsets_execution(MergeInputRowsetsContext* context);

    void snapshot_changelog_ttl();

    void snapshot_row_ttl();

    bool pick_expired_changelog_rowset(const std::vector<RowsetSharedPtr>& candidates);

    void filter_changelog_ttl_rowsets();

    // Whether the rowset is expired changelog, which this compaction drops instead of merging it.

    bool is_expired_changelog_rowset(const RowsetSharedPtr& rowset) const;

    // Counts this compaction in the changelog TTL cleanup metrics as it finishes, if it drops expired changelog.

    void record_changelog_ttl_cleanup(bool succeeded) const;

    virtual Status prepare_merge_input_rowsets(MergeInputRowsetsResult* /*result*/) {

        return Status::OK();

    }

    virtual Status do_merge_input_rowsets(

            const std::vector<RowsetReaderSharedPtr>& input_rs_readers,

            MergeInputRowsetsResult* result);

    virtual void update_output_rowset_after_build(const MergeInputRowsetsResult& /*result*/) {}

    // Maps one segment-range merge's column-group progress into the whole compaction task.

    struct VerticalMergeProgressContext {

        int64_t total_ranges;

        int64_t range_index;

    };

    Status execute_merge(const std::vector<RowsetReaderSharedPtr>& input_rs_readers,

                         int64_t merge_way_num, Merger::Statistics* stats,

                         std::optional<std::pair<int64_t, int64_t>> segment_range = std::nullopt,

                         VerticalMergeProgressContext progress = {1, 0});

    // merge inverted index files

    Status do_inverted_index_compaction();

    // mark all columns in columns_to_do_index_compaction to skip index compaction next time.

    void mark_skip_index_compaction(const RowsetWriterContext& context,

                                    const std::function<void(int64_t, int64_t)>& error_handler);

    void construct_index_compaction_columns(RowsetWriterContext& ctx);

    virtual Status construct_output_rowset_writer(RowsetWriterContext& ctx) = 0;

    Status check_correctness();

    int64_t get_avg_segment_rows();

    void init_profile(const std::string& label);

    void _load_segment_to_cache();

    int64_t merge_way_num();

    virtual Status update_delete_bitmap() = 0;

    int64_t _compaction_id {0};

    void submit_profile_record(bool success, int64_t start_time_ms,

                               const std::string& status_msg = "");

    // the root tracker for this compaction

    std::shared_ptr<MemTrackerLimiter> _mem_tracker;

    BaseTabletSPtr _tablet;

    std::vector<RowsetSharedPtr> _input_rowsets;

    std::vector<RowsetSharedPtr> _data_input_rowsets;

    int64_t _changelog_ttl_cutoff_tso {-1};

    int64_t _changelog_ttl_seconds {0};

    int64_t _changelog_ttl_visible_version {std::numeric_limits<int64_t>::max()};

    int64_t _changelog_ttl_filtered_rows {0};

    int64_t _input_rowsets_data_size {0};

    int64_t _input_rowsets_index_size {0};

    int64_t _input_rowsets_total_size {0};

    int64_t _input_row_num {0};

    int64_t _input_num_segments {0};

    int64_t _local_read_bytes_total {};

    int64_t _remote_read_bytes_total {};

    int64_t _input_rowsets_cached_data_size {0};

    int64_t _input_rowsets_cached_index_size {0};

    Merger::Statistics _stats;

    RowsetSharedPtr _output_rowset;

    std::unique_ptr<RowsetWriter> _output_rs_writer;

    enum class CompactionState : uint8_t { INITED = 0, SUCCESS = 1 };

    CompactionState _state {CompactionState::INITED};

    bool _is_vertical;

    bool _is_ordered_data_compaction {false};

    bool _trigger_quick_merge_by_changelog {false};

    bool _allow_delete_in_cumu_compaction;

    bool _enable_vertical_compact_variant_subcolumns;

    Version _output_version;

    int64_t _newest_write_timestamp {-1};

    std::unique_ptr<RowIdConversion> _rowid_conversion = nullptr;

    TabletSchemaSPtr _cur_tablet_schema;

    // The variant path layout of _cur_tablet_schema; empty unless it is an extended schema.

    VariantCompactionPathsSPtr _cur_variant_compaction_paths;

    std::unique_ptr<RuntimeProfile> _profile;

    bool _enable_inverted_index_compaction {false};

    RuntimeProfile::Counter* _input_rowsets_data_size_counter = nullptr;

    RuntimeProfile::Counter* _input_rowsets_counter = nullptr;

    RuntimeProfile::Counter* _input_row_num_counter = nullptr;

    RuntimeProfile::Counter* _input_segments_num_counter = nullptr;

    RuntimeProfile::Counter* _merged_rows_counter = nullptr;

    RuntimeProfile::Counter* _filtered_rows_counter = nullptr;

    RuntimeProfile::Counter* _output_rowset_data_size_counter = nullptr;

    RuntimeProfile::Counter* _output_row_num_counter = nullptr;

    RuntimeProfile::Counter* _output_segments_num_counter = nullptr;

    RuntimeProfile::Counter* _merge_rowsets_latency_timer = nullptr;

};

// `StorageEngine` mixin for `Compaction`

class CompactionMixin : public Compaction {

public:

    CompactionMixin(StorageEngine& engine, TabletSharedPtr tablet, const std::string& label);

    ~CompactionMixin() override;

    Status execute_compact() override;

    int64_t get_compaction_permits();

    int64_t initiator() const { return INVALID_COMPACTION_INITIATOR_ID; }

    int64_t calc_input_rowsets_total_size() const;

    int64_t calc_input_rowsets_row_num() const;

protected:

    // Convert `_tablet` from `BaseTablet` to `Tablet`

    Tablet* tablet();

    Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;

    virtual Status modify_rowsets();

    Status update_delete_bitmap() override;

    void find_longest_consecutive_version(std::vector<RowsetSharedPtr>* rowsets,

                                          std::vector<Version>* missing_version);

    StorageEngine& _engine;

private:

    Status execute_compact_impl(int64_t permits);

    Status build_basic_info(bool is_ordered_compaction = false);

    // Return true if do ordered data compaction successfully

    bool handle_ordered_data_compaction();

    Status do_compact_ordered_rowsets();

    bool _check_if_includes_input_rowsets(const RowsetIdUnorderedSet& commit_rowset_ids_set) const;

    void update_compaction_level();

    PendingRowsetGuard _pending_rs_guard;

};

class CloudCompactionMixin : public Compaction {

public:

    CloudCompactionMixin(CloudStorageEngine& engine, CloudTabletSPtr tablet,

                         const std::string& label);

    ~CloudCompactionMixin() override = default;

    Status execute_compact() override;

    int64_t initiator() const;

    int64_t num_input_rowsets() const;

protected:

    CloudTablet* cloud_tablet() { return static_cast<CloudTablet*>(_tablet.get()); }

    Status update_delete_bitmap() override;

    virtual Status garbage_collection();

    Status construct_output_rowset_writer(RowsetWriterContext& ctx) override;

    // Helper function to apply truncation and log the result

    // Returns the number of rowsets that were truncated

    size_t apply_txn_size_truncation_and_log(const std::string& compaction_name);

    // Caller must hold the tablet header lock.

    bool should_apply_cumulative_compaction_result(int64_t response_cumulative_compaction_cnt);

    Status prepare_execute_compact(int64_t permits);

    Status finish_execute_compact(int64_t execution_start_time_us);

    int64_t get_compaction_permits();

    CloudStorageEngine& _engine;

    std::string _uuid;

    int64_t _expiration = 0;

    virtual Status rebuild_tablet_schema() { return Status::OK(); }

private:

    Status set_storage_resource_from_input_rowsets(RowsetWriterContext& ctx);

    Status execute_compact_impl(int64_t permits);

    Status build_basic_info();

    virtual Status modify_rowsets();

    void update_compaction_level();

    bool should_cache_compaction_output();

};

namespace cloud {

// Truncate rowsets based on estimated transaction metadata size

// Returns the number of rowsets that were truncated (removed from the end)

// Only considers rowset meta size (using doris_rowset_meta_to_cloud for estimation)

size_t truncate_rowsets_by_txn_size(std::vector<RowsetSharedPtr>& rowsets, int64_t& kept_size_bytes,

                                    int64_t& truncated_size_bytes);

} // namespace cloud

} // namespace doris