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

#include <algorithm>

#include <atomic>

#include <functional>

#include <optional>

#include <utility>

#include <vector>

#include "common/status.h"

#include "core/block/block.h"

#include "exec/scan/task_executor/listenable_future.h"

#include "runtime/exec_env.h"

#include "runtime/runtime_state.h"

#include "storage/tablet/tablet.h"

#include "util/stopwatch.hpp"

namespace doris {

class RuntimeProfile;

class TupleDescriptor;

class VExprContext;

class ScanLocalStateBase;

} // namespace doris

namespace doris {

// Counter for load

struct ScannerCounter {

    ScannerCounter() : num_rows_filtered(0), num_rows_unselected(0) {}

    int64_t num_rows_filtered;   // unqualified rows (unmatched the dest schema, or no partition)

    int64_t num_rows_unselected; // rows filtered by predicates

};

class Scanner {

public:

    Scanner(RuntimeState* state, ScanLocalStateBase* local_state, int64_t limit,

            RuntimeProfile* profile);

    //only used for FileScanner read one line.

    Scanner(RuntimeState* state, RuntimeProfile* profile)

            : _state(state), _limit(1), _profile(profile), _total_rf_num(0), _has_prepared(false) {

        DorisMetrics::instance()->scanner_cnt->increment(1);

    };

    virtual ~Scanner() {

        SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_state->query_mem_tracker());

        _input_block.clear();

        _conjuncts.clear();

        _projections.clear();

        _origin_block.clear();

        _common_expr_ctxs_push_down.clear();

        DorisMetrics::instance()->scanner_cnt->increment(-1);

    }

    virtual Status init(RuntimeState* state, const VExprContextSPtrs& conjuncts);

    Status prepare() {

        SCOPED_RAW_TIMER(&_per_scanner_timer);

        SCOPED_RAW_TIMER(&_per_scanner_prepare_timer);

        return _prepare_impl();

    }

    Status open(RuntimeState* state) {

        SCOPED_RAW_TIMER(&_per_scanner_timer);

        SCOPED_RAW_TIMER(&_per_scanner_open_timer);

        return _open_impl(state);

    }

    Status get_block(RuntimeState* state, Block* block, bool* eos);

    Status get_block_after_projects(RuntimeState* state, Block* block, bool* eos);

    virtual Status close(RuntimeState* state);

    // Try to stop scanner, and all running readers.

    virtual void try_stop() { _should_stop = true; };

    // What the turn that just ended without a block waits for, if anything: get_block() returned no

    // rows and no eof because the scanner cannot read on until this future is done - for now, the

    // JNI heap gate admitting the Java scanner of the split it is on. The scheduler parks the scanner

    // instead of keeping a worker meanwhile, and runs it again once the future is done.

    std::optional<SharedListenableFuture<Void>> take_waiting_for() {

        return std::exchange(_waiting_for, std::nullopt);

    }

    virtual std::string get_name() { return ""; }

    // return the readable name of current scan range.

    // eg, for file scanner, return the current file path.

    virtual std::string get_current_scan_range_name() { return "not implemented"; }

#ifdef BE_TEST

    static uint64_t TEST_build_condition_cache_digest(uint64_t seed,

                                                      const VExprContextSPtrs& conjuncts);

#endif

protected:

    // Rebuild the condition-cache digest from the scanner's current conjunct snapshot. The local

    // state's digest is used only as a safety gate: zero means condition cache was disabled during

    // scan-node open (for example by TopN or an expression without a reliable digest).

    uint64_t _current_condition_cache_digest() const;

    static uint64_t _build_condition_cache_digest(uint64_t seed,

                                                  const VExprContextSPtrs& conjuncts);

    virtual Status _prepare_impl() {

        _has_prepared = true;

        return Status::OK();

    }

    virtual Status _open_impl(RuntimeState* state) {

        _block_avg_bytes = state->batch_size() * 8;

        return Status::OK();

    }

    // Subclass should implement this to return data.

    virtual Status _get_block_impl(RuntimeState* state, Block* block, bool* eof) = 0;

    virtual bool _can_merge_padding_blocks(const Block& /*left*/, const Block& /*right*/) const {

        return true;

    }

    Status _merge_padding_block() {

        if (_padding_block.empty()) {

            _padding_block.swap(_origin_block);

        } else if (_origin_block.rows()) {

            ScopedMutableBlock scoped_mutable_block(&_padding_block);

            auto& mutable_block = scoped_mutable_block.mutable_block();

            RETURN_IF_ERROR(mutable_block.merge(_origin_block));

        }

        return Status::OK();

    }

    // Update the counters before closing this scanner

    virtual void _collect_profile_before_close();

    // Whether rows filtered/unselected by this scanner should be reported to the load

    // counters in RuntimeState. Only the scanner reading the load source data should

    // report, otherwise rows filtered by query predicates (e.g. in INSERT INTO ... SELECT

    // or DELETE FROM ... WHERE) would be mixed into load counters and make

    // num_rows_load_success() negative.

    virtual bool _should_update_load_counters() const { return _is_load; }

    // Check if scanner is already closed, if not, mark it as closed.

    // Returns true if the scanner was successfully marked as closed (first time).

    // Returns false if the scanner was already closed.

    bool _try_close();

    // Filter the output block finally.

    virtual Status _filter_output_block(Block* block);

    Status _do_projections(Block* origin_block, Block* output_block);

private:

    void _start_scan_cpu_timer() {

        _cpu_watch.reset();

        _cpu_watch.start();

    }

    void _update_wait_worker_timer() { _scanner_wait_worker_timer += _watch.elapsed_time(); }

    void _update_scan_cpu_timer();

public:

    // Call start_wait_worker_timer() when submit the scanner to the thread pool.

    // And call update_wait_worker_timer() when it is actually being executed.

    void start_wait_worker_timer() {

        _watch.reset();

        _watch.start();

    }

    void resume() {

        _update_wait_worker_timer();

        _start_scan_cpu_timer();

    }

    void pause() {

        _update_scan_cpu_timer();

        start_wait_worker_timer();

    }

    int64_t get_time_cost_ns() const { return _per_scanner_timer; }

    int64_t get_prepare_time_cost_ns() const { return _per_scanner_prepare_timer; }

    int64_t get_open_time_cost_ns() const { return _per_scanner_open_timer; }

    int64_t projection_time() const { return _projection_timer; }

    int64_t get_rows_read() const { return _num_rows_read; }

    bool has_prepared() const { return _has_prepared; }

    Status try_append_late_arrival_runtime_filter();

    int64_t get_scanner_wait_worker_timer() const { return _scanner_wait_worker_timer; }

    // Some counters need to be updated realtime, for example, workload group policy need

    // scan bytes to cancel the query exceed limit.

    virtual void update_realtime_counters() {}

    RuntimeState* runtime_state() { return _state; }

    bool is_open() const { return _is_open; }

    void set_opened() { _is_open = true; }

    virtual doris::TabletStorageType get_storage_type() {

        return doris::TabletStorageType::STORAGE_TYPE_REMOTE;

    }

    // Returns true if this scanner's scan range has been pruned by a runtime filter.

    virtual bool is_pruned_by_runtime_filter() const { return false; }

    // Releases resources owned by a scanner that runtime-filter pruning makes unnecessary before

    // open(). The scanner will not be scheduled again after this call.

    virtual void release_unopened_resources() {

        DORIS_CHECK(!_is_open);

        _has_prepared = false;

    }

    bool need_to_close() const { return _need_to_close; }

    void mark_to_need_to_close() {

        // If the scanner is failed during init or open, then not need update counters

        // because the query is fail and the counter is useless. And it may core during

        // update counters. For example, update counters depend on scanner's tablet, but

        // the tablet == null when init failed.

        if (_is_open) {

            _collect_profile_before_close();

        }

        _need_to_close = true;

    }

    void set_status_on_failure(const Status& st) { _status = st; }

    int64_t limit() const { return _limit; }

    auto get_block_avg_bytes() const { return _block_avg_bytes; }

    void update_block_avg_bytes(size_t block_avg_bytes) { _block_avg_bytes = block_avg_bytes; }

protected:

    RuntimeState* _state = nullptr;

    ScanLocalStateBase* _local_state = nullptr;

    // Set if scan node has sort limit info

    int64_t _limit = -1;

    RuntimeProfile* _profile = nullptr;

    const TupleDescriptor* _output_tuple_desc = nullptr;

    std::optional<std::reference_wrapper<const RowDescriptor>> _projection_output_row_descriptor;

    bool _has_projection = false;

    // If _input_tuple_desc is set, the scanner will read data into

    // this _input_block first, then convert to the output block.

    Block _input_block;

    bool _is_open = false;

    std::atomic<bool> _is_closed {false};

    bool _need_to_close = false;

    Status _status;

    // If _applied_rf_num == _total_rf_num

    // means all runtime filters are arrived and applied.

    int _applied_rf_num = 0;

    int _total_rf_num = 0;

    // Cloned from _conjuncts of scan node.

    // It includes predicate in SQL and runtime filters.

    VExprContextSPtrs _conjuncts;

    VExprContextSPtrs _projections;

    // Used in common subexpression elimination to compute intermediate results.

    std::vector<VExprContextSPtrs> _intermediate_projections;

    Block _origin_block;

    Block _padding_block;

    VExprContextSPtrs _common_expr_ctxs_push_down;

    // num of rows read from scanner

    int64_t _num_rows_read = 0;

    int64_t _num_byte_read = 0;

    // num of rows return from scanner, after filter block

    int64_t _num_rows_return = 0;

    size_t _block_avg_bytes = 0;

    // Set true after counter is updated finally

    bool _has_updated_counter = false;

    // watch to count the time wait for scanner thread

    MonotonicStopWatch _watch;

    // Do not use ScopedTimer. There is no guarantee that, the counter

    ThreadCpuStopWatch _cpu_watch;

    int64_t _scanner_wait_worker_timer = 0;

    int64_t _scan_cpu_timer = 0;

    bool _is_load = false;

    bool _has_prepared = false;

    ScannerCounter _counter;

    int64_t _per_scanner_timer = 0;

    int64_t _per_scanner_prepare_timer = 0;

    int64_t _per_scanner_open_timer = 0;

    int64_t _projection_timer = 0;

    bool _should_stop = false;

    // Set by _get_block_impl() when it returns without a block because the scanner has to wait; see

    // take_waiting_for().

    std::optional<SharedListenableFuture<Void>> _waiting_for;

    // Cached pointer to ScanOperator's remaining-limit counter. Null when

    // this scanner is on the topn path or the query has no LIMIT.

    std::atomic<int64_t>* _shared_scan_limit = nullptr;

};

using ScannerSPtr = std::shared_ptr<Scanner>;

} // namespace doris