[Enhancement](load) Add LoadTableReader for FileScannerV2 ingestion

be/src/exec/operator/file_scan_operator.cpp

@@ -145,6 +145,36 @@ bool FileScanLocalState::TEST_should_use_file_scanner_v2(const TQueryOptions& qu
 bool FileScanLocalState::should_use_file_scanner_v2(const TQueryOptions& query_options,
                                                     bool is_load,
                                                     const TFileScanRangeParams& scan_params) {
+    if (is_load) {
+        // ADBC and Fluss are query-only connectors. Their default Parquet format must not
+        // admit them to the ordinary file-load conversion path.
+        if (scan_params.__isset.table_format_params &&
+            (scan_params.table_format_params.table_format_type == "adbc" ||
+             scan_params.table_format_params.table_format_type == "fluss")) {
+            return false;
+        }
+        // Load planners omit query session options. Thrift's default makes an omitted V2
+        // option indistinguishable from false, so ingestion selects by format instead.
+        switch (scan_params.format_type) {
+        case TFileFormatType::FORMAT_PARQUET:
+        case TFileFormatType::FORMAT_ORC:
+        case TFileFormatType::FORMAT_JSON:
+        case TFileFormatType::FORMAT_CSV_PLAIN:
+        case TFileFormatType::FORMAT_CSV_GZ:
+        case TFileFormatType::FORMAT_CSV_BZ2:
+        case TFileFormatType::FORMAT_CSV_LZ4FRAME:
+        case TFileFormatType::FORMAT_CSV_LZ4BLOCK:
+        case TFileFormatType::FORMAT_CSV_LZOP:
+        case TFileFormatType::FORMAT_CSV_DEFLATE:
+        case TFileFormatType::FORMAT_CSV_SNAPPYBLOCK:
+        case TFileFormatType::FORMAT_PROTO:
+        case TFileFormatType::FORMAT_TEXT:
+        case TFileFormatType::FORMAT_WAL:
+            return true;
+        default:
+            return false;
+        }
+    }
     const bool is_transactional_hive =
             scan_params.__isset.table_format_params &&
             scan_params.table_format_params.table_format_type == "transactional_hive";
@@ -174,7 +204,7 @@ bool FileScanLocalState::_can_generate_physical_splits(const TQueryOptions& quer
                                                        bool is_load,
                                                        const TFileScanRangeParams& scan_params,
                                                        const TFileRangeDesc& range) {
-    if (!should_use_file_scanner_v2(query_options, is_load, scan_params)) {
+    if (is_load || !should_use_file_scanner_v2(query_options, is_load, scan_params)) {
         return false;
     }
     const auto format = range.__isset.format_type ? range.format_type : scan_params.format_type;

be/src/exec/scan/file_scanner_v2.cpp

@@ -63,6 +63,7 @@
 #include "format_v2/table/iceberg_position_delete_sys_table_reader.h"
 #include "format_v2/table/iceberg_reader.h"
 #include "format_v2/table/lance_reader.h"
+#include "format_v2/table/load_reader.h"
 #include "format_v2/table/paimon_reader.h"
 #include "format_v2/table/remote_doris_reader.h"
 #include "format_v2/table_reader.h"
@@ -422,6 +423,7 @@ FileScannerV2::FileScannerV2(RuntimeState* state, FileScanLocalState* local_stat
 
 Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs& conjuncts) {
     RETURN_IF_ERROR(Scanner::init(state, conjuncts));
+    _is_load = state->desc_tbl().get_tuple_descriptor(_params->src_tuple_id) != nullptr;
     auto* profile = _local_state->scanner_profile();
     const auto hierarchy = file_scan_profile::ensure_hierarchy(profile);
     _scanner_total_timer = hierarchy.scanner;
@@ -485,7 +487,13 @@ Status FileScannerV2::_open_impl(RuntimeState* state) {
     }
     RETURN_IF_ERROR(_get_next_scan_range(&_first_scan_range));
     if (_first_scan_range) {
-        RETURN_IF_ERROR(_create_table_reader_for_format(_current_range, &_table_reader));
+        if (_is_load) {
+            _table_reader = std::make_unique<format::LoadTableReader>(
+                    state->desc_tbl().get_tuple_descriptor(_params->src_tuple_id),
+                    _output_tuple_desc);
+        } else {
+            RETURN_IF_ERROR(_create_table_reader_for_format(_current_range, &_table_reader));
+        }
         DORIS_CHECK(_table_reader != nullptr);
         RETURN_IF_ERROR(_init_expr_ctxes());
         RETURN_IF_ERROR(_init_table_reader(_current_range));
@@ -1069,7 +1077,18 @@ Status FileScannerV2::_init_expr_ctxes() {
         _slot_id_to_desc.emplace(slot_desc->id(), slot_desc);
     }
     DORIS_CHECK(_table_reader != nullptr);
-    RETURN_IF_ERROR(_build_projected_columns(*_table_reader));
+    if (_is_load) {
+        // Load required_slots describe the source tuple, not this destination tuple.
+        // The wrapper owns source projection, defaults and partition materialization.
+        _projected_columns.clear();
+        for (size_t i = 0; i < _output_tuple_desc->slots().size(); ++i) {
+            const auto* slot = _output_tuple_desc->slots()[i];
+            _projected_columns.push_back(_build_table_column(slot));
+            _slot_id_to_global_index.emplace(slot->id(), format::GlobalIndex(i));
+        }
+    } else {
+        RETURN_IF_ERROR(_build_projected_columns(*_table_reader));
+    }
     return Status::OK();
 }
 
@@ -1156,6 +1175,9 @@ format::ColumnDefinition FileScannerV2::_build_table_column(const SlotDescriptor
 Status FileScannerV2::_build_table_conjuncts(VExprContextSPtrs* conjuncts) const {
     DORIS_CHECK(conjuncts != nullptr);
     conjuncts->clear();
+    if (_is_load) {
+        return Status::OK();
+    }
     conjuncts->reserve(_conjuncts.size());
     for (const auto& conjunct : _conjuncts) {
         VExprSPtr root;
@@ -1522,6 +1544,13 @@ bool FileScannerV2::_should_update_load_counters() const {
 }
 
 void FileScannerV2::_report_file_reader_predicate_filtered_rows() {
+    if (_io_ctx != nullptr && _is_load) {
+        _counter.num_rows_filtered += _io_ctx->load_filtered_rows - _reported_load_filtered_rows;
+        _counter.num_rows_unselected +=
+                _io_ctx->load_unselected_rows - _reported_load_unselected_rows;
+        _reported_load_filtered_rows = _io_ctx->load_filtered_rows;
+        _reported_load_unselected_rows = _io_ctx->load_unselected_rows;
+    }
     const int64_t filtered_rows = _io_ctx != nullptr ? _io_ctx->predicate_filtered_rows : 0;
     const int64_t filtered_delta = filtered_rows - _reported_predicate_filtered_rows;
     if (filtered_delta > 0) {

be/src/exec/scan/file_scanner_v2.h

@@ -281,6 +281,8 @@ class FileScannerV2 final : public Scanner {
     RuntimeProfile::Counter* _adaptive_batch_probe_count_counter = nullptr;
     std::unique_ptr<AdaptiveBlockSizePredictor> _block_size_predictor;
     int64_t _reported_predicate_filtered_rows = 0;
+    int64_t _reported_load_filtered_rows = 0;
+    int64_t _reported_load_unselected_rows = 0;
     int64_t _reported_condition_cache_hit_count = 0;
     int64_t _reported_condition_cache_filtered_rows = 0;
     int64_t _last_read_bytes = 0;

be/src/format_v2/delimited_text/csv_reader.cpp

@@ -159,6 +159,9 @@ Status CsvReader::_create_line_reader() {
 Status CsvReader::_validate_line(const Slice& line) {
     if (_file_format_type != TFileFormatType::FORMAT_PROTO && _enable_text_validate_utf8 &&
         !validate_utf8(line.data, line.size)) {
+        if (_request != nullptr && _request->reject_invalid_rows) {
+            return Status::DataQualityError("Only support csv data in utf8 codec");
+        }
         return Status::InternalError<false>("Only support csv data in utf8 codec");
     }
     return Status::OK();

be/src/format_v2/delimited_text/delimited_text_reader.cpp

@@ -344,7 +344,30 @@ Status DelimitedTextReader::get_block(Block* file_block, size_t* rows, bool* eof
             if (line_eof) {
                 break;
             }
-            RETURN_IF_ERROR(_fill_columns_from_line(line, &columns, rows));
+            const auto status = _fill_columns_from_line(line, &columns, rows);
+            if (!status.ok()) {
+                if (!_request->reject_invalid_rows ||
+                    (!status.is<ErrorCode::DATA_QUALITY_ERROR>() &&
+                     !status.is<ErrorCode::INVALID_ARGUMENT>())) {
+                    return status;
+                }
+                // A bad cell may follow already appended cells. Roll back the entire record.
+                for (auto& column : columns) {
+                    if (column->size() > *rows) {
+                        column->pop_back(column->size() - *rows);
+                    }
+                }
+                RETURN_IF_ERROR(_runtime_state->append_error_msg_to_file(
+                        [&] {
+                            return _scan_params->format_type == TFileFormatType::FORMAT_PROTO
+                                           ? std::string("binary record")
+                                           : line.to_string();
+                        },
+                        [&] { return status.to_string(); }));
+                if (_io_ctx != nullptr) {
+                    ++_io_ctx->load_filtered_rows;
+                }
+            }
         }
     }
 
@@ -623,6 +646,15 @@ Status DelimitedTextReader::_fill_columns_from_line(const Slice& line,
         _split_line(line);
     }
     SCOPED_TIMER(_text_profile.deserialize_time);
+    if (_request->reject_invalid_rows) {
+        const auto expected = cast_set<size_t>(_scan_params->num_of_columns_from_file);
+        const bool ignore_extra = _scan_params->file_attributes.ignore_csv_redundant_col;
+        if (_split_values.size() < expected ||
+            (!ignore_extra && _split_values.size() != expected)) {
+            return Status::DataQualityError("Expected {} fields, found {}", expected,
+                                            _split_values.size());
+        }
+    }
     for (const auto& column : _requested_columns) {
         auto* output = (*columns)[column.block_position.value()].get();
         const int32_t field_index = column.file_column_id.value();

be/src/format_v2/file_reader.h

@@ -165,6 +165,9 @@ struct FileScanRequest {
         }
         return count;
     }
+
+    // Ingestion rejects malformed records instead of aborting a whole file.
+    bool reject_invalid_rows = false;
 };
 
 // Helper for constructing the scan-column layout in FileScanRequest.
@@ -374,6 +377,12 @@ class FileReader {
               _profile(profile) {}
     virtual ~FileReader() = default;
 
+    // Per-batch presence metadata distinguishes absent fields from explicit NULLs. These are
+    // physical field identities; TableColumnMapper owns their translation to table positions.
+    const std::map<LocalColumnId, std::vector<uint8_t>>& missing_fields() const {
+        return _missing_fields;
+    }
+
     // Initialize file reader and parse file metadata.
     virtual Status init(RuntimeState* state);
 
@@ -506,6 +515,7 @@ class FileReader {
     std::unique_ptr<io::FileDescription> _file_description;
     std::shared_ptr<io::IOContext> _io_ctx;
     RuntimeProfile* _profile = nullptr;
+    std::map<LocalColumnId, std::vector<uint8_t>> _missing_fields;
 };
 
 } // namespace doris::format