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

#include <string.h>

#include <roaring/roaring.hh>

#include "olap/rowset/segment_v2/bitmap_index_reader.h"

#include "olap/rowset/segment_v2/inverted_index_cache.h"

#include "olap/rowset/segment_v2/inverted_index_reader.h"

#include "vec/columns/column.h"

#include "vec/columns/column_nullable.h"

#include "vec/runtime/vdatetime_value.h"

using namespace doris::vectorized;

namespace doris {

NullPredicate::NullPredicate(uint32_t column_id, bool is_null, bool opposite)

        : ColumnPredicate(column_id), _is_null(opposite != is_null) {}

PredicateType NullPredicate::type() const {

    return _is_null ? PredicateType::IS_NULL : PredicateType::IS_NOT_NULL;

}

Status NullPredicate::evaluate(BitmapIndexIterator* iterator, uint32_t num_rows,

                               roaring::Roaring* roaring) const {

    if (iterator != nullptr) {

        roaring::Roaring null_bitmap;

        RETURN_IF_ERROR(iterator->read_null_bitmap(&null_bitmap));

        if (_is_null) {

            *roaring &= null_bitmap;

        } else {

            *roaring -= null_bitmap;

        }

    }

    return Status::OK();

}

Status NullPredicate::evaluate(const vectorized::IndexFieldNameAndTypePair& name_with_type,

                               InvertedIndexIterator* iterator, uint32_t num_rows,

                               roaring::Roaring* bitmap) const {

    if (iterator->has_null()) {

        InvertedIndexQueryCacheHandle null_bitmap_cache_handle;

        RETURN_IF_ERROR(iterator->read_null_bitmap(&null_bitmap_cache_handle));

        std::shared_ptr<roaring::Roaring> null_bitmap = null_bitmap_cache_handle.get_bitmap();

        if (null_bitmap) {

            if (_is_null) {

                *bitmap &= *null_bitmap; // Keep only nulls in bitmap if _is_null is true

            } else {

                *bitmap -= *null_bitmap; // Remove nulls from bitmap if _is_null is false

            }

        } else if (_is_null) {

            *bitmap = roaring::Roaring(); // Reset bitmap to an empty bitmap

        }

    } else if (_is_null) {

        *bitmap = roaring::Roaring(); // Reset bitmap to an empty bitmap

    }

    return Status::OK();

}

uint16_t NullPredicate::_evaluate_inner(const vectorized::IColumn& column, uint16_t* sel,

                                        uint16_t size) const {

    uint16_t new_size = 0;

    if (auto* nullable = check_and_get_column<ColumnNullable>(column)) {

        if (!nullable->has_null()) {

            return _is_null ? 0 : size;

        }

        auto& pred_col = nullable->get_null_map_data();

        constexpr bool is_nullable = true;

#define EVALUATE_WITH_NULL_IMPL(IDX) pred_col[IDX] == _is_null

#define EVALUATE_WITHOUT_NULL_IMPL(IDX) true

        EVALUATE_BY_SELECTOR(EVALUATE_WITH_NULL_IMPL, EVALUATE_WITHOUT_NULL_IMPL)

#undef EVALUATE_WITH_NULL_IMPL

#undef EVALUATE_WITHOUT_NULL_IMPL

        return new_size;

    } else {

        if (_is_null) return 0;

    }

    return size;

}

void NullPredicate::evaluate_or(const IColumn& column, const uint16_t* sel, uint16_t size,

                                bool* flags) const {

    if (auto* nullable = check_and_get_column<ColumnNullable>(column)) {

        if (!nullable->has_null()) {

            if (!_is_null) {

                memset(flags, true, size);

            }

        } else {

            auto& null_map = nullable->get_null_map_data();

            for (uint16_t i = 0; i < size; ++i) {

                if (flags[i]) continue;

                uint16_t idx = sel[i];

                flags[i] |= (null_map[idx] == _is_null);

            }

        }

    } else {

        if (!_is_null) memset(flags, true, size);

    }

}

void NullPredicate::evaluate_and(const IColumn& column, const uint16_t* sel, uint16_t size,

                                 bool* flags) const {

    if (auto* nullable = check_and_get_column<ColumnNullable>(column)) {

        if (!nullable->has_null()) {

            if (_is_null) {

                memset(flags, false, size);

            }

        } else {

            auto& null_map = nullable->get_null_map_data();

            for (uint16_t i = 0; i < size; ++i) {

                if (flags[i]) continue;

                uint16_t idx = sel[i];

                flags[i] &= (null_map[idx] == _is_null);

            }

        }

    } else {

        if (_is_null) memset(flags, false, size);

    }

}

void NullPredicate::evaluate_vec(const vectorized::IColumn& column, uint16_t size,

                                 bool* flags) const {

    if (auto* nullable = check_and_get_column<ColumnNullable>(column)) {

        if (!nullable->has_null()) {

            memset(flags, !_is_null, size);

        }

        auto& null_map = nullable->get_null_map_data();

        for (uint16_t i = 0; i < size; ++i) {

            flags[i] = (null_map[i] == _is_null);

        }

    } else {

        if (_is_null) memset(flags, false, size);

    }

}

} //namespace doris