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

#include <cstdint>

#include <deque>

#include <shared_mutex>

#include <string>

#include <unordered_map>

#include <vector>

#include "storage/compaction_task_types.h"

namespace doris {

// Standardized compaction type enum covering base / cumulative / full.

enum class CompactionProfileType : uint8_t {

    BASE = 0,

    CUMULATIVE = 1,

    FULL = 2,

};

const char* to_string(CompactionProfileType type);

// Task lifecycle status.

enum class CompactionTaskStatus : uint8_t {

    PENDING = 0,

    RUNNING = 1,

    FINISHED = 2,

    FAILED = 3,

};

const char* to_string(CompactionTaskStatus status);

const char* to_string(TriggerMethod method);

// Incremental info when transitioning from PENDING to RUNNING.

struct RunningStats {

    int64_t start_time_ms {0};

    bool is_vertical {false};

    int64_t permits {0};

};

// Result info collected when a task completes or fails.

struct CompletionStats {

    // Input stats backfill: local compaction populates these in build_basic_info()

    // which runs inside execute_compact_impl(), after register_task().

    std::string input_version_range;

    int64_t input_rowsets_count {0};

    int64_t input_row_num {0};

    int64_t input_data_size {0};

    int64_t input_index_size {0};

    int64_t input_total_size {0};

    int64_t input_segments_num {0};

    int64_t end_time_ms {0};

    int64_t merged_rows {0};

    int64_t filtered_rows {0};

    int64_t output_rows {0}; // _stats.output_rows (Merger statistics)

    int64_t output_row_num {0};

    int64_t output_data_size {0};

    int64_t output_index_size {0}; // _output_rowset->index_disk_size()

    int64_t output_total_size {0}; // _output_rowset->total_disk_size()

    int64_t output_segments_num {0};

    std::string output_version;

    bool is_ordered_data_compaction {false};

    int64_t merge_latency_ms {0}; // _merge_rowsets_latency_timer (converted to ms)

    int64_t bytes_read_from_local {0};

    int64_t bytes_read_from_remote {0};

    int64_t peak_memory_bytes {0};

};

// Unified metadata describing a compaction task across its full lifecycle.

struct CompactionTaskInfo {

    // ===== Identity =====

    int64_t compaction_id {0}; // unique task ID assigned by Tracker

    int64_t backend_id {0};    // BE node ID

    int64_t table_id {0};      // table ID

    int64_t partition_id {0};  // partition ID

    int64_t tablet_id {0};     // tablet ID

    // ===== Task attributes =====

    CompactionProfileType compaction_type {CompactionProfileType::BASE};

    CompactionTaskStatus status {CompactionTaskStatus::PENDING};

    TriggerMethod trigger_method {TriggerMethod::AUTO};

    int64_t compaction_score {0}; // tablet compaction score at register time

    // ===== Timestamps =====

    int64_t scheduled_time_ms {0}; // task registration time

    int64_t start_time_ms {0};     // task execution start time (0 while PENDING)

    int64_t end_time_ms {0};       // task end time (0 while not completed)

    // ===== Input statistics (available after prepare_compact) =====

    int64_t input_rowsets_count {0};

    int64_t input_row_num {0};

    int64_t input_data_size {0};  // bytes, corresponds to _input_rowsets_data_size

    int64_t input_index_size {0}; // bytes, corresponds to _input_rowsets_index_size

    int64_t input_total_size {0}; // bytes, = data + index

    int64_t input_segments_num {0};

    std::string input_version_range; // e.g. "[0-5]"

    // ===== Output statistics (written at complete/fail) =====

    int64_t merged_rows {0};

    int64_t filtered_rows {0};

    int64_t output_rows {0};       // Merger output rows (_stats.output_rows; 0 for ordered path)

    int64_t output_row_num {0};    // from _output_rowset->num_rows()

    int64_t output_data_size {0};  // bytes, from _output_rowset->data_disk_size()

    int64_t output_index_size {0}; // bytes, from _output_rowset->index_disk_size()

    int64_t output_total_size {0}; // bytes, from _output_rowset->total_disk_size()

    int64_t output_segments_num {0};

    std::string output_version; // e.g. "[0-5]"

    bool is_ordered_data_compaction {

            false}; // ordered/meta-link compaction path, including changelog quick path

    // ===== Merge performance =====

    int64_t merge_latency_ms {0}; // merge rowsets latency (ms; 0 for ordered path)

    // ===== IO statistics (written at complete/fail) =====

    int64_t bytes_read_from_local {0};

    int64_t bytes_read_from_remote {0};

    // ===== Resources =====

    int64_t peak_memory_bytes {0}; // peak memory usage (bytes)

    bool is_vertical {false};      // whether vertical merge is used

    int64_t permits {0};           // compaction permits used

    // ===== Vertical compaction progress =====

    int64_t vertical_total_groups {0}; // total column groups (0 for horizontal)

    int64_t vertical_completed_groups {

            0}; // completed column groups (updated in real-time during RUNNING)

    // ===== Error =====

    std::string status_msg; // failure message (empty on success)

};

// Global singleton managing compaction task lifecycle.

// Receives push reports from compaction entries and execution layer,

// provides pull query interfaces for system table and HTTP API.

class CompactionTaskTracker {

public:

    static CompactionTaskTracker* instance();

    // ID allocation: globally unique monotonically increasing, restarts from 1 after BE restart.

    int64_t next_compaction_id() { return _next_id.fetch_add(1, std::memory_order_relaxed); }

    // ===== Push interfaces: lifecycle management (write lock) =====

    // All push interfaces are no-op when enable_compaction_task_tracker=false.

    void register_task(CompactionTaskInfo info);

    void update_to_running(int64_t compaction_id, const RunningStats& stats);

    void update_progress(int64_t compaction_id, int64_t total_groups, int64_t completed_groups);

    void complete(int64_t compaction_id, const CompletionStats& stats);

    void fail(int64_t compaction_id, const CompletionStats& stats, const std::string& msg);

    void remove_task(int64_t compaction_id);

    // ===== Pull interfaces: queries (read lock) =====

    // For system table: returns full snapshot copy of _active_tasks + _completed_tasks.

    std::vector<CompactionTaskInfo> get_all_tasks() const;

    // For HTTP API: iterates _completed_tasks only, returns filtered subset copy.

    std::vector<CompactionTaskInfo> get_completed_tasks(int64_t tablet_id = 0, int64_t top_n = 0,

                                                        const std::string& compaction_type = "",

                                                        int success_filter = -1) const;

    // Test only: clear all active and completed tasks.

    void clear_for_test();

private:

    CompactionTaskTracker() = default;

    void _apply_completion(CompactionTaskInfo& info, const CompletionStats& stats);

    void _trim_completed_locked();

    std::atomic<int64_t> _next_id {1};

    mutable std::shared_mutex _mutex;

    // Active tasks (PENDING + RUNNING), indexed by compaction_id.

    // Removed on complete/fail and moved to _completed_tasks.

    std::unordered_map<int64_t, CompactionTaskInfo> _active_tasks;

    // Completed tasks (FINISHED + FAILED), FIFO ring buffer.

    // Oldest records are evicted when exceeding compaction_task_tracker_max_records.

    std::deque<CompactionTaskInfo> _completed_tasks;

};

} // namespace doris