be/src/format_v2/column_mapper.h

// 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;
    // One-row constant owns variable-width payloads; Field<TYPE_VARBINARY> is only a borrowed
    // StringView and cannot safely outlive the Base64 decode buffer used to construct it.
    ColumnPtr initial_default_column;

    std::string debug_string() const;
};

struct TableColumnMapperOptions {
    TableColumnMappingMode mode = TableColumnMappingMode::BY_FIELD_ID;
    bool allow_idless_complex_wrapper_projection = false;
    bool enable_row_lineage_virtual_columns = false;

    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);
// Apply the same case-insensitive logical name, string identifier, and bidirectional alias rules
// used by TableColumnMapper's BY_NAME mode.
const ColumnDefinition* find_column_by_name(const ColumnDefinition& table_column,
                                            const std::vector<ColumnDefinition>& file_schema);

// 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 by the
    // 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