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

#include <cstdint>

#include <memory>

#include <optional>

#include <string>

#include <utility>

#include <vector>

#include "common/status.h"

#include "core/data_type/data_type.h"

#include "core/field.h"

#include "exprs/vexpr_fwd.h"

#include "format_v2/file_reader.h"

namespace doris {

class RuntimeState;

} // namespace doris

namespace doris::format {

struct ColumnDefinition;

struct TableFilter;

enum class TableColumnMappingMode {

    // Match by ColumnDefinition::identifier TYPE_INT as field id.

    BY_FIELD_ID,

    // Match by ColumnDefinition::identifier TYPE_STRING, or logical name when identifier is null.

    BY_NAME,

    // Match top-level columns by file position. This mainly serves Hive1 ORC style files whose

    // column names are placeholder values such as `_col0` / `_col1`, where position is the only

    // reliable way to select the correct column.

    BY_INDEX,

};

enum TableVirtualColumnType {

    INVALID = 0, // not a virtual column

    // Iceberg v3 row lineage metadata column `_row_id`. Physical non-null values

    // are preserved; NULL or missing values inherit first_row_id + row_position.

    ROW_ID = 1,

    // Iceberg v3 row lineage metadata column `_last_updated_sequence_number`.

    // Physical non-null values are preserved; NULL or missing values inherit the

    // data file's last_updated_sequence_number.

    LAST_UPDATED_SEQUENCE_NUMBER = 2,

    // Doris internal Iceberg row locator column `__DORIS_ICEBERG_ROWID_COL__`.

    // It is a struct used by delete/update/merge, not the Iceberg `_row_id`.

    ICEBERG_ROWID = 3,

};

enum class FilterConversionType {

    COPY_DIRECTLY, // filter can be copied directly from file layer without any change, e.g. column type and table type are the same and no complex nested projection is involved.

    CAST_FILTER, // filter can be converted from file layer by adding a cast, e.g. column type is nullable but table type is not, or file column has a trivial nested projection but table column has a complex nested projection.

    READER_EXPRESSION,

    FINALIZE_ONLY, // filter cannot be converted to file layer and should be evaluated at table reader finalize phase, e.g. predicates on ICEBERG_ROW_ID column which is generated by IcebergReader.

    CONSTANT,

};

// Nested global-to-local child mapping. The root index points either to a request-local slot or to

// a child id, depending on the owner. child_mapping keeps the recursive table-child to file-child

// relationship explicit instead of encoding it in ColumnMapping flags.

struct IndexMapping {

    int32_t index = -1;

    std::map<int32_t, std::shared_ptr<IndexMapping>> child_mapping;

};

// Recursive result produced after one table/global column is assigned to a file-local source.

struct ColumnMapResult {

    std::optional<LocalColumnId> local_column_id;

    std::optional<LocalColumnIndex> column_index;

    std::optional<IndexMapping> mapping;

};

// Final mapping entry from one global result column to one file-local source.

struct ColumnMapEntry {

    IndexMapping mapping;

    DataTypePtr local_type;

    DataTypePtr global_type;

    FilterConversionType filter_conversion = FilterConversionType::FINALIZE_ONLY;

};

// Collection of final result-column mappings produced for one file/split.

struct ResultColumnMapping {

    std::map<GlobalIndex, ColumnMapEntry> global_to_local;

};

// Mapping result from one table column to one file column.

// This is the main boundary object between table-level schema semantics and file-local schema

// semantics.

struct ColumnMapping {

    // Position of the top-level projected column in the table/global output block. Table-level

    // filters and column predicates refer to this index after FileScannerV2 translates FE ids at

    // the scanner boundary.

    GlobalIndex global_index;

    std::string table_column_name;

    // File-reader local id for the mapped node.

    //

    // For a root mapping it is convertible to LocalColumnId. For a nested mapping it is the

    // LocalColumnIndex child id under the parent projection. This is deliberately separated from

    // ColumnDefinition::identifier, which is the table-to-file matching key such as Parquet/Iceberg

    // field_id or column name.

    //

    // Empty means the table column is constant, missing, partition-only, or virtual.

    std::optional<int32_t> file_local_id;

    std::string file_column_name;

    // Full file type/children before nested projection pruning. Used to rebuild projected types

    // and to localize nested filters that reference children not present in the output projection.

    DataTypePtr original_file_type;

    std::vector<ColumnDefinition> original_file_children;

    // File children after applying the scan projection. The order follows the file-local semantic

    // schema, not table child order. TableReader uses this to map table-output children back to the

    // file-local block layout when projection, predicate-only children, and schema evolution mix.

    std::vector<ColumnDefinition> projected_file_children;

    // Split/file-local constant entry when this mapping is produced from partition/default/virtual

    // expression instead of physical file data.

    std::optional<ConstantIndex> constant_index;

    // Effective file type after applying casts/remaps/nested projection pruning.

    DataTypePtr file_type;

    // Target table/global type after final materialization.

    DataTypePtr table_type;

    // Final projection expression used to convert file-local values into table/global values, such

    // as casts, defaults, partition values, generated columns, or complex-column remaps.

    VExprContextSPtr projection;

    // Mapping tree for nested table children. The order follows table output children, while file

    // children can be pruned/reordered through each child mapping's file-reader local id.

    std::vector<ColumnMapping> child_mappings;

    // True when file value can be used directly as table value without cast or child remap.

    bool is_trivial = false;

    // How filters referencing this table/global column can be converted below table-reader

    // finalize. This is metadata for localize_filters() and future constant-filter evaluation.

    FilterConversionType filter_conversion = FilterConversionType::FINALIZE_ONLY;

    TableVirtualColumnType virtual_column_type = TableVirtualColumnType::INVALID;

    VExprContextSPtr default_expr;

    std::string debug_string() const;

};

struct TableColumnMapperOptions {

    TableColumnMappingMode mode = TableColumnMappingMode::BY_FIELD_ID;

    std::string debug_string() const;

};

Status clone_table_expr_tree(const VExprSPtr& expr, VExprSPtr* cloned_expr);

const Field* find_partition_value(const ColumnDefinition& table_column,

                                  const std::map<std::string, Field>& partition_values);

// Generic mapping layer from table schema to file schema.

// Iceberg uses BY_FIELD_ID. Plain by-name formats can reuse this component as well, so keep this

// abstraction table-format neutral instead of making it Iceberg-only.

class TableColumnMapper {

public:

    explicit TableColumnMapper(TableColumnMapperOptions options = {})

            : _options(std::move(options)) {}

    virtual ~TableColumnMapper() = default;

    // Build column mappings from table schema to file schema.

    // The resulting ColumnMapping describes how each table column is produced from a file column,

    // a constant, or an expression. Later projection, filter localization, and table-block

    // finalization should all reuse the same mapping.

    virtual Status create_mapping(const std::vector<ColumnDefinition>& projected_columns,

                                  const std::map<std::string, Field>& partition_values,

                                  const std::vector<ColumnDefinition>& file_schema);

    // Convert a table-level scan request into a file-local scan request. table_filters preserve

    // row-level filtering semantics and are rewritten as file-local conjuncts. File-layer pruning

    // such as ZoneMap, dictionary, and bloom filter derives from those localized VExpr conjuncts.

    virtual Status create_scan_request(const std::vector<TableFilter>& table_filters,

                                       const std::vector<ColumnDefinition>& projected_columns,

                                       FileScanRequest* file_request,

                                       RuntimeState* runtime_state = nullptr);

    // Localize table-level filters to the file schema.

    // Trivial mappings can copy structured predicates directly. Type changes may be localized with

    // a safe cast. Expressions that cannot be pushed down safely should be handled through

    // reader_expression_map or table-level finalize/filter fallback.

    virtual Status localize_filters(const std::vector<TableFilter>& table_filters,

                                    FileScanRequest* file_request,

                                    RuntimeState* runtime_state = nullptr);

    void clear() {

        _mappings.clear();

        _hidden_mappings.clear();

        _constant_map.clear();

        _filter_entries.clear();

        _file_schema.clear();

        _partition_values.clear();

    }

    const std::vector<ColumnMapping>& mappings() const { return _mappings; }

    const std::map<GlobalIndex, FilterEntry>& filter_entries() const { return _filter_entries; }

    const ConstantMap& constant_map() const { return _constant_map; }

    std::string debug_string() const;

protected:

    // Columnar readers such as Parquet can read predicate columns first, evaluate row filters, and

    // lazily read the rest. Row-oriented readers such as CSV/Text materialize one row at a time and

    // should keep all required columns in one scan list.

    virtual bool enable_lazy_materialization() const { return true; }

    // Row-oriented readers such as CSV/Text cannot physically read only a nested child from one

    // delimited text field. They must scan the whole complex top-level field and let TableReader

    // rematerialize the requested table child after row-level filters have run.

    virtual bool force_full_complex_scan_projection() const { return false; }

    const ColumnDefinition* _find_file_field(

            const ColumnDefinition& table_column,

            const std::vector<ColumnDefinition>& file_schema) const;

    Status _create_direct_mapping(const ColumnDefinition& table_column,

                                  const ColumnDefinition& file_field, ColumnMapping* mapping) const;

    Status _create_by_index_mapping(const ColumnDefinition& table_column,

                                    const std::vector<ColumnDefinition>& file_schema,

                                    ColumnMapping* mapping);

    Status _build_filter_entries(const FileScanRequest& file_request);

    Status _build_result_column_mapping(const FileScanRequest& file_request);

    void _set_constant_mapping(ColumnMapping* mapping, VExprContextSPtr expr);

    Status _create_mapping_for_column(const ColumnDefinition& table_column,

                                      GlobalIndex global_index, ColumnMapping* mapping);

    Status _create_hidden_filter_mapping(const ColumnDefinition& table_column,

                                         GlobalIndex global_index, ColumnMapping* mapping);

    Status _build_hidden_filter_mappings(const std::vector<TableFilter>& table_filters);

    std::vector<ColumnMapping> _filter_visible_mappings() const;

    ColumnMapping* _find_mapping(GlobalIndex global_index);

    ColumnMapping* _find_filter_mapping(GlobalIndex global_index);

    TableColumnMapperOptions _options;

    // Column mapping for each projected column, in the same order as projected_columns. Each entry

    // describes how to get one table/global column from file-local sources, and carries metadata

    // for filter localization and result finalize.

    std::vector<ColumnMapping> _mappings;

    // Predicate-only top-level columns are not output projection columns, so keep their mappings

    // here. They are visible only to filter localization and file-reader predicate construction.

    std::vector<ColumnMapping> _hidden_mappings;

    std::map<GlobalIndex, FilterEntry> _filter_entries;

    ConstantMap _constant_map;

    // Split-local schema state retained from create_mapping() so create_scan_request() can build

    // hidden mappings for top-level filter slots that are absent from projected_columns.

    std::vector<ColumnDefinition> _file_schema;

    std::map<std::string, Field> _partition_values;

};

// Parquet consumes the full FileScanRequest shape: predicate columns for lazy materialization and

// file-local conjuncts for ZoneMap, dictionary, and bloom-filter pruning.

class ParquetColumnMapper final : public TableColumnMapper {

public:

    using TableColumnMapper::TableColumnMapper;

};

// Mapper for readers that always materialize every required file column before filtering. The

// table-to-file schema mapping is still generic, but the FileScanRequest layout is simpler:

// predicate_columns are not populated.

class MaterializedColumnMapper final : public TableColumnMapper {

public:

    using TableColumnMapper::TableColumnMapper;

protected:

    bool enable_lazy_materialization() const override { return false; }

    bool force_full_complex_scan_projection() const override { return true; }

};

} // namespace doris::format