// 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 <algorithm>

#include <cstddef>

#include <cstdint>

#include <memory>

#include <ostream>

#include <string>

#include <utility>

#include <vector>

#include "common/consts.h"

#include "common/status.h"

#include "core/data_type/data_type.h"

#include "core/data_type/data_type_number.h"

#include "core/data_type/data_type_string.h"

#include "core/field.h"

#include "exprs/vexpr_fwd.h"

namespace doris::format {

// File-local top-level column id.

//

// Scope:

// - Only valid inside one physical file schema returned by FileReader::get_schema().

// - For Parquet, this is the top-level field ordinal in the new reader schema.

// - The synthetic row-position column also uses this type, with a reserved negative id.

//

// Do not use this for table/global column unique ids, block positions, nested child ids, or

// slot ids. Nested child ids are carried by LocalColumnIndex::index below.

class LocalColumnId {

public:

    constexpr LocalColumnId() = default;

    explicit constexpr LocalColumnId(int32_t id) : _id(id) {}

    static constexpr LocalColumnId invalid() { return LocalColumnId(); }

    constexpr int32_t value() const { return _id; }

    constexpr bool is_valid() const { return _id >= 0; }

    constexpr bool operator==(const LocalColumnId& other) const { return _id == other._id; }

    constexpr bool operator!=(const LocalColumnId& other) const { return !(*this == other); }

    constexpr bool operator<(const LocalColumnId& other) const { return _id < other._id; }

private:

    int32_t _id = -1;

};

// Position of a file-local column in the Block produced by one FileScanRequest.

//

// This is assigned by TableColumnMapper/TableReader after predicate/non-predicate columns are

// deduplicated. It is not a file schema id and it is not stable across requests. Use value() only

// at the boundary where an existing Block or expression API still expects a size_t/int position.

class LocalIndex {

public:

    constexpr LocalIndex() = default;

    explicit constexpr LocalIndex(size_t index) : _index(index) {}

    constexpr size_t value() const { return _index; }

    constexpr bool operator==(const LocalIndex& other) const { return _index == other._index; }

    constexpr bool operator<(const LocalIndex& other) const { return _index < other._index; }

private:

    size_t _index = 0;

};

// Position of a table/global output column in the final Block returned by TableReader.

//

// This type is reserved for boundaries that need to refer to caller-visible column order. It must

// not be used to index a file-local Block, because schema evolution and lazy materialization can

// make file-local order different from table output order.

class GlobalIndex {

public:

    constexpr GlobalIndex() = default;

    explicit constexpr GlobalIndex(size_t index) : _index(index) {}

    constexpr size_t value() const { return _index; }

    constexpr bool operator==(const GlobalIndex& other) const { return _index == other._index; }

    constexpr bool operator<(const GlobalIndex& other) const { return _index < other._index; }

private:

    size_t _index = 0;

};

// Index of a split-local constant/default value used to materialize columns that are not read from

// the physical file, such as partition columns, added columns with default values, and virtual

// table-format columns.

//

// It is separate from LocalIndex because constants do not occupy a position in the file reader

// output block unless an expression explicitly materializes them.

class ConstantIndex {

public:

    constexpr ConstantIndex() = default;

    explicit constexpr ConstantIndex(size_t index) : _index(index) {}

    constexpr size_t value() const { return _index; }

    constexpr bool operator==(const ConstantIndex& other) const { return _index == other._index; }

    constexpr bool operator<(const ConstantIndex& other) const { return _index < other._index; }

private:

    size_t _index = 0;

};

inline std::ostream& operator<<(std::ostream& os, const LocalColumnId& id) {

    return os << id.value();

}

inline std::ostream& operator<<(std::ostream& os, const LocalIndex& index) {

    return os << index.value();

}

inline std::ostream& operator<<(std::ostream& os, const GlobalIndex& index) {

    return os << index.value();

}

inline std::ostream& operator<<(std::ostream& os, const ConstantIndex& index) {

    return os << index.value();

}

// A split/file-local constant value used to materialize a table/global column without reading a

// physical file column.

//

// Common producers are partition values, schema-evolution default expressions, generated columns

// and table-format virtual columns. The entry is keyed by ConstantIndex in ConstantMap; global_index

// keeps the link back to the caller-visible output column.

struct ConstantEntry {

    GlobalIndex global_index;

    VExprContextSPtr expr;

    DataTypePtr type;

};

// Per mapping/split collection of constants.

//

// ConstantIndex only has meaning within this container. Keeping constants separate from LocalIndex

// makes it explicit that these values do not occupy positions in the file reader output Block.

class ConstantMap {

public:

    ConstantIndex add(ConstantEntry entry) {

        const auto index = ConstantIndex(_entries.size());

        _entries.push_back(std::move(entry));

        return index;

    }

    const ConstantEntry& get(ConstantIndex index) const {

        DORIS_CHECK(index.value() < _entries.size());

        return _entries[index.value()];

    }

    void clear() { _entries.clear(); }

    bool empty() const { return _entries.empty(); }

    size_t size() const { return _entries.size(); }

    const std::vector<ConstantEntry>& entries() const { return _entries; }

private:

    std::vector<ConstantEntry> _entries;

};

// Target of a localized filter.

//

// A filter can either reference a file-local Block position or a constant entry. Unset entries mean

// the filter cannot be evaluated below the table-reader finalize stage.

struct FilterEntry {

    enum class Kind {

        UNSET,

        LOCAL,

        CONSTANT,

    };

    static FilterEntry local(LocalIndex index) {

        return {.kind = Kind::LOCAL, .index = index.value()};

    }

    static FilterEntry constant(ConstantIndex index) {

        return {.kind = Kind::CONSTANT, .index = index.value()};

    }

    bool is_set() const { return kind != Kind::UNSET; }

    bool is_local() const { return kind == Kind::LOCAL; }

    bool is_constant() const { return kind == Kind::CONSTANT; }

    LocalIndex local_index() const {

        DORIS_CHECK(is_local());

        return LocalIndex(index);

    }

    ConstantIndex constant_index() const {

        DORIS_CHECK(is_constant());

        return ConstantIndex(index);

    }

    Kind kind = Kind::UNSET;

    size_t index = 0;

};

enum ColumnType {

    DATA_COLUMN = 0,  // normal data column

    ROW_NUMBER = 1,   // row number in a file

    GLOBAL_ROWID = 2, // global unique row id across files, used by TopN filter

};

struct GlobalRowIdContext {

    uint8_t version = 0;

    int64_t backend_id = 0;

    uint32_t file_id = 0;

};

// Column schema definition shared by table/global projection and file-local schema matching.

//

// ColumnDefinition intentionally carries schema identity only. FE column unique ids are translated

// to GlobalIndex at the FileScannerV2 boundary and must not appear in table/file reader APIs.

struct ColumnDefinition {

    // Typed identifier value used to match a column against another schema.

    //

    // - TYPE_NULL: no explicit identifier. BY_NAME falls back to ColumnDefinition::name.

    // - TYPE_INT: interpreted by TableColumnMapperOptions::mode as a field id or file position.

    // - TYPE_STRING: explicit name identifier.

    //

    // This is not the id that FileReader uses to read data. For example, a Parquet column can be

    // matched by its optional Parquet field_id, while the reader still addresses it by a file-local

    // ordinal.

    Field identifier;

    // Reader-local id of this node inside the file schema returned by FileReader::get_schema().

    // Top-level fields use the root column ordinal and nested fields use the child ordinal under

    // their parent. -1 means unset; special virtual file columns may use other negative ids.

    // Table/global ColumnDefinition values can leave this as -1 because they are not read directly

    // by a FileReader.

    int32_t local_id = -1;

    // Logical table column name. This is also the matching name for by-name file formats.

    std::string name;

    // Historical or external names for the same logical field. Table formats such as Iceberg can

    // use this to resolve partition path keys after column rename.

    std::vector<std::string> name_mapping {};

    // Distinguishes no table-level mapping from an explicit empty field mapping. The latter must

    // not fall back to the current name when matching fields in legacy files.

    bool has_name_mapping = false;

    DataTypePtr type;

    // Semantic nested children for this schema node.

    //

    // Table/global columns carry projected table children. File-local schemas returned by

    // FileReader::get_schema() also expose semantic children, not physical reader wrappers. For

    // example, MAP children are key/value and ARRAY children contain only the element field.

    std::vector<ColumnDefinition> children {};

    // Full table-schema identity subtree before access-path pruning. ID-less physical complex

    // wrappers must be discovered from this view without adding unrequested children to output.

    std::vector<ColumnDefinition> identity_children {};

    // Logical object-key paths requested from a Variant column. An empty collection means the

    // whole Variant is required; non-empty paths may be resolved to format-specific shredded

    // physical children after the per-file schema is known.

    std::vector<std::vector<std::string>> variant_access_paths {};

    // Predicate access paths are kept separately from the final union projection. File Scanner V2

    // can lower this smaller semantic tree to an eager predicate projection while deferring the

    // final children until rows survive. The flag distinguishes no predicate metadata from a

    // whole-root predicate, whose child/path collections are intentionally empty.

    bool has_predicate_access_paths = false;

    std::vector<ColumnDefinition> predicate_children {};

    std::vector<std::vector<std::string>> predicate_variant_access_paths {};

    // Expression used to materialize missing/default/generated values when the column is not read

    // directly from the file.

    VExprContextSPtr default_expr = nullptr;

    // Table-format initial default normalized for transport from FE. Binary-like values use Base64

    // and set initial_default_value_is_base64 because they can map to STRING/CHAR or VARBINARY.

    // Unlike default_expr, this metadata is also available for hidden delete-predicate columns

    // that are absent from the query projection.

    std::optional<std::string> initial_default_value = std::nullopt;

    bool initial_default_value_is_base64 = false;

    // Table-format field optionality. std::nullopt means the format did not provide this semantic

    // metadata. Iceberg uses an explicit false value to reject old files that are missing a

    // required field without an initial default.

    std::optional<bool> is_optional = std::nullopt;

    // Logical timestamp semantic supplied by a table format when the physical encoding cannot

    // carry it (for example, Paimon TIMESTAMP versus TIMESTAMP_LTZ stored as INT96).

    std::optional<bool> timestamp_is_adjusted_to_utc = std::nullopt;

    // Partition columns are constants from split metadata and should not be matched against file

    // schema unless table-format logic explicitly asks for it.

    bool is_partition_key = false;

    // File-local column kind. For table/global columns this remains DATA_COLUMN.

    ColumnType column_type = ColumnType::DATA_COLUMN;

    bool has_identifier() const { return !identifier.is_null(); }

    bool has_identifier_field_id() const { return identifier.get_type() == TYPE_INT; }

    bool has_identifier_name() const { return identifier.get_type() == TYPE_STRING; }

    // DuckDB-style helper for BY_FIELD_ID matching. The mapper binds the matching mode once, so a

    // TYPE_INT identifier is interpreted as a field id only by the field-id matcher.

    int32_t get_identifier_field_id() const {

        DORIS_CHECK(has_identifier_field_id());

        return identifier.get<TYPE_INT>();

    }

    // DuckDB-style helper for BY_NAME matching. When no explicit string identifier is present, the

    // logical column name is the identifier.

    const std::string& get_identifier_name() const {

        if (identifier.is_null()) {

            return name;

        }

        DORIS_CHECK(has_identifier_name());

        return identifier.get<TYPE_STRING>();

    }

    // Helper for BY_INDEX matching. BY_INDEX reuses the TYPE_INT identifier as the table-side file

    // position, matching DuckDB's typed identifier plus mapper-mode interpretation.

    int32_t get_identifier_position() const {

        DORIS_CHECK(has_identifier_field_id());

        return identifier.get<TYPE_INT>();

    }

    // Helper for reader-local projection and scan requests.

    int32_t file_local_id() const {

        if (local_id != -1) {

            return local_id;

        }

        return get_identifier_field_id();

    }

    std::string debug_string() const;

};

static constexpr int ROW_POSITION_COLUMN_ID = -10001;

static constexpr const char* ROW_POSITION_COLUMN_NAME = "__file_row_position";

static constexpr int GLOBAL_ROWID_COLUMN_ID = -10002;

inline ColumnDefinition row_position_column_definition() {

    ColumnDefinition field;

    field.identifier = Field::create_field<TYPE_INT>(ROW_POSITION_COLUMN_ID);

    field.local_id = ROW_POSITION_COLUMN_ID;

    field.name = ROW_POSITION_COLUMN_NAME;

    field.type = std::make_shared<DataTypeInt64>();

    field.column_type = ColumnType::ROW_NUMBER;

    return field;

}

inline ColumnDefinition global_rowid_column_definition() {

    ColumnDefinition field;

    field.identifier = Field::create_field<TYPE_STRING>(BeConsts::GLOBAL_ROWID_COL);

    field.local_id = GLOBAL_ROWID_COLUMN_ID;

    field.name = BeConsts::GLOBAL_ROWID_COL;

    field.type = std::make_shared<DataTypeString>();

    field.column_type = ColumnType::GLOBAL_ROWID;

    return field;

}

// Recursive file-local projection path.

//

// For a root entry in FileScanRequest::{predicate_columns, non_predicate_columns}, index is the

// top-level file column id and column_id() is valid. For children, index is the file-local child id

// under the parent node. This is the reader schema local id, not an Iceberg/Parquet field id, not a

// table child id, and not a child output ordinal.

//

// project_all_children=true means the whole subtree under this node is needed. When false, children

// lists the selected child paths. File readers can use this to avoid constructing readers for

// unprojected nested children.

struct LocalColumnIndex {

    int32_t index = -1;

    bool project_all_children = true;

    std::vector<LocalColumnIndex> children {};

    std::optional<bool> timestamp_is_adjusted_to_utc = std::nullopt;

    static LocalColumnIndex top_level(LocalColumnId column_id) {

        return {.index = column_id.value()};

    }

    static LocalColumnIndex local(int32_t local_id) { return {.index = local_id}; }

    static LocalColumnIndex partial_local(int32_t local_id) {

        return {.index = local_id, .project_all_children = false};

    }

    LocalColumnId column_id() const { return LocalColumnId(index); }

    int32_t local_id() const { return index; }

    std::string debug_string() const;

};

inline bool same_local_column_index(const LocalColumnIndex& lhs, const LocalColumnIndex& rhs) {

    if (lhs.index != rhs.index || lhs.project_all_children != rhs.project_all_children ||

        lhs.children.size() != rhs.children.size()) {

        return false;

    }

    for (size_t i = 0; i < lhs.children.size(); ++i) {

        if (!same_local_column_index(lhs.children[i], rhs.children[i])) {

            return false;

        }

    }

    return true;

}

inline bool is_full_projection(const LocalColumnIndex* projection) {

    return projection == nullptr || projection->project_all_children;

}

inline bool is_partial_projection(const LocalColumnIndex* projection) {

    return projection != nullptr && !projection->project_all_children;

}

inline const LocalColumnIndex* find_child_projection(const LocalColumnIndex* projection,

                                                     int32_t local_id) {

    if (is_full_projection(projection)) {

        return nullptr;

    }

    const auto child_it = std::find_if(

            projection->children.begin(), projection->children.end(),

            [&](const LocalColumnIndex& child) { return child.local_id() == local_id; });

    return child_it == projection->children.end() ? nullptr : &*child_it;

}

inline bool is_child_projected(const LocalColumnIndex* projection, int32_t local_id) {

    return is_full_projection(projection) || find_child_projection(projection, local_id) != nullptr;

}

// Merge two projection trees that point to the same file-local node.

//

// A full projection dominates a partial projection. Two partial projections are merged by child id

// and recursively union their child paths. The caller must only merge projections for the same

// root/child node.

inline Status merge_local_column_index(LocalColumnIndex* target, const LocalColumnIndex& source) {

    DORIS_CHECK(target != nullptr);

    DORIS_CHECK(target->index == source.index);

    if (!target->timestamp_is_adjusted_to_utc.has_value()) {

        target->timestamp_is_adjusted_to_utc = source.timestamp_is_adjusted_to_utc;

    } else if (source.timestamp_is_adjusted_to_utc.has_value() &&

               target->timestamp_is_adjusted_to_utc != source.timestamp_is_adjusted_to_utc) {

        return Status::InvalidArgument("Conflicting timestamp semantics for file-local column {}",

                                       target->index);

    }

    if (target->project_all_children) {

        return Status::OK();

    }

    if (source.project_all_children) {

        target->project_all_children = true;

        target->children.clear();

        return Status::OK();

    }

    for (const auto& source_child : source.children) {

        auto target_child_it = std::find_if(

                target->children.begin(), target->children.end(),

                [&](const LocalColumnIndex& child) { return child.index == source_child.index; });

        if (target_child_it == target->children.end()) {

            target->children.push_back(source_child);

            continue;

        }

        RETURN_IF_ERROR(merge_local_column_index(&*target_child_it, source_child));

    }

    return Status::OK();

}

} // namespace doris::format