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

#include <optional>

#include <string>

#include <unordered_map>

#include <vector>

#include "common/status.h"

#include "core/block/block.h"

#include "format/table/iceberg_delete_file_reader_helper.h"

#include "format_v2/file_reader.h"

#include "format_v2/table/iceberg_schema_utils.h"

#include "format_v2/table_reader.h"

#include "gen_cpp/PlanNodes_types.h"

namespace doris {

class Block;

class EqualityDeleteHashIndex;

struct DeleteFileDesc;

namespace io {

struct FileDescription;

struct FileSystemProperties;

} // namespace io

} // namespace doris

namespace doris::format::iceberg {

Status prepare_iceberg_initial_default_exprs(format::ColumnDefinition* column);

// Iceberg table-level reader.

// It reuses TableReader for split orchestration, dynamic partition pruning and table-block

// finalization, while composing a FileReader for physical data-file reads instead of inheriting

// from a concrete file-format reader.

class IcebergTableReader : public format::TableReader {

public:

    ~IcebergTableReader() override = default;

    static Status validate_variant_file_mappings(

            FileFormat format, const std::vector<format::ColumnMapping>& mappings);

    Status init(format::TableReadOptions&& options) override {

        RETURN_IF_ERROR(format::TableReader::init(std::move(options)));

        _mapper_options.mode = format::TableColumnMappingMode::BY_FIELD_ID;

        _mapper_options.reject_missing_required_field =

                supports_iceberg_scan_semantics_v2(_scan_params);

        return Status::OK();

    }

    Status prepare_split(const format::SplitReadOptions& options) override;

    Status annotate_projected_column(const TFileScanSlotInfo& slot_info,

                                     format::ProjectedColumnBuildContext* context,

                                     format::ColumnDefinition* column) const override;

    std::string debug_string() const override;

    format::TableColumnMappingMode mapping_mode() const override {

        return _mapping_mode(_data_reader.file_schema);

    }

protected:

    // Iceberg UUID uses the same 16-byte STRING/VARBINARY carrier in data, defaults and deletes.

    bool preserve_binary_uuid() const override { return true; }

    Status validate_file_mapping(const format::TableColumnMapper& mapper) const override;

    // Iceberg's Parquet reader resolves a data file that carries no field ids at all through the

    // table's `schema.name-mapping.default`; when that mapping is absent it falls back to

    // positional ids (ParquetSchemaUtil.addFallbackIds + pruneColumnsFallback, gated by

    // `iceberg.netflix.unsafe-parquet-id-fallback.enabled`, default true through 1.12 and

    // scheduled for removal in 2.0). Mirror that decision for Parquet by stamping ordinal ids

    // (1..N) on the file's top-level fields, so the shared BY_FIELD_ID path resolves those files

    // instead of silently materializing NULL. Only top-level fields are stamped: Iceberg assigns

    // no ids to nested fields, so struct children of an id-less file stay unreadable without a

    // name mapping (or a rewrite that persists field ids).

    Status annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) override;

    // Iceberg's positional fallback matches an ID-less Parquet file by physical ordinals, so the

    // same table column can bind a different file column when the table's

    // `schema.name-mapping.default` is added or removed without touching the data file. The

    // condition cache key covers the file identity and the table-level predicate digest only, so it

    // cannot tell those bindings apart; keep such splits out of the cache (neither read nor

    // written) instead of reusing a verdict produced by the other binding.

    bool _should_enable_condition_cache(

            const format::FileScanRequest& file_request) const override {

        return !_positional_id_fallback &&

               format::TableReader::_should_enable_condition_cache(file_request);

    }

    void configure_mapper_options(format::TableColumnMapperOptions* options) const override {

        options->enable_row_lineage_virtual_columns = true;

        options->enable_iceberg_metadata_virtual_columns = true;

        options->reject_missing_required_field = supports_iceberg_scan_semantics_v2(_scan_params);

        options->allow_idless_complex_wrapper_projection =

                supports_iceberg_scan_semantics_v1(_scan_params) && _format == FileFormat::PARQUET;

    }

    Status materialize_virtual_columns(Block* table_block) override;

    Status customize_file_scan_request(format::FileScanRequest* file_request) override;

    bool _supports_aggregate_pushdown(TPushAggOp::type agg_type) const override;

    Status _parse_deletion_vector_file(const TTableFormatFileDesc& t_desc, DeleteFileDesc* desc,

                                       bool* has_delete_file) override;

    Status _init_delete_predicates(const TTableFormatFileDesc& t_desc);

private:

    struct EqualityDeleteFilter;

    static constexpr int MIN_SUPPORT_DELETE_FILES_VERSION = 2;

    static constexpr int POSITION_DELETE = 1;

    static constexpr int EQUALITY_DELETE = 2;

    static constexpr int DELETION_VECTOR = 3;

    struct RowLineageColumns {

        int64_t first_row_id = -1;

        int64_t last_updated_sequence_number = -1;

    };

    static constexpr const char* ICEBERG_FILE_PATH = "file_path";

    static constexpr const char* ICEBERG_ROW_POS = "pos";

    static constexpr size_t ICEBERG_FILE_PATH_BLOCK_POSITION = 0;

    static constexpr size_t ICEBERG_ROW_POS_BLOCK_POSITION = 1;

    bool _scan_has_any_authoritative_name_mapping() const;

    // annotate_file_schema() runs before TableReader stores the file schema, so the mode is also

    // computable from the schema being annotated.

    format::TableColumnMappingMode _mapping_mode(

            const std::vector<format::ColumnDefinition>& file_schema) const {

        const bool has_field_ids = supports_iceberg_scan_semantics_v1(_scan_params)

                                           ? schema_has_any_field_id(file_schema)

                                           : schema_has_all_field_ids(file_schema);

        if (!file_schema.empty() && has_field_ids) {

            return format::TableColumnMappingMode::BY_FIELD_ID;

        }

        if (!file_schema.empty() && supports_iceberg_scan_semantics_v2(_scan_params) &&

            !_scan_has_any_authoritative_name_mapping()) {

            // ID-less migrated files are name-readable only while Iceberg's explicit default name

            // mapping exists; current names must not resurrect file fields after it is removed.

            return format::TableColumnMappingMode::BY_FIELD_ID;

        }

        return format::TableColumnMappingMode::BY_NAME;

    }

    void _apply_positional_id_fallback(std::vector<format::ColumnDefinition>* file_schema);

    Status _apply_table_timestamp_semantics(std::vector<format::ColumnDefinition>* file_schema,

                                            format::TableColumnMappingMode mode) const;

    class PositionDeleteRowsCollector final {

    public:

        using PositionDeleteFile = std::unordered_map<std::string, format::DeleteRows>;

        explicit PositionDeleteRowsCollector(PositionDeleteFile* rows_by_data_file);

        Status collect(const Block& block, size_t read_rows);

    private:

        PositionDeleteFile* _rows_by_data_file = nullptr;

    };

    static std::shared_ptr<io::FileSystemProperties> _delete_file_system_properties(

            const TFileScanRangeParams& scan_params);

    static std::unique_ptr<io::FileDescription> _delete_file_description(

            const TFileRangeDesc& range);

    std::string _data_file_path() const;

    // Append row position column to file scan request for position delete handling.

    Status _append_row_position_output_column(format::FileScanRequest* request);

    // Append equality delete predicates to file scan request based on the delete files in iceberg

    // params. DeleteVector and position delete files use the common DeleteRows path in TableReader.

    using EqualityDeleteColumnPath = std::vector<const format::ColumnDefinition*>;

    Status _append_equality_delete_predicates(format::FileScanRequest* request);

    Status _build_missing_equality_delete_key_expr(const EqualityDeleteFilter& filter,

                                                   size_t key_idx,

                                                   const EqualityDeleteColumnPath& data_path,

                                                   format::FileScanRequest* request,

                                                   VExprSPtr* key_expr,

                                                   DataTypePtr* comparison_type);

    // Returns `filter` with its delete keys converted to `comparison_types`, cached per scanner.

    Status _convert_equality_delete_filter(const EqualityDeleteFilter& filter,

                                           const DataTypes& comparison_types,

                                           EqualityDeleteFilter* result);

    Status _build_converted_equality_delete_filter(const EqualityDeleteFilter& filter,

                                                   const DataTypes& comparison_types,

                                                   EqualityDeleteFilter* result) const;

    Status _find_equality_delete_data_field(const EqualityDeleteFilter& filter, size_t key_idx,

                                            EqualityDeleteColumnPath* data_path,

                                            bool* complete_path) const;

    Status _find_equality_delete_table_field(const EqualityDeleteFilter& filter, size_t key_idx,

                                             format::ColumnDefinition* table_field) const;

    void _append_equality_delete_row_count_carrier(format::FileScanRequest* request);

    std::string _delete_file_cache_key(const char* prefix, const std::string& path) const;

    Status _init_equality_delete_predicates(

            const std::vector<TIcebergDeleteFileDesc>& delete_files);

    // Read equality/position delete files.

    Status _create_delete_file_reader(const TIcebergDeleteFileDesc& delete_file,

                                      const TFileScanRangeParams& scan_params,

                                      IcebergDeleteFileIOContext* delete_io_ctx,

                                      std::unique_ptr<format::FileReader>* reader);

    Status _read_equality_delete_file(const TIcebergDeleteFileDesc& delete_file,

                                      const TFileScanRangeParams& scan_params,

                                      IcebergDeleteFileIOContext* delete_io_ctx);

    Status _load_equality_delete_file(const TIcebergDeleteFileDesc& delete_file,

                                      const TFileScanRangeParams& scan_params,

                                      IcebergDeleteFileIOContext* delete_io_ctx,

                                      EqualityDeleteFilter* result);

    Status _resolve_equality_delete_fields(const TIcebergDeleteFileDesc& delete_file,

                                           const std::vector<format::ColumnDefinition>& schema,

                                           std::vector<EqualityDeleteColumnPath>* delete_paths,

                                           EqualityDeleteFilter* result) const;

    Status _read_position_delete_file(const TIcebergDeleteFileDesc& delete_file,

                                      const TFileScanRangeParams& scan_params,

                                      IcebergDeleteFileIOContext* delete_io_ctx,

                                      PositionDeleteRowsCollector* collector);

    // Read position delete files and collect deleted row positions to update DeletePredicate.

    Status _init_position_delete_rows(const std::vector<TIcebergDeleteFileDesc>& delete_files);

    // Materialize row lineage virtual columns based on the position delete file.

    Status _materialize_iceberg_rowid(Block* table_block, size_t column_idx);

    Status _materialize_iceberg_file_path(Block* table_block, size_t column_idx);

    Status _materialize_iceberg_row_position(Block* table_block, size_t column_idx);

    Status _materialize_row_lineage_row_id(Block* table_block, size_t column_idx);

    Status _materialize_row_lineage_last_updated_sequence_number(Block* table_block,

                                                                 size_t column_idx);

    RowLineageColumns _row_lineage_columns;

    size_t _row_position_block_position = 0;

    std::optional<TIcebergFileDesc> _iceberg_params;

    bool _delete_predicates_initialized = false;

    // Set when this split's Parquet file carried no field ids and the table exposed no usable name

    // mapping, so the reader stamped Iceberg's positional fallback ids on the file's top-level

    // fields. Observability only; the decision itself lives in annotate_file_schema().

    bool _positional_id_fallback = false;

    format::DeleteRows _position_delete_rows_storage;

    struct EqualityDeleteFilter {

        std::vector<int> field_ids;

        // Delete-file names are retained for Iceberg tables imported from formats that did not

        // persist field ids. In BY_NAME mode they are the fallback binding key.

        std::vector<std::string> field_names;

        // Table-side types of the keys, from the scan's schema metadata. They resolve table

        // fields and build defaults for keys a data file lacks. The comparison type comes from

        // the file types; a table type only lets a timestamptz key read as DATETIMEV2 compare

        // as TIMESTAMPTZ, and a STRING placeholder (an Iceberg TIME or timestamptz_ns, which

        // Doris cannot represent) lets an ORC TIME key and its text default compare as TIMEV2.

        std::vector<DataTypePtr> key_types;

        // Types of the delete block columns: the delete file's own key types, or the comparison

        // types of a converted filter.

        std::vector<DataTypePtr> delete_key_types;

        Block delete_block;

        std::shared_ptr<const EqualityDeleteHashIndex> hash_index;

        // Split-cache key of this filter, extended by the comparison types for converted copies.

        std::string cache_key;

    };

    std::vector<EqualityDeleteFilter> _equality_delete_filters;

    // Scanner-shared cache supplied in SplitReadOptions. Parsed delete files outlive one data-file

    // split and can be reused by every split referencing the same delete file.

    ShardedKVCache* _split_cache = nullptr;

    bool _need_row_lineage_row_id() const;

    bool _need_iceberg_rowid() const;

    bool _need_iceberg_metadata() const;

};

} // namespace doris::format::iceberg