// 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 "core/block/block.h"

#include "exprs/hybrid_set.h"

#include "runtime/runtime_profile.h"

namespace doris::schema::external {

class TField;

} // namespace doris::schema::external

namespace doris {

// The type an equality-delete key is compared in: the field's type in the scan schema (`field`, which

// is null when the plan carries no schema), as in the V2 reader. Data files are read in that type, so

// an equality delete written before a type promotion is widened to it rather than the data narrowed

// to the delete file's type. Example: `id` promoted from INT to BIGINT, a data file written after the

// promotion holds 4294967296, and a delete file written before it holds the NULL key. Narrowed to

// INT, 4294967296 became NULL and was deleted by that NULL key; widened, the delete key is compared

// as BIGINT and deletes only rows whose id is NULL. STRING / VARBINARY keys keep `delete_key_type`:

// they match by their bytes whatever the two sides were read as.

DataTypePtr equality_delete_key_type(const schema::external::TField* field,

                                     const DataTypePtr& delete_key_type);

// Iceberg BINARY / FIXED keys are read as VARBINARY and string keys as STRING, and the delete file and

// the data file may carry the same key in different ones of these classes (a delete block holding

// STRING bytes for a VARBINARY data column, say). Keys of these types therefore match by their bytes.

bool is_equality_delete_byte_type(const DataTypePtr& type);

// Widens an equality-delete key read in `column_type` to `key_type`, the type the key is compared in.

// A file written before an Iceberg type promotion holds the key in the old type: the INT key of a

// column since promoted to BIGINT becomes BIGINT here. Iceberg's promotions (INT to BIGINT, FLOAT to

// DOUBLE, a wider DECIMAL precision) only widen, so every value is kept. Returns `column` itself when

// the types are equal.

Status widen_equality_delete_key(ColumnPtr column, const DataTypePtr& column_type,

                                 const DataTypePtr& key_type, ColumnPtr* result);

/**

 * Support Iceberg equality delete.

 * If there's only one delete column in delete file, use `SimpleEqualityDelete`,

 * which uses optimized `HybridSetBase` to build the hash set.

 * If there are more delete columns in delete file, use `MultiEqualityDelete`,

 * which generates a hash column from all delete columns, and only compare the values

 * when the hash values are the same.

 */

class EqualityDeleteBase {

protected:

    RuntimeProfile::Counter* num_delete_rows;

    RuntimeProfile::Counter* build_set_time;

    RuntimeProfile::Counter* equality_delete_time;

    const Block* _delete_block;

    std::vector<int> _delete_col_ids;

    virtual Status _build_set() = 0;

public:

    EqualityDeleteBase(const Block* delete_block, const std::vector<int> delete_col_ids)

            : _delete_block(delete_block), _delete_col_ids(delete_col_ids) {}

    virtual ~EqualityDeleteBase() = default;

    Status init(RuntimeProfile* profile) {

        static const char* delete_profile = "EqualityDelete";

        ADD_TIMER_WITH_LEVEL(profile, delete_profile, 1);

        num_delete_rows = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "NumRowsInDeleteFile", TUnit::UNIT,

                                                       delete_profile, 1);

        build_set_time = ADD_CHILD_TIMER_WITH_LEVEL(profile, "BuildHashSetTime", delete_profile, 1);

        equality_delete_time =

                ADD_CHILD_TIMER_WITH_LEVEL(profile, "EqualityDeleteFilterTime", delete_profile, 1);

        SCOPED_TIMER(build_set_time);

        return _build_set();

    }

    virtual Status filter_data_block(

            Block* data_block,

            const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx,

            const std::unordered_map<int, std::string>& id_to_block_column_name,

            IColumn::Filter& filter) = 0;

    static std::unique_ptr<EqualityDeleteBase> get_delete_impl(

            const Block* delete_block, const std::vector<int>& delete_col_ids);

};

class SimpleEqualityDelete : public EqualityDeleteBase {

protected:

    std::shared_ptr<HybridSetBase> _hybrid_set;

    std::unique_ptr<IColumn::Filter> _single_filter;

    Status _build_set() override;

public:

    SimpleEqualityDelete(const Block* delete_block, const std::vector<int>& delete_col_ids)

            : EqualityDeleteBase(delete_block, delete_col_ids) {}

    Status filter_data_block(Block* data_block,

                             const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx,

                             const std::unordered_map<int, std::string>& id_to_block_column_name,

                             IColumn::Filter& filter) override;

};

/**

 * `MultiEqualityDelete` will generate the hash column for delete block and data block.

 */

class MultiEqualityDelete : public EqualityDeleteBase {

protected:

    // hash column for delete block

    std::vector<uint64_t> _delete_hashes;

    // hash column for data block

    std::vector<uint64_t> _data_hashes;

    // hash code => row index

    // if hash values are equal, then compare the real values

    // the row index records the row number of the delete row in delete block

    std::multimap<uint64_t, size_t> _delete_hash_map;

    // the delete column indexes in data block

    std::vector<size_t> _data_column_index;

    Status _build_set() override;

    bool _equal(Block* data_block, size_t data_row_index, size_t delete_row_index);

public:

    MultiEqualityDelete(const Block* delete_block, const std::vector<int>& delete_col_ids)

            : EqualityDeleteBase(delete_block, delete_col_ids) {}

    Status filter_data_block(Block* data_block,

                             const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx,

                             const std::unordered_map<int, std::string>& id_to_block_column_name,

                             IColumn::Filter& filter) override;

};

} // namespace doris