be/src/olap/column_predicate.h

// 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 <memory>
#include <roaring/roaring.hh>

#include "common/exception.h"
#include "olap/rowset/segment_v2/bloom_filter.h"
#include "olap/rowset/segment_v2/inverted_index_iterator.h"
#include "runtime/define_primitive_type.h"
#include "util/defer_op.h"
#include "util/runtime_profile.h"
#include "vec/columns/column.h"
#include "vec/exec/format/parquet/parquet_predicate.h"
#include "vec/exprs/vruntimefilter_wrapper.h"

using namespace doris::segment_v2;

namespace doris {

enum class PredicateType {
    UNKNOWN = 0,
    EQ = 1,
    NE = 2,
    LT = 3,
    LE = 4,
    GT = 5,
    GE = 6,
    IN_LIST = 7,
    NOT_IN_LIST = 8,
    IS_NULL = 9,
    IS_NOT_NULL = 10,
    BF = 11,            // BloomFilter
    BITMAP_FILTER = 12, // BitmapFilter
    MATCH = 13,         // fulltext match
};

template <PrimitiveType primitive_type, typename ResultType>
ResultType get_zone_map_value(void* data_ptr) {
    ResultType res;
    // DecimalV2's storage value is different from predicate or compute value type
    // need convert it to DecimalV2Value
    if constexpr (primitive_type == PrimitiveType::TYPE_DECIMALV2) {
        decimal12_t decimal_12_t_value;
        memcpy((char*)(&decimal_12_t_value), data_ptr, sizeof(decimal12_t));
        res.from_olap_decimal(decimal_12_t_value.integer, decimal_12_t_value.fraction);
    } else if constexpr (primitive_type == PrimitiveType::TYPE_DATE) {
        static_assert(std::is_same_v<ResultType, VecDateTimeValue>);
        uint24_t date;
        memcpy(&date, data_ptr, sizeof(uint24_t));
        res.from_olap_date(date);
    } else if constexpr (primitive_type == PrimitiveType::TYPE_DATETIME) {
        static_assert(std::is_same_v<ResultType, VecDateTimeValue>);
        uint64_t datetime;
        memcpy(&datetime, data_ptr, sizeof(uint64_t));
        res.from_olap_datetime(datetime);
    } else {
        memcpy(reinterpret_cast<void*>(&res), data_ptr, sizeof(ResultType));
    }
    return res;
}

inline std::string type_to_string(PredicateType type) {
    switch (type) {
    case PredicateType::UNKNOWN:
        return "UNKNOWN";

    case PredicateType::EQ:
        return "EQ";

    case PredicateType::NE:
        return "NE";

    case PredicateType::LT:
        return "LT";

    case PredicateType::LE:
        return "LE";

    case PredicateType::GT:
        return "GT";

    case PredicateType::GE:
        return "GE";

    case PredicateType::IN_LIST:
        return "IN_LIST";

    case PredicateType::NOT_IN_LIST:
        return "NOT_IN_LIST";

    case PredicateType::IS_NULL:
        return "IS_NULL";

    case PredicateType::IS_NOT_NULL:
        return "IS_NOT_NULL";

    case PredicateType::BF:
        return "BF";
    default:
        return "";
    };

    return "";
}

inline std::string type_to_op_str(PredicateType type) {
    switch (type) {
    case PredicateType::EQ:
        return "=";

    case PredicateType::NE:
        return "!=";

    case PredicateType::LT:
        return "<<";

    case PredicateType::LE:
        return "<=";

    case PredicateType::GT:
        return ">>";

    case PredicateType::GE:
        return ">=";

    case PredicateType::IN_LIST:
        return "*=";

    case PredicateType::NOT_IN_LIST:
        return "!*=";

    case PredicateType::IS_NULL:
    case PredicateType::IS_NOT_NULL:
        return "is";

    default:
        break;
    };

    return "";
}

struct PredicateTypeTraits {
    static constexpr bool is_range(PredicateType type) {
        return (type == PredicateType::LT || type == PredicateType::LE ||
                type == PredicateType::GT || type == PredicateType::GE);
    }

    static constexpr bool is_bloom_filter(PredicateType type) { return type == PredicateType::BF; }

    static constexpr bool is_list(PredicateType type) {
        return (type == PredicateType::IN_LIST || type == PredicateType::NOT_IN_LIST);
    }

    static constexpr bool is_equal_or_list(PredicateType type) {
        return (type == PredicateType::EQ || type == PredicateType::IN_LIST);
    }

    static constexpr bool is_comparison(PredicateType type) {
        return (type == PredicateType::EQ || type == PredicateType::NE ||
                type == PredicateType::LT || type == PredicateType::LE ||
                type == PredicateType::GT || type == PredicateType::GE);
    }
};

#define EVALUATE_BY_SELECTOR(EVALUATE_IMPL_WITH_NULL_MAP, EVALUATE_IMPL_WITHOUT_NULL_MAP) \
    const bool is_dense_column = pred_col.size() == size;                                 \
    for (uint16_t i = 0; i < size; i++) {                                                 \
        uint16_t idx = is_dense_column ? i : sel[i];                                      \
        if constexpr (is_nullable) {                                                      \
            if (EVALUATE_IMPL_WITH_NULL_MAP(idx)) {                                       \
                sel[new_size++] = idx;                                                    \
            }                                                                             \
        } else {                                                                          \
            if (EVALUATE_IMPL_WITHOUT_NULL_MAP(idx)) {                                    \
                sel[new_size++] = idx;                                                    \
            }                                                                             \
        }                                                                                 \
    }

class ColumnPredicate : public std::enable_shared_from_this<ColumnPredicate> {
public:
    explicit ColumnPredicate(uint32_t column_id, PrimitiveType primitive_type,
                             bool opposite = false)
            : _column_id(column_id), _primitive_type(primitive_type), _opposite(opposite) {
        reset_judge_selectivity();
    }
    ColumnPredicate(const ColumnPredicate& other, uint32_t col_id) : ColumnPredicate(other) {
        _column_id = col_id;
    }

    virtual ~ColumnPredicate() = default;

    virtual PredicateType type() const = 0;
    virtual PrimitiveType primitive_type() const { return _primitive_type; }
    virtual std::shared_ptr<ColumnPredicate> clone(uint32_t col_id) const = 0;

    //evaluate predicate on inverted
    virtual Status evaluate(const vectorized::IndexFieldNameAndTypePair& name_with_type,
                            IndexIterator* iterator, uint32_t num_rows,
                            roaring::Roaring* bitmap) const {
        return Status::NotSupported(
                "Not Implemented evaluate with inverted index, please check the predicate");
    }

    virtual double get_ignore_threshold() const { return 0; }
    // Return the size of value set for IN/NOT IN predicates and 0 for others.
    virtual std::string debug_string() const {
        fmt::memory_buffer debug_string_buffer;
        fmt::format_to(debug_string_buffer,
                       "Column ID: {}, Data Type: {}, PredicateType: {}, opposite: {}, Runtime "
                       "Filter ID: {}",
                       _column_id, type_to_string(primitive_type()), pred_type_string(type()),
                       _opposite, _runtime_filter_id);
        return fmt::to_string(debug_string_buffer);
    }

    // evaluate predicate on IColumn
    // a short circuit eval way
    uint16_t evaluate(const vectorized::IColumn& column, uint16_t* sel, uint16_t size) const {
        Defer defer([&] { try_reset_judge_selectivity(); });

        if (always_true()) {
            update_filter_info(0, 0, size);
            return size;
        }

        uint16_t new_size = _evaluate_inner(column, sel, size);
        if (_can_ignore()) {
            do_judge_selectivity(size - new_size, size);
        }
        update_filter_info(size - new_size, size, 0);
        return new_size;
    }
    virtual void evaluate_and(const vectorized::IColumn& column, const uint16_t* sel, uint16_t size,
                              bool* flags) const {}
    virtual void evaluate_or(const vectorized::IColumn& column, const uint16_t* sel, uint16_t size,
                             bool* flags) const {}

    virtual bool support_zonemap() const { return true; }

    virtual bool evaluate_and(const std::pair<WrapperField*, WrapperField*>& statistic) const {
        return true;
    }

    virtual bool is_always_true(const std::pair<WrapperField*, WrapperField*>& statistic) const {
        return false;
    }

    virtual bool evaluate_del(const std::pair<WrapperField*, WrapperField*>& statistic) const {
        return false;
    }

    virtual bool evaluate_and(const vectorized::ParquetBlockSplitBloomFilter* bf) const {
        return true;
    }

    virtual bool evaluate_and(const BloomFilter* bf) const { return true; }

    virtual bool evaluate_and(const StringRef* dict_words, const size_t dict_count) const {
        return true;
    }

    virtual bool can_do_bloom_filter(bool ngram) const { return false; }

    /**
     * Figure out whether this page is matched partially or completely.
     */
    virtual bool evaluate_and(vectorized::ParquetPredicate::ColumnStat* statistic) const {
        throw Exception(ErrorCode::INTERNAL_ERROR,
                        "ParquetPredicate is not supported by this predicate!");
        return true;
    }

    virtual bool evaluate_and(vectorized::ParquetPredicate::CachedPageIndexStat* statistic,
                              RowRanges* row_ranges) const {
        throw Exception(ErrorCode::INTERNAL_ERROR,
                        "ParquetPredicate is not supported by this predicate!");
        return true;
    }

    // used to evaluate pre read column in lazy materialization
    // now only support integer/float
    // a vectorized eval way
    virtual void evaluate_vec(const vectorized::IColumn& column, uint16_t size, bool* flags) const {
        DCHECK(false) << "should not reach here";
    }
    virtual void evaluate_and_vec(const vectorized::IColumn& column, uint16_t size,
                                  bool* flags) const {
        DCHECK(false) << "should not reach here";
    }

    virtual std::string get_search_str() const {
        DCHECK(false) << "should not reach here";
        return "";
    }

    virtual void set_page_ng_bf(std::unique_ptr<segment_v2::BloomFilter>) {
        DCHECK(false) << "should not reach here";
    }
    uint32_t column_id() const { return _column_id; }

    bool opposite() const { return _opposite; }

    void attach_profile_counter(
            int filter_id, std::shared_ptr<RuntimeProfile::Counter> predicate_filtered_rows_counter,
            std::shared_ptr<RuntimeProfile::Counter> predicate_input_rows_counter,
            std::shared_ptr<RuntimeProfile::Counter> predicate_always_true_rows_counter) {
        _runtime_filter_id = filter_id;
        DCHECK(predicate_filtered_rows_counter != nullptr);
        DCHECK(predicate_input_rows_counter != nullptr);

        if (predicate_filtered_rows_counter != nullptr) {
            _predicate_filtered_rows_counter = predicate_filtered_rows_counter;
        }
        if (predicate_input_rows_counter != nullptr) {
            _predicate_input_rows_counter = predicate_input_rows_counter;
        }
        if (predicate_always_true_rows_counter != nullptr) {
            _predicate_always_true_rows_counter = predicate_always_true_rows_counter;
        }
    }

    /// TODO: Currently we only record statistics for runtime filters, in the future we should record for all predicates
    void update_filter_info(int64_t filter_rows, int64_t input_rows,
                            int64_t always_true_rows) const {
        if (_predicate_input_rows_counter == nullptr ||
            _predicate_filtered_rows_counter == nullptr ||
            _predicate_always_true_rows_counter == nullptr) {
            throw Exception(INTERNAL_ERROR, "Predicate profile counters are not initialized");
        }
        COUNTER_UPDATE(_predicate_input_rows_counter, input_rows);
        COUNTER_UPDATE(_predicate_filtered_rows_counter, filter_rows);
        COUNTER_UPDATE(_predicate_always_true_rows_counter, always_true_rows);
    }

    static std::string pred_type_string(PredicateType type) {
        switch (type) {
        case PredicateType::EQ:
            return "eq";
        case PredicateType::NE:
            return "ne";
        case PredicateType::LT:
            return "lt";
        case PredicateType::LE:
            return "le";
        case PredicateType::GT:
            return "gt";
        case PredicateType::GE:
            return "ge";
        case PredicateType::IN_LIST:
            return "in";
        case PredicateType::NOT_IN_LIST:
            return "not_in";
        case PredicateType::IS_NULL:
            return "is_null";
        case PredicateType::IS_NOT_NULL:
            return "is_not_null";
        case PredicateType::BF:
            return "bf";
        case PredicateType::MATCH:
            return "match";
        default:
            return "unknown";
        }
    }

    bool always_true() const { return _always_true; }
    // Return whether the ColumnPredicate was created by a runtime filter.
    // If true, it was definitely created by a runtime filter.
    // If false, it may still have been created by a runtime filter,
    // as certain filters like "in filter" generate key ranges instead of ColumnPredicate.
    // is_runtime_filter uses _can_ignore, except for BitmapFilter,
    // as BitmapFilter cannot ignore data.
    virtual bool is_runtime_filter() const { return _can_ignore(); }

protected:
    virtual bool _can_ignore() const { return _runtime_filter_id != -1; }
    virtual uint16_t _evaluate_inner(const vectorized::IColumn& column, uint16_t* sel,
                                     uint16_t size) const {
        throw Exception(INTERNAL_ERROR, "Not Implemented _evaluate_inner");
    }

    void reset_judge_selectivity() const {
        _always_true = false;
        _judge_counter = config::runtime_filter_sampling_frequency;
        _judge_input_rows = 0;
        _judge_filter_rows = 0;
    }

    void try_reset_judge_selectivity() const {
        if (_can_ignore() && ((_judge_counter--) == 0)) {
            reset_judge_selectivity();
        }
    }

    void do_judge_selectivity(uint64_t filter_rows, uint64_t input_rows) const {
        if (!_always_true) {
            _judge_filter_rows += filter_rows;
            _judge_input_rows += input_rows;
            vectorized::VRuntimeFilterWrapper::judge_selectivity(
                    get_ignore_threshold(), _judge_filter_rows, _judge_input_rows, _always_true);
        }
    }

    uint32_t _column_id;
    PrimitiveType _primitive_type;
    // TODO: the value is only in delete condition, better be template value
    bool _opposite;
    int _runtime_filter_id = -1;
    // VRuntimeFilterWrapper and ColumnPredicate share the same logic,
    // but it's challenging to unify them, so the code is duplicated.
    // _judge_counter, _judge_input_rows, _judge_filter_rows, and _always_true
    // are variables used to implement the _always_true logic, calculated periodically
    // based on runtime_filter_sampling_frequency. During each period, if _always_true
    // is evaluated as true, the logic for always_true is applied for the rest of that period
    // without recalculating. At the beginning of the next period,
    // reset_judge_selectivity is used to reset these variables.
    mutable int _judge_counter = 0;
    mutable uint64_t _judge_input_rows = 0;
    mutable uint64_t _judge_filter_rows = 0;
    mutable bool _always_true = false;

    std::shared_ptr<RuntimeProfile::Counter> _predicate_filtered_rows_counter =
            std::make_shared<RuntimeProfile::Counter>(TUnit::UNIT, 0);
    std::shared_ptr<RuntimeProfile::Counter> _predicate_input_rows_counter =
            std::make_shared<RuntimeProfile::Counter>(TUnit::UNIT, 0);
    std::shared_ptr<RuntimeProfile::Counter> _predicate_always_true_rows_counter =
            std::make_shared<RuntimeProfile::Counter>(TUnit::UNIT, 0);

private:
    ColumnPredicate(const ColumnPredicate& other) = default;
};

} //namespace doris