// 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 "format_v2/materialized_reader_util.h"

#include <utility>

#include "core/block/block.h"

#include "core/column/column_vector.h"

#include "core/data_type/data_type_nullable.h"

#include "core/types.h"

#include "exprs/vexpr_context.h"

#include "format_v2/file_reader.h"

#include "io/io_common.h"

namespace doris::format {

namespace {

void update_counter(RuntimeProfile::Counter* counter, int64_t value) {

    if (counter != nullptr) {

        COUNTER_UPDATE(counter, value);

    }

}

} // namespace

ColumnPtr make_column_nullable_if_needed(ColumnPtr column, const DataTypePtr& target_type) {

    if (target_type != nullptr && target_type->is_nullable() && column.get() != nullptr &&

        !column->is_nullable()) {

        return make_nullable(std::move(column));

    }

    return column;

}

Status apply_materialized_reader_filters(const FileScanRequest* request, io::IOContext* io_ctx,

                                         Block* file_block, size_t* rows,

                                         const MaterializedReaderFilterProfile* profile) {

    DORIS_CHECK(file_block != nullptr);

    DORIS_CHECK(rows != nullptr);

    const size_t rows_before_filter = *rows;

    size_t rows_after_delete_filter = rows_before_filter;

    if (request != nullptr && rows_before_filter > 0 && !request->delete_conjuncts.empty()) {

        {

            SCOPED_TIMER(profile == nullptr ? nullptr : profile->delete_conjunct_filter_time);

            // Delete conjuncts use the opposite polarity from ordinary predicates: TRUE marks a

            // row for deletion. Multiple delete files are combined as a union, so a row remains

            // visible only when every delete conjunct returns FALSE.

            IColumn::Filter keep_filter(rows_before_filter, 1);

            for (const auto& delete_conjunct : request->delete_conjuncts) {

                DORIS_CHECK(delete_conjunct != nullptr);

                int result_column_id = -1;

                RETURN_IF_ERROR(delete_conjunct->root()->execute(delete_conjunct.get(), file_block,

                                                                 &result_column_id));

                DORIS_CHECK(result_column_id >= 0 &&

                            result_column_id < static_cast<int>(file_block->columns()));

                const auto& delete_filter =

                        assert_cast<const ColumnUInt8&>(

                                *file_block->get_by_position(result_column_id).column)

                                .get_data();

                DORIS_CHECK(delete_filter.size() == rows_before_filter);

                for (size_t row = 0; row < rows_before_filter; ++row) {

                    keep_filter[row] &= !delete_filter[row];

                }

                file_block->erase(result_column_id);

            }

            RETURN_IF_CATCH_EXCEPTION(Block::filter_block_internal(file_block, keep_filter));

        }

        rows_after_delete_filter =

                file_block->columns() == 0 ? rows_before_filter : file_block->rows();

        if (profile != nullptr) {

            update_counter(profile->rows_filtered_by_delete_conjunct,

                           rows_before_filter - rows_after_delete_filter);

        }

    }

    size_t rows_after_filter = rows_after_delete_filter;

    if (request != nullptr && rows_after_delete_filter > 0 && !request->conjuncts.empty()) {

        {

            SCOPED_TIMER(profile == nullptr ? nullptr : profile->conjunct_filter_time);

            RETURN_IF_ERROR(VExprContext::filter_block(request->conjuncts, file_block,

                                                       file_block->columns()));

        }

        rows_after_filter =

                file_block->columns() == 0 ? rows_after_delete_filter : file_block->rows();

        const auto rows_filtered_by_conjunct = rows_after_delete_filter - rows_after_filter;

        if (profile != nullptr) {

            update_counter(profile->rows_filtered_by_conjunct, rows_filtered_by_conjunct);

        }

        if (io_ctx != nullptr) {

            io_ctx->predicate_filtered_rows += rows_filtered_by_conjunct;

        }

    }

    *rows = rows_after_filter;

    return Status::OK();

}

} // namespace doris::format