This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 9ace7cc2851 [Enhancement](load) Add LoadTableReader for FileScannerV2
ingestion (#69320)
9ace7cc2851 is described below
commit 9ace7cc2851c7f4afa8b54a23a96e9c8d6a0b767
Author: Gabriel <[email protected]>
AuthorDate: Mon Oct 26 19:47:44 2026 +0800
[Enhancement](load) Add LoadTableReader for FileScannerV2 ingestion (#69320)
### What problem does this PR solve?
Traditional file loads with source/destination tuples always select
FileScanner V1. Add `LoadTableReader` as a `TableReader` subclass so
supported loads can use FileScannerV2 without FE or Thrift changes.
`LoadTableReader::finalize_chunk()` uses base source materialization,
then applies per-record defaults, source pre-filters, destination
expressions, strict-mode conversion checks, nullability checks, and
flexible partial-update skip bitmaps and sequence-column marks.
`customize_file_scan_request()` enables malformed-record rejection. The
generic TableReader owns file and split lifecycle, with no load-specific
flag, branches, or presence-mask storage. Scanner continues to evaluate
destination conjuncts before ordinary output projections. Source readers
receive no destination predicate, aggregate pushdown, or condition-cache
snapshot. Missing physical source aliases fail before reader open unless
the FE default map declares a fallback; an explicit NULL default remains
valid, and path columns are exempt.
CSV/JSON readers reject malformed load records with row rollback and
error accounting. JSON reports field-presence masks separately from NULL
values so missing defaults and partial updates retain their meaning.
Empty outer arrays and array elements with missing or non-object
configured roots count as rejected load records so the existing
filter-ratio quality gate sees them. Valid siblings remain intact. Fully
rejected batches do not end a split. Non-line JSON streams consume each
pipe message until pipe EOF, including Kafka routine loads; chunked HTTP
buffers remain one document and regular file ranges retain
single-document behavior.
Supported load formats select V2 independently of the query session
option: the Thrift default does not distinguish an omitted option from
false in load plans. Query selection remains unchanged. Avro and Arrow
IPC stream loads keep their existing path; the current V2 Arrow
implementation only supports remote-Doris Flight scans. Load ranges keep
their FE split boundaries.
This PR targets `branch-4.2` and is independent of #69312.
### Validation
- Passed 28 targeted tests in a local ASAN/UBSAN harness, linking
rebuilt changed translation units against existing project libraries.
Coverage includes strict/non-strict conversion, pre-filter order,
explicit NULL versus missing defaults, non-nullability, flexible updates
and sequence marks, rejected-batch EOF, scanner selection, real CSV/JSON
readers through the load reader, missing physical sources versus
explicit NULL defaults and path columns, and ten JSON pipe cases:
multiple Kafka messages, batch boundaries, malformed messages,
fixed-length HTTP, chunked HTTP, empty pipe EOF, empty outer arrays,
missing/null roots, valid siblings around invalid roots, and fully
rejected root messages followed by valid messages. The harness also
executes two existing scanner-selector scenarios against the production
selector.
- Recompiled the changed BE implementation units and syntax-checked the
affected existing scanner, CSV, JSON, and TableReader test suites.
- Passed clang-format 16 checks for all 17 affected C++ source/header
files and `git diff --check`.
- Added parser unit tests and a Stream Load regression covering
target-column WHERE, rejected/unselected counts, JSON defaults, flexible
partial updates, and rejection of empty arrays or invalid configured
roots with max_filter_ratio=0 versus 1. Full BE build and cluster
regression execution remain for CI; they were not run locally.
### Release note
Route supported traditional file loads through FileScannerV2 with a
table reader for load conversion and validation.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- Behavior changed:
- [x] Yes. Supported traditional loads select V2 independently of the
query-only session switch.
- Does this need documentation?
- [x] Yes. Updated the in-repository FileScannerV2 design document.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exec/operator/file_scan_operator.cpp | 32 +-
be/src/exec/scan/file_scanner_v2.cpp | 33 +-
be/src/exec/scan/file_scanner_v2.h | 2 +
be/src/format_v2/delimited_text/csv_reader.cpp | 3 +
.../delimited_text/delimited_text_reader.cpp | 34 +-
be/src/format_v2/file_reader.h | 10 +
be/src/format_v2/json/json_reader.cpp | 87 +-
be/src/format_v2/json/json_reader.h | 4 +-
be/src/format_v2/table/load_reader.cpp | 386 ++++++++
be/src/format_v2/table/load_reader.h | 73 ++
be/src/format_v2/table_reader.h | 21 +-
be/src/io/io_common.h | 3 +
be/test/exec/scan/file_load_scanner_v2_test.cpp | 76 ++
be/test/exec/scan/file_scanner_v2_test.cpp | 9 +-
.../format_v2/delimited_text/csv_reader_test.cpp | 26 +
be/test/format_v2/json/json_reader_test.cpp | 44 +-
be/test/format_v2/table/load_reader_test.cpp | 971 +++++++++++++++++++++
docs/file-scanner-v2-design.md | 37 +-
.../load_p0/stream_load/file_scanner_v2_load.csv | 5 +
.../load_p0/stream_load/file_scanner_v2_load.json | 5 +
.../file_scanner_v2_load_empty_array.json | 2 +
.../file_scanner_v2_load_invalid_roots.json | 1 +
.../stream_load/file_scanner_v2_load_partial.json | 3 +
.../test_csv_with_none_utf8_data.groovy | 3 +-
.../stream_load/test_file_scanner_v2_load.groovy | 157 ++++
.../load_p0/stream_load/test_json_load.groovy | 6 +-
.../stream_load/test_stream_load_error_url.groovy | 3 +-
27 files changed, 2000 insertions(+), 36 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index 396ca3326b8..3a39de474f4 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/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;
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index 2ad9897a6bb..18dc5a6ce49 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -64,6 +64,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"
@@ -423,6 +424,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;
@@ -486,7 +488,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));
@@ -1076,7 +1084,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();
}
@@ -1163,6 +1182,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;
@@ -1529,6 +1551,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) {
diff --git a/be/src/exec/scan/file_scanner_v2.h
b/be/src/exec/scan/file_scanner_v2.h
index dd10d2952b1..b7de342988a 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -281,6 +281,8 @@ private:
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;
diff --git a/be/src/format_v2/delimited_text/csv_reader.cpp
b/be/src/format_v2/delimited_text/csv_reader.cpp
index e5eca54df6f..95144d1cc95 100644
--- a/be/src/format_v2/delimited_text/csv_reader.cpp
+++ b/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();
diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
b/be/src/format_v2/delimited_text/delimited_text_reader.cpp
index fa185600cc7..eacb78942ea 100644
--- a/be/src/format_v2/delimited_text/delimited_text_reader.cpp
+++ b/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();
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index e6a56c508c7..c1f11f2566f 100644
--- a/be/src/format_v2/file_reader.h
+++ b/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 @@ public:
_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 @@ protected:
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
diff --git a/be/src/format_v2/json/json_reader.cpp
b/be/src/format_v2/json/json_reader.cpp
index 78601f90584..390936308ce 100644
--- a/be/src/format_v2/json/json_reader.cpp
+++ b/be/src/format_v2/json/json_reader.cpp
@@ -310,6 +310,7 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
const auto max_block_bytes = _runtime_state->preferred_block_size_bytes();
*rows = 0;
*eof = false;
+ _missing_fields.clear();
while (file_block->rows() < batch_size && !_reader_eof &&
file_block->bytes() < max_block_bytes) {
@@ -355,6 +356,9 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
}
*rows = file_block->rows();
+ for (auto& [_, mask] : _missing_fields) {
+ mask.resize(*rows, 0);
+ }
_record_scan_rows(cast_set<int64_t>(*rows));
{
SCOPED_TIMER(_filter_time);
@@ -542,16 +546,19 @@ Status JsonReader::_read_one_document(size_t* size, bool*
eof) {
_document_view = std::string_view(reinterpret_cast<const char*>(line),
*size);
return Status::OK();
}
- // Non-line mode treats the split as one JSON document. This supports a
single object or an
- // array with strip_outer_array=true.
+ if (_scan_params->file_type == TFileType::FILE_STREAM) {
+ // A routine-load pipe contains one document per message. Only pipe
EOF ends the
+ // stream; treating its first message as a complete file can commit
unread offsets.
+ RETURN_IF_ERROR(_read_one_document_from_pipe(size));
+ *eof = *size == 0;
+ return Status::OK();
+ }
+ // Non-line file mode treats the split as one JSON document, optionally an
outer array.
if (_single_document_read) {
*eof = true;
return Status::OK();
}
_single_document_read = true;
- if (_scan_params->file_type == TFileType::FILE_STREAM) {
- return _read_one_document_from_pipe(size);
- }
auto read_size = _reader_range.size;
if (read_size <= 0 && _reader_range.__isset.file_size) {
@@ -579,6 +586,11 @@ Status JsonReader::_read_one_document_from_pipe(size_t*
read_size) {
}
DorisUniqueBufferPtr<uint8_t> file_buf;
RETURN_IF_ERROR(stream_load_pipe->read_one_message(&file_buf, read_size));
+ if (*read_size == 0) {
+ _document_buffer.clear();
+ _document_view = {};
+ return Status::OK();
+ }
_document_buffer.assign(reinterpret_cast<const char*>(file_buf.get()),
*read_size);
_document_view = _document_buffer;
if (!stream_load_pipe->is_chunked_transfer()) {
@@ -696,13 +708,25 @@ Status JsonReader::_append_simple_json_rows(Block* block,
bool* is_empty_row, bo
if (_json_value.type().value() == simdjson::ondemand::json_type::array) {
_array = _json_value.get_array();
if (_array.count_elements() == 0) {
+ // Empty load documents must contribute to max_filter_ratio just
like malformed rows.
+ if (_request->reject_invalid_rows) {
+ return Status::DataQualityError("Empty json line");
+ }
*is_empty_row = true;
return Status::OK();
}
_array_iter = _array.begin();
while (_array_iter != _array.end()) {
simdjson::ondemand::object object_value =
(*_array_iter).get_object();
- RETURN_IF_ERROR(_set_column_values_from_object(&object_value,
block, &valid));
+ const size_t row_start = block->rows();
+ const auto status = _set_column_values_from_object(&object_value,
block, &valid);
+ if (!status.ok()) {
+ if (!_request->reject_invalid_rows) {
+ return status;
+ }
+ RETURN_IF_ERROR(_handle_json_error(status, block, row_start,
is_empty_row));
+ valid = true;
+ }
++_array_iter;
if (!valid) {
*is_empty_row = true;
@@ -730,27 +754,43 @@ Status
JsonReader::_append_flat_array_jsonpath_rows(Block* block, bool* is_empty
_array = _json_value.get_array();
_array_iter = _array.begin();
while (_array_iter != _array.end()) {
+ const size_t row_start = block->rows();
simdjson::ondemand::object object_value = (*_array_iter).get_object();
if (!_parsed_from_json_root && !_parsed_json_root.empty()) {
- // For strip_outer_array, json_root is evaluated against each
element. Elements without
- // the requested root do not produce rows, matching the load
reader behavior.
+ // Reject roots per element for loads, preserving earlier siblings
and their presence
+ // masks while counting each invalid element against the load
quality gate.
simdjson::ondemand::value rooted_value;
Status st = JsonFunctions::extract_from_object(object_value,
_parsed_json_root,
&rooted_value);
if (!st.ok()) {
if (st.is<ErrorCode::NOT_FOUND>()) {
+ if (_request->reject_invalid_rows) {
+ RETURN_IF_ERROR(
+
_handle_json_error(Status::DataQualityError("{}", st.to_string()),
+ block, row_start,
is_empty_row));
+ }
++_array_iter;
continue;
}
return st;
}
if (rooted_value.type().value() !=
simdjson::ondemand::json_type::object) {
+ if (_request->reject_invalid_rows) {
+
RETURN_IF_ERROR(_handle_json_error(Status::DataQualityError("Not object item"),
+ block, row_start,
is_empty_row));
+ }
++_array_iter;
continue;
}
object_value = rooted_value.get_object();
}
- RETURN_IF_ERROR(_write_columns_by_jsonpath(&object_value, block,
&valid));
+ const auto status = _write_columns_by_jsonpath(&object_value, block,
&valid);
+ if (!status.ok()) {
+ if (!_request->reject_invalid_rows) {
+ return status;
+ }
+ RETURN_IF_ERROR(_handle_json_error(status, block, row_start,
is_empty_row));
+ }
++_array_iter;
}
*is_empty_row = block->rows() == original_rows;
@@ -787,6 +827,10 @@ Status
JsonReader::_set_column_values_from_object(simdjson::ondemand::object* ob
if (column_index == static_cast<size_t>(-1)) {
continue;
}
+ if (_request->reject_invalid_rows &&
+ _requested_columns[column_index].slot_desc->is_skip_bitmap_col()) {
+ continue;
+ }
if (seen_columns[column_index]) {
if (_is_hive_table) {
// Hive JSON keeps the last duplicate key ignoring case. The
earlier value has
@@ -809,6 +853,9 @@ Status
JsonReader::_set_column_values_from_object(simdjson::ondemand::object* ob
has_valid_value = true;
}
+ if (!has_valid_value && _request->reject_invalid_rows) {
+ return Status::DataQualityError("No source column matches this JSON
object");
+ }
for (size_t i = 0; i < _requested_columns.size(); ++i) {
if (seen_columns[i]) {
continue;
@@ -881,6 +928,9 @@ Status
JsonReader::_write_columns_by_jsonpath(simdjson::ondemand::object* object
}
}
+ if (!has_valid_value && _request->reject_invalid_rows) {
+ return Status::DataQualityError("No source column matches the JSON
paths");
+ }
if (!has_valid_value) {
// jsonpaths can legally match nothing. Roll the row back so an
all-missing path set does
// not create a synthetic row of nulls.
@@ -1063,6 +1113,11 @@ Status
JsonReader::_write_data_to_column(simdjson::ondemand::value& value,
Status JsonReader::_fill_missing_column(const RequestedColumn& column,
IColumn* column_ptr,
bool* valid) {
+ if (_request->reject_invalid_rows) {
+ auto& missing = _missing_fields[column.file_column_id];
+ missing.resize(column_ptr->size() + 1, 0);
+ missing.back() = 1;
+ }
if (column.slot_desc->is_nullable()) {
auto* nullable_column = assert_cast<ColumnNullable*>(column_ptr);
nullable_column->insert_default();
@@ -1097,6 +1152,20 @@ Status JsonReader::_handle_json_error(const Status&
status, Block* block, size_t
// block to the row count before this document before either surfacing the
error or appending
// the ignore-malformed null row.
_truncate_block_to_rows(block, original_rows);
+ for (auto& [_, mask] : _missing_fields) {
+ mask.resize(original_rows, 0);
+ }
+ if (_request->reject_invalid_rows &&
+ (status.is<ErrorCode::DATA_QUALITY_ERROR>() ||
status.is<ErrorCode::INVALID_ARGUMENT>())) {
+ RETURN_IF_ERROR(_runtime_state->append_error_msg_to_file(
+ [&] { return std::string(_padding_buffer.data(),
_original_doc_size); },
+ [&] { return status.to_string(); }));
+ if (_io_ctx != nullptr) {
+ ++_io_ctx->load_filtered_rows;
+ }
+ *is_empty_row = true;
+ return Status::OK();
+ }
if (_openx_json_ignore_malformed &&
status.is<ErrorCode::DATA_QUALITY_ERROR>()) {
RETURN_IF_ERROR(_append_null_for_malformed_json(block));
*is_empty_row = false;
diff --git a/be/src/format_v2/json/json_reader.h
b/be/src/format_v2/json/json_reader.h
index f5e6614b2da..c823536d52b 100644
--- a/be/src/format_v2/json/json_reader.h
+++ b/be/src/format_v2/json/json_reader.h
@@ -133,8 +133,8 @@ private:
Status _create_decompressor();
Status _create_line_reader();
Status _parse_jsonpath_and_json_root();
- // Reads one logical JSON document: one line for JSON Lines, or the whole
range/pipe payload for
- // single-document mode.
+ // Reads one logical JSON document: one line, one file range, or one pipe
message.
+ // Chunked HTTP transfers concatenate their pipe buffers into one document.
Status _read_one_document(size_t* size, bool* eof);
Status _read_one_document_from_pipe(size_t* read_size);
// Moves the logical document into a simdjson-padded buffer and creates an
ondemand document.
diff --git a/be/src/format_v2/table/load_reader.cpp
b/be/src/format_v2/table/load_reader.cpp
new file mode 100644
index 00000000000..2f592272045
--- /dev/null
+++ b/be/src/format_v2/table/load_reader.cpp
@@ -0,0 +1,386 @@
+// 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.
+
+#include "format_v2/table/load_reader.h"
+
+#include "core/column/column_complex.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "format_v2/wal/wal_reader.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format {
+
+LoadTableReader::LoadTableReader(TupleDescriptor* source, const
TupleDescriptor* destination)
+ : _source_tuple(source), _destination_tuple(destination),
_source_row(source, false) {}
+
+LoadTableReader::~LoadTableReader() = default;
+
+Status LoadTableReader::init(TableReadOptions&& options) {
+ _scan_params = options.scan_params;
+ _runtime_state = options.runtime_state;
+ _format = options.format;
+ RETURN_IF_ERROR(_init_source_columns());
+ if (_scan_params->__isset.pre_filter_exprs_list) {
+ RETURN_IF_ERROR(
+ VExpr::create_expr_trees(_scan_params->pre_filter_exprs_list,
_pre_filters));
+ } else if (_scan_params->__isset.pre_filter_exprs) {
+ VExprContextSPtr expr;
+
RETURN_IF_ERROR(VExpr::create_expr_tree(_scan_params->pre_filter_exprs, expr));
+ _pre_filters.push_back(std::move(expr));
+ }
+ RETURN_IF_ERROR(VExpr::prepare(_pre_filters, _runtime_state, _source_row));
+ RETURN_IF_ERROR(VExpr::open(_pre_filters, _runtime_state));
+ for (auto* slot : _destination_tuple->slots()) {
+ const auto it = _scan_params->expr_of_dest_slot.find(slot->id());
+ if (it == _scan_params->expr_of_dest_slot.end() ||
it->second.nodes.empty()) {
+ return Status::InvalidArgument("Missing load expression for
destination slot {}",
+ slot->id());
+ }
+ VExprContextSPtr expr;
+ RETURN_IF_ERROR(VExpr::create_expr_tree(it->second, expr));
+ RETURN_IF_ERROR(expr->prepare(_runtime_state, _source_row));
+ RETURN_IF_ERROR(expr->open(_runtime_state));
+ _destination_exprs.push_back(std::move(expr));
+ int source_position = -1;
+ const auto direct =
_scan_params->dest_sid_to_src_sid_without_trans.find(slot->id());
+ if (direct != _scan_params->dest_sid_to_src_sid_without_trans.end()) {
+ for (size_t i = 0; i < _source_tuple->slots().size(); ++i) {
+ if (_source_tuple->slots()[i]->id() == direct->second) {
+ source_position = cast_set<int>(i);
+ break;
+ }
+ }
+ if (source_position < 0) {
+ return Status::InvalidArgument("Unknown load source slot {}",
direct->second);
+ }
+ }
+ _direct_source_positions.push_back(source_position);
+ if (slot->is_sequence_col()) {
+ _sequence_uid = slot->col_unique_id();
+ }
+ }
+ // Destination predicates must run after conversions and error accounting.
Passing them to
+ // the source reader would both bind the wrong slots and change
rejected-row statistics.
+ return TableReader::init({.projected_columns = std::move(_source_columns),
+ .conjuncts = {},
+ .format = options.format,
+ .scan_params = options.scan_params,
+ .io_ctx = options.io_ctx,
+ .runtime_state = options.runtime_state,
+ .scanner_profile = options.scanner_profile,
+ .file_slot_descs = &_source_file_slots,
+ .push_down_agg_type = TPushAggOp::NONE,
+ .condition_cache_digest = 0});
+}
+
+Status LoadTableReader::_init_source_columns() {
+ const auto& slots = _source_tuple->slots();
+ for (size_t i = 0; i < slots.size(); ++i) {
+ auto* slot = slots[i];
+ ColumnDefinition column;
+ column.identifier = Field::create_field<TYPE_STRING>(slot->col_name());
+ column.name = slot->col_name();
+ column.type = slot->get_data_type_ptr();
+ const auto required =
+ std::ranges::find_if(_scan_params->required_slots,
+ [&](const auto& info) { return
info.slot_id == slot->id(); });
+ if (required == _scan_params->required_slots.end()) {
+ return Status::InvalidArgument("Missing load source slot {} in
required slots",
+ slot->id());
+ }
+ if (required->is_file_slot) {
+ _source_file_slots.push_back(slot);
+ } else {
+ column.is_partition_key = true;
+ _partition_positions.push_back(cast_set<int>(i));
+ }
+ const auto default_it =
_scan_params->default_value_of_src_slot.find(slot->id());
+ if (default_it != _scan_params->default_value_of_src_slot.end() &&
+ !default_it->second.nodes.empty()) {
+ RETURN_IF_ERROR(VExpr::create_expr_tree(default_it->second,
column.default_expr));
+ VExprContextSPtr expr;
+ RETURN_IF_ERROR(VExpr::create_expr_tree(default_it->second, expr));
+ RETURN_IF_ERROR(expr->prepare(_runtime_state, _source_row));
+ RETURN_IF_ERROR(expr->open(_runtime_state));
+ _source_defaults.emplace(i, std::move(expr));
+ }
+ if (slot->is_skip_bitmap_col()) {
+ _skip_bitmap_position = cast_set<int>(i);
+ }
+ if (_scan_params->__isset.sequence_map_col &&
+ slot->col_name() == _scan_params->sequence_map_col) {
+ _sequence_map_uid = slot->col_unique_id();
+ }
+ ProjectedColumnBuildContext context {
+ .scan_params = _scan_params, .runtime_state = _runtime_state,
.slot_desc = slot};
+ if (_format == FileFormat::WAL) {
+ if (slot->col_unique_id() < 0) {
+ return Status::InternalError("WAL projected column {} has no
valid unique id",
+ column.name);
+ }
+ // WAL headers identify columns by stable unique id, including
after a rename.
+ column.identifier =
Field::create_field<TYPE_INT>(slot->col_unique_id());
+ } else {
+ RETURN_IF_ERROR(annotate_projected_column(*required, &context,
&column));
+ }
+ _source_columns.push_back(std::move(column));
+ }
+ return Status::OK();
+}
+
+Status LoadTableReader::prepare_split(const SplitReadOptions& options) {
+ auto source_options = options;
+ source_options.conjuncts = VExprContextSPtrs {};
+ source_options.partition_prune_conjuncts.clear();
+ source_options.condition_cache_digest = 0;
+ source_options.condition_cache_split_context.reset();
+ source_options.all_runtime_filters_applied = true;
+ const auto& range = options.current_range;
+ if (_partition_positions.size() > range.columns_from_path.size()) {
+ return Status::InvalidArgument("Load range has {} partition values for
{} source slots",
+ range.columns_from_path.size(),
_partition_positions.size());
+ }
+ for (size_t i = 0; i < _partition_positions.size(); ++i) {
+ const auto* slot = _source_tuple->slots()[_partition_positions[i]];
+ auto column = slot->get_data_type_ptr()->create_column();
+ const bool is_null = range.__isset.columns_from_path_is_null &&
+ i < range.columns_from_path_is_null.size() &&
+ range.columns_from_path_is_null[i];
+ if (is_null) {
+ column->insert_default();
+ } else {
+ Slice value(range.columns_from_path[i]);
+ DataTypeSerDe::FormatOptions serde_options;
+ RETURN_IF_ERROR(
+
slot->get_data_type_ptr()->get_serde()->deserialize_one_cell_from_hive_text(
+ *column, value, serde_options));
+ }
+ Field field;
+ column->get(0, field);
+ source_options.partition_values.insert_or_assign(slot->col_name(),
std::move(field));
+ }
+ return TableReader::prepare_split(source_options);
+}
+
+Status LoadTableReader::refresh_conjuncts(VExprContextSPtrs,
std::optional<uint64_t>, bool) {
+ return Status::OK();
+}
+
+Status LoadTableReader::build_physical_splits(const FileScanSplit&,
std::vector<FileScanSplit>*,
+ bool* was_split) {
+ // Keep ingestion records and their error accounting in the original FE
source range.
+ *was_split = false;
+ return Status::OK();
+}
+
+Status LoadTableReader::_reject_row(size_t row, const std::string& reason) {
+ RETURN_IF_ERROR(_runtime_state->append_error_msg_to_file(
+ [&] { return _source.dump_one_line(row,
_scan_params->num_of_columns_from_file); },
+ [&] { return reason; }));
+ ++_io_ctx->load_filtered_rows;
+ return Status::OK();
+}
+
+Status LoadTableReader::_materialize_missing_values() {
+ const size_t rows = _source.rows();
+ std::map<size_t, const std::vector<uint8_t>*> missing;
+ // Translate presence metadata only through the mapper, while the physical
reader is alive.
+ const auto& fields = _data_reader.reader->missing_fields();
+ for (const auto& mapping : _data_reader.column_mapper->mappings()) {
+ if (mapping.file_local_id.has_value()) {
+ const auto it = fields.find(LocalColumnId(*mapping.file_local_id));
+ if (it != fields.end()) {
+ missing.emplace(mapping.global_index.value(), &it->second);
+ }
+ }
+ }
+ if (_skip_bitmap_position >= 0) {
+ auto bitmap =
+
_source_tuple->slots()[_skip_bitmap_position]->get_data_type_ptr()->create_column();
+ bitmap->insert_many_defaults(rows);
+ auto& nullable = assert_cast<ColumnNullable&>(*bitmap);
+ std::fill(nullable.get_null_map_data().begin(),
nullable.get_null_map_data().end(), 0);
+ auto& values =
assert_cast<ColumnBitmap&>(nullable.get_nested_column()).get_data();
+ auto keep = ColumnUInt8::create(rows, 1);
+ for (const auto& [position, mask] : missing) {
+ const auto* slot = _source_tuple->slots()[position];
+ if (slot->is_skip_bitmap_col()) {
+ continue;
+ }
+ for (size_t row = 0; row < rows; ++row) {
+ if (!(*mask)[row]) {
+ continue;
+ }
+ if (slot->is_key() && !slot->is_auto_increment()) {
+ if (keep->get_data()[row]) {
+ RETURN_IF_ERROR(_reject_row(
+ row, fmt::format("Missing key column {} in
flexible partial update",
+ slot->col_name())));
+ keep->get_data()[row] = 0;
+ }
+ } else {
+ values[row].add(slot->col_unique_id());
+ if (_sequence_map_uid == slot->col_unique_id() &&
_sequence_uid >= 0) {
+ values[row].add(_sequence_uid);
+ }
+ }
+ }
+ }
+ _source.replace_by_position(_skip_bitmap_position, std::move(bitmap));
+ const auto size = _source.columns();
+ _source.insert({std::move(keep), std::make_shared<DataTypeUInt8>(),
"load_valid"});
+ RETURN_IF_ERROR(Block::filter_block(&_source, size, size));
+ } else {
+ for (const auto& [position, mask] : missing) {
+ const auto expr = _source_defaults.find(position);
+ if (expr == _source_defaults.end()) {
+ continue;
+ }
+ ColumnPtr default_values;
+ RETURN_IF_ERROR(expr->second->execute(&_source, default_values));
+ default_values = default_values->convert_to_full_column_if_const();
+ auto& input = _source.get_by_position(position);
+ default_values = input.type->is_nullable() ?
make_nullable(default_values)
+ :
remove_nullable(default_values);
+ auto result = input.type->create_column();
+ for (size_t row = 0; row < rows; ++row) {
+ result->insert_from((*mask)[row] ? *default_values :
*input.column, row);
+ }
+ _source.replace_by_position(position, std::move(result));
+ }
+ }
+ return Status::OK();
+}
+
+Status LoadTableReader::_convert(Block* output) {
+ const size_t rows = _source.rows();
+ auto keep = ColumnUInt8::create(rows, 1);
+ const std::vector<BitmapValue>* skip = nullptr;
+ if (_skip_bitmap_position >= 0) {
+ const auto& nullable = assert_cast<const ColumnNullable&>(
+ *_source.get_by_position(_skip_bitmap_position).column);
+ skip = &assert_cast<const
ColumnBitmap&>(nullable.get_nested_column()).get_data();
+ }
+ for (size_t i = 0; i < _destination_exprs.size(); ++i) {
+ const auto* slot = _destination_tuple->slots()[i];
+ ColumnPtr column;
+ RETURN_IF_ERROR(_destination_exprs[i]->execute(&_source, column));
+ column = column->convert_to_full_column_if_const();
+ for (size_t row = 0; row < rows; ++row) {
+ if (!keep->get_data()[row] || !column->is_null_at(row) ||
+ (skip != nullptr &&
(*skip)[row].contains(slot->col_unique_id()))) {
+ continue;
+ }
+ const int source_position = _direct_source_positions[i];
+ const bool invalid_cast =
+ _scan_params->strict_mode && source_position >= 0 &&
+
!_source.get_by_position(source_position).column->is_null_at(row);
+ if (invalid_cast || !slot->is_nullable()) {
+ RETURN_IF_ERROR(_reject_row(
+ row, fmt::format("Column {} {}", slot->col_name(),
+ invalid_cast ? "cannot be converted
in strict mode"
+ : "is NULL but is not
nullable")));
+ keep->get_data()[row] = 0;
+ }
+ }
+ column = slot->is_nullable() ? make_nullable(column) :
remove_nullable(column);
+ output->replace_by_position(i, std::move(column));
+ }
+ const auto size = output->columns();
+ output->insert({std::move(keep), std::make_shared<DataTypeUInt8>(),
"load_valid"});
+ RETURN_IF_ERROR(Block::filter_block(output, size, size));
+ return Status::OK();
+}
+
+Status LoadTableReader::validate_file_mapping(const TableColumnMapper& mapper)
const {
+ RETURN_IF_ERROR(TableReader::validate_file_mapping(mapper));
+ for (const auto& mapping : mapper.mappings()) {
+ const auto position = mapping.global_index.value();
+ if (mapping.file_local_id.has_value() ||
_projected_columns[position].is_partition_key) {
+ continue;
+ }
+ const auto* slot = _source_tuple->slots()[position];
+ // FE uses an empty expression for an explicit NULL default. An absent
map entry instead
+ // means the file source alias has no fallback and must not silently
materialize as NULL.
+ if (!_scan_params->default_value_of_src_slot.contains(slot->id())) {
+ return Status::InternalError("failed to find default value expr
for slot: {}",
+ slot->col_name());
+ }
+ }
+ return Status::OK();
+}
+
+Status LoadTableReader::finalize_chunk(Block* block, size_t physical_rows) {
+ _source = Block(_source_tuple->slots(), 0);
+ RETURN_IF_ERROR(TableReader::finalize_chunk(&_source, physical_rows));
+ RETURN_IF_ERROR(_materialize_missing_values());
+ const auto rows = _source.rows();
+ RETURN_IF_ERROR(
+ VExprContext::filter_block(_pre_filters, &_source,
_source_tuple->slots().size()));
+ _io_ctx->load_unselected_rows += cast_set<int64_t>(rows - _source.rows());
+ if (_source.rows() > 0) {
+ RETURN_IF_ERROR(_convert(block));
+ }
+ _source.clear();
+ return Status::OK();
+}
+
+Status LoadTableReader::abort_split() {
+ _source.clear();
+ return TableReader::abort_split();
+}
+
+Status LoadTableReader::customize_file_scan_request(FileScanRequest* request) {
+ // Decoders must reject malformed load records before source expressions
run.
+ request->reject_invalid_rows = true;
+ return Status::OK();
+}
+
+TableColumnMappingMode LoadTableReader::mapping_mode() const {
+ return _format == FileFormat::WAL ? TableColumnMappingMode::BY_FIELD_ID
+ : TableColumnMappingMode::BY_NAME;
+}
+
+Status LoadTableReader::create_file_reader(std::unique_ptr<FileReader>*
reader) {
+ if (_format == FileFormat::WAL) {
+ *reader = std::make_unique<wal::WalReader>(_system_properties,
_current_task->data_file,
+ _io_ctx, _scanner_profile,
_projected_columns);
+ return Status::OK();
+ }
+ return TableReader::create_file_reader(reader);
+}
+
+Status LoadTableReader::close() {
+ if (_closed) {
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(TableReader::close());
+ _pre_filters.clear();
+ _destination_exprs.clear();
+ _source_defaults.clear();
+ _source.clear();
+ _closed = true;
+ return Status::OK();
+}
+
+} // namespace doris::format
diff --git a/be/src/format_v2/table/load_reader.h
b/be/src/format_v2/table/load_reader.h
new file mode 100644
index 00000000000..1c3d5421ace
--- /dev/null
+++ b/be/src/format_v2/table/load_reader.h
@@ -0,0 +1,73 @@
+// 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 "format_v2/table_reader.h"
+
+namespace doris::format {
+
+// Adapts the existing file-load source/destination tuple protocol to a table
reader.
+// Source materialization uses TableReader; finalization returns the
destination schema.
+class LoadTableReader : public TableReader {
+public:
+ LoadTableReader(TupleDescriptor* source, const TupleDescriptor*
destination);
+ ~LoadTableReader() override;
+
+ Status init(TableReadOptions&& options) override;
+ Status prepare_split(const SplitReadOptions& options) override;
+ Status refresh_conjuncts(VExprContextSPtrs conjuncts,
+ std::optional<uint64_t> digest = std::nullopt,
+ bool all_runtime_filters_applied = false)
override;
+ Status build_physical_splits(const FileScanSplit& source,
std::vector<FileScanSplit>* splits,
+ bool* was_split) override;
+ bool current_split_uses_metadata_count() const override { return false; }
+ Status abort_split() override;
+ Status close() override;
+ int64_t condition_cache_hit_count() const override { return 0; }
+
+protected:
+ Status finalize_chunk(Block* block, size_t rows) override;
+ Status validate_file_mapping(const TableColumnMapper& mapper) const
override;
+ Status customize_file_scan_request(FileScanRequest* request) override;
+ Status create_file_reader(std::unique_ptr<FileReader>* reader) override;
+ TableColumnMappingMode mapping_mode() const override;
+
+private:
+ Status _init_source_columns();
+ Status _materialize_missing_values();
+ Status _convert(Block* output);
+ Status _reject_row(size_t row, const std::string& reason);
+
+ TupleDescriptor* _source_tuple;
+ const TupleDescriptor* _destination_tuple;
+ RowDescriptor _source_row;
+ Block _source;
+ std::vector<ColumnDefinition> _source_columns;
+ std::vector<SlotDescriptor*> _source_file_slots;
+ VExprContextSPtrs _pre_filters;
+ VExprContextSPtrs _destination_exprs;
+ std::map<size_t, VExprContextSPtr> _source_defaults;
+ std::vector<int> _direct_source_positions;
+ std::vector<int> _partition_positions;
+ int _skip_bitmap_position = -1;
+ int _sequence_map_uid = -1;
+ int _sequence_uid = -1;
+ bool _closed = false;
+};
+
+} // namespace doris::format
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 2c3879cb198..b71f1be3c99 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -287,8 +287,7 @@ public:
virtual Status get_block(Block* block, bool* eos) {
SCOPED_TIMER(_profile.total_timer);
SCOPED_TIMER(_profile.exec_timer);
- DORIS_CHECK(block->columns() == _projected_columns.size());
- block->clear_column_data(_projected_columns.size());
+ block->clear_column_data();
while (true) {
if (*eos) {
@@ -358,12 +357,7 @@ public:
#ifndef NDEBUG
RETURN_IF_ERROR(_check_file_block_columns("after file reader
get_block", current_rows));
#endif
- DORIS_CHECK(block->columns() ==
_data_reader.column_mapper->mappings().size());
RETURN_IF_ERROR(finalize_chunk(block, current_rows));
-#ifndef NDEBUG
- RETURN_IF_ERROR(
- _check_table_block_columns("after finalize_chunk", block,
current_rows));
-#endif
if (current_eof) {
_current_reader_reached_eof = !stopped_during_read;
RETURN_IF_ERROR(close_current_reader());
@@ -826,9 +820,12 @@ protected:
}
}
- // Finalize file-local block to table/global schema block.
- Status finalize_chunk(Block* block, const size_t rows) {
+ // Materialize the projected schema. Subclasses may then transform its
columns and rows
+ // into their output schema; the active reader remains alive until this
hook returns.
+ virtual Status finalize_chunk(Block* block, const size_t rows) {
SCOPED_TIMER(_profile.finalize_timer);
+ DORIS_CHECK(block->columns() == _projected_columns.size());
+ DORIS_CHECK(block->columns() ==
_data_reader.column_mapper->mappings().size());
size_t idx = 0;
const auto& mappings = _data_reader.column_mapper->mappings();
for (const auto& mapping : mappings) {
@@ -842,6 +839,10 @@ protected:
// Enforce CHAR/VARCHAR length declared by the table schema after all
file-to-table
// materialization has finished.
RETURN_IF_ERROR(_truncate_char_or_varchar_columns(block));
+#ifndef NDEBUG
+ // Validate the materialized schema before a subclass transforms rows
or output columns.
+ RETURN_IF_ERROR(_check_table_block_columns("after table
materialization", block, rows));
+#endif
return Status::OK();
}
@@ -1076,11 +1077,11 @@ protected:
DORIS_CHECK(block != nullptr);
DORIS_CHECK(pushed_down != nullptr);
*pushed_down = false;
- block->clear_column_data(_projected_columns.size());
_aggregate_pushdown_tried = true;
if (_row_ids.has_value() ||
!_supports_aggregate_pushdown(_push_down_agg_type)) {
return Status::OK();
}
+ block->clear_column_data(_projected_columns.size());
FileAggregateRequest file_request;
RETURN_IF_ERROR(_build_file_aggregate_request(_push_down_agg_type,
&file_request));
diff --git a/be/src/io/io_common.h b/be/src/io/io_common.h
index 19d66efde70..4bcab929411 100644
--- a/be/src/io/io_common.h
+++ b/be/src/io/io_common.h
@@ -259,6 +259,9 @@ struct IOContext {
// index reads and data reads on the normal write-back path.
bool inverted_index_snii_read_no_write_file_cache = false;
RemoteScanCacheWriteLimiter* remote_scan_cache_write_limiter = nullptr; //
Ref
+ // Scanner-local ingestion counters shared by physical decoding and load
materialization.
+ int64_t load_filtered_rows = 0;
+ int64_t load_unselected_rows = 0;
};
} // namespace io
diff --git a/be/test/exec/scan/file_load_scanner_v2_test.cpp
b/be/test/exec/scan/file_load_scanner_v2_test.cpp
new file mode 100644
index 00000000000..2cce04108b9
--- /dev/null
+++ b/be/test/exec/scan/file_load_scanner_v2_test.cpp
@@ -0,0 +1,76 @@
+// 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.
+
+#include <gtest/gtest.h>
+
+#include "exec/operator/file_scan_operator.h"
+
+namespace doris {
+
+TEST(FileLoadScannerV2Test, LoadUsesV2WithoutQueryOption) {
+ TQueryOptions options;
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_CSV_PLAIN);
+ // Stream and routine load plans need not carry the external-query session
option.
+ EXPECT_TRUE(FileScanLocalState::should_use_file_scanner_v2(options, true,
params));
+ EXPECT_FALSE(FileScanLocalState::should_use_file_scanner_v2(options,
false, params));
+}
+
+TEST(FileLoadScannerV2Test, LoadSelectionDoesNotDependOnQueryOption) {
+ TQueryOptions options;
+ TFileScanRangeParams params;
+ for (auto format : {TFileFormatType::FORMAT_CSV_PLAIN,
TFileFormatType::FORMAT_JSON,
+ TFileFormatType::FORMAT_PARQUET,
TFileFormatType::FORMAT_ORC,
+ TFileFormatType::FORMAT_PROTO,
TFileFormatType::FORMAT_WAL,
+ TFileFormatType::FORMAT_TEXT}) {
+ params.__set_format_type(format);
+ options.__set_enable_file_scanner_v2(true);
+ EXPECT_TRUE(FileScanLocalState::should_use_file_scanner_v2(options,
true, params));
+ options.__set_enable_file_scanner_v2(false);
+ EXPECT_TRUE(FileScanLocalState::should_use_file_scanner_v2(options,
true, params));
+ }
+ options.__set_enable_file_scanner_v2(true);
+ for (auto format : {TFileFormatType::FORMAT_AVRO,
TFileFormatType::FORMAT_ARROW}) {
+ params.__set_format_type(format);
+ EXPECT_FALSE(FileScanLocalState::should_use_file_scanner_v2(options,
true, params));
+ }
+}
+
+// A connector marker must not become a file-load path just because its
default format is Parquet.
+TEST(FileLoadScannerV2Test, QueryOnlyConnectorsAreNotSupportedLoads) {
+ for (const auto* table_format : {"adbc", "fluss"}) {
+ TFileScanRangeParams params;
+ TTableFormatFileDesc table;
+ table.__set_table_format_type(table_format);
+ params.__set_table_format_params(table);
+ TQueryOptions options;
+ EXPECT_FALSE(FileScanLocalState::should_use_file_scanner_v2(options,
true, params));
+ for (bool enabled : {false, true}) {
+ options.__set_enable_file_scanner_v2(enabled);
+ for (auto format : {TFileFormatType::FORMAT_PARQUET,
TFileFormatType::FORMAT_ORC,
+ TFileFormatType::FORMAT_CSV_PLAIN,
TFileFormatType::FORMAT_JSON,
+ TFileFormatType::FORMAT_WAL,
TFileFormatType::FORMAT_JNI,
+ TFileFormatType::FORMAT_ARROW}) {
+ params.__set_format_type(format);
+
EXPECT_FALSE(FileScanLocalState::should_use_file_scanner_v2(options, true,
params))
+ << table_format << ": " << format << ", V2 option=" <<
enabled;
+ }
+ }
+ }
+}
+
+} // namespace doris
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index ef7c6ab19e2..fba2acb97cd 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -428,7 +428,7 @@ TEST(FileScannerV2Test,
IcebergPositionDeletesSupportNativeFormats) {
EXPECT_FALSE(FileScannerV2::is_supported(params, avro_position_delete));
}
-TEST(FileScannerV2Test, FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
+TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesAndLoads) {
TQueryOptions query_options;
TFileScanRangeParams params;
params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
@@ -437,7 +437,7 @@ TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
query_options.__set_enable_file_scanner_v2(true);
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
-
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
params.__set_format_type(TFileFormatType::FORMAT_WAL);
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
@@ -472,7 +472,7 @@ TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
params.__isset.hive_parquet_time_zone = false;
params.__set_parquet_timestamp_semantics_version(1);
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
-
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
// Paimon keeps FORMAT_JNI at scan level even when its ranges are native
Parquet files.
params.format_type = TFileFormatType::FORMAT_JNI;
params.__set_paimon_predicate("encoded-predicate");
@@ -502,7 +502,8 @@ TEST(FileScannerV2Test,
IcebergOrcDefaultCannotHideDeferredParquetRanges) {
// Scanner construction precedes remote split delivery; the default says
nothing about
// retained Parquet files after changing write.format.default to ORC.
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(options,
false, params));
- EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(options,
true, params));
+ // ORC loads select V2 independently of the external-query session option.
+ EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(options,
true, params));
params.__isset.parquet_timestamp_semantics_version = false;
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(options,
false, params));
params.__set_hive_parquet_time_zone("");
diff --git a/be/test/format_v2/delimited_text/csv_reader_test.cpp
b/be/test/format_v2/delimited_text/csv_reader_test.cpp
index 8c537b1e60e..32d4255acee 100644
--- a/be/test/format_v2/delimited_text/csv_reader_test.cpp
+++ b/be/test/format_v2/delimited_text/csv_reader_test.cpp
@@ -651,6 +651,32 @@ TEST_F(CsvV2ReaderTest,
SchemaSynthesizesComplexChildrenForColumnMapper) {
EXPECT_EQ(schema[2].children[1].children[1].name, "b");
}
+TEST_F(CsvV2ReaderTest,
LoadRejectsWrongFieldCountsWithoutShiftingFollowingRows) {
+ std::ofstream output(_file_path, std::ios::binary);
+ output <<
"id,name,score\n1,alice,10\n2,bob\n3,carol,30,extra\n4,dave,40\n";
+ output.close();
+ _params.__set_num_of_columns_from_file(3);
+ auto io_ctx = std::make_shared<io::IOContext>();
+ auto reader = create_reader(_file_path, &_params, _slots, &_state,
&_profile, 0, -1,
+ TFileCompressType::UNKNOWN, io_ctx);
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<FileScanRequest>();
+ request->reject_invalid_rows = true;
+ for (int i = 0; i < 3; ++i) {
+ request->local_positions.emplace(LocalColumnId(i), LocalIndex(i));
+ }
+ ASSERT_TRUE(reader->open(request).ok());
+ auto block = make_block(schema, {0, 1, 2});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 2);
+ EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 0), 1);
+ EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 1), 4);
+ EXPECT_EQ(io_ctx->load_filtered_rows, 2);
+}
+
// Scenario: CSV v2 honors FileScanRequest local positions, so TableReader can
request a subset of
// CSV fields in an order different from the physical CSV field order.
TEST_F(CsvV2ReaderTest, ReadsRequestedColumnsInFileLocalBlockOrder) {
diff --git a/be/test/format_v2/json/json_reader_test.cpp
b/be/test/format_v2/json/json_reader_test.cpp
index 1d94e244f64..18b1ff2946b 100644
--- a/be/test/format_v2/json/json_reader_test.cpp
+++ b/be/test/format_v2/json/json_reader_test.cpp
@@ -176,13 +176,15 @@ struct ReadResult {
size_t second_rows = 0;
bool second_eof = false;
size_t document_buffer_size = 0;
+ int64_t rejected_rows = 0;
+ std::map<LocalColumnId, std::vector<uint8_t>> missing_fields;
std::vector<ColumnDefinition> schema;
};
ReadResult read_once(const std::string& file_name, const std::string& content,
TFileScanRangeParams params, const
std::vector<SlotDescriptor*>& slots,
const std::vector<int32_t>& requested_local_ids, bool
read_twice = false,
- bool is_hive_table = false) {
+ bool is_hive_table = false, bool reject_invalid_rows =
false) {
const auto file_path = write_json_file(file_name, content);
auto range = file_range(file_path);
if (is_hive_table) {
@@ -196,7 +198,8 @@ ReadResult read_once(const std::string& file_name, const
std::string& content,
auto desc = file_description(file_path.string());
RuntimeProfile profile("json_v2_reader_test");
MockRuntimeState state;
- JsonReader reader(system_properties, desc, nullptr, &profile, ¶ms,
range, slots);
+ auto io_ctx = std::make_shared<io::IOContext>();
+ JsonReader reader(system_properties, desc, io_ctx, &profile, ¶ms,
range, slots);
ReadResult result;
result.status = reader.init(&state);
@@ -209,6 +212,7 @@ ReadResult read_once(const std::string& file_name, const
std::string& content,
}
auto request = std::make_shared<FileScanRequest>();
+ request->reject_invalid_rows = reject_invalid_rows;
for (size_t i = 0; i < requested_local_ids.size(); ++i) {
request->local_positions.emplace(LocalColumnId(requested_local_ids[i]),
LocalIndex(i));
}
@@ -220,6 +224,8 @@ ReadResult read_once(const std::string& file_name, const
std::string& content,
result.block = make_block(result.schema, requested_local_ids);
result.status = reader.get_block(&result.block, &result.rows, &result.eof);
result.document_buffer_size = reader.TEST_document_buffer_size();
+ result.rejected_rows = io_ctx->load_filtered_rows;
+ result.missing_fields = reader.missing_fields();
if (result.status.ok() && read_twice) {
auto eof_block = make_block(result.schema, requested_local_ids);
result.second_status =
@@ -712,6 +718,40 @@ TEST(JsonReaderTest,
RejectsInvalidFileScanRequestsBeforeOpeningFile) {
EXPECT_NE(status.to_string().find("reader is not open"),
std::string::npos);
}
+TEST(JsonReaderTest, LoadRejectsMalformedRecordsAndPreservesFieldPresence) {
+ ObjectPool pool;
+ auto slots = build_slots(&pool);
+ auto result = read_once("load_missing.jsonl",
+ "not-json\n"
+ R"({"id":1})"
+ "\n"
+ R"({"id":2,"name":null})"
+ "\n"
+ R"({"id":3,"name":"ok"})"
+ "\n",
+ json_scan_params(), slots, {1, 0}, true, false,
true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ ASSERT_EQ(result.rows, 3);
+ EXPECT_EQ(result.rejected_rows, 1);
+ EXPECT_TRUE(result.block.get_by_position(0).column->is_null_at(0));
+ EXPECT_TRUE(result.block.get_by_position(0).column->is_null_at(1));
+ EXPECT_EQ(result.missing_fields.at(LocalColumnId(1)),
(std::vector<uint8_t> {1, 0, 0}));
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, LoadRejectsUnmatchedArrayElementWithoutLosingSiblings) {
+ ObjectPool pool;
+ auto slots = build_slots(&pool);
+ auto result = read_once("load_array.json",
+
R"([{"id":1,"name":"a"},{"unknown":2},{"id":3,"name":"c"}])",
+ json_scan_params(false, true), slots, {0, 1},
true, false, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_EQ(result.rejected_rows, 1);
+ EXPECT_TRUE(result.second_eof);
+}
+
TEST(JsonReaderTest, ReturnsErrorForMalformedJsonByDefault) {
ObjectPool pool;
auto slots = build_slots(&pool);
diff --git a/be/test/format_v2/table/load_reader_test.cpp
b/be/test/format_v2/table/load_reader_test.cpp
new file mode 100644
index 00000000000..a88931bd4d6
--- /dev/null
+++ b/be/test/format_v2/table/load_reader_test.cpp
@@ -0,0 +1,971 @@
+// 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.
+
+#include "format_v2/table/load_reader.h"
+
+#include <gtest/gtest.h>
+#include <unistd.h>
+
+#include <filesystem>
+#include <fstream>
+
+#include "core/column/column_complex.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/data_type/data_type_bitmap.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "io/fs/kafka_consumer_pipe.h"
+#include "load/stream_load/new_load_stream_mgr.h"
+#include "load/stream_load/stream_load_context.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format {
+namespace {
+
+TTypeDesc scalar_type(const DataTypePtr& type) {
+ TScalarType scalar;
+ scalar.__set_type(doris::to_thrift(type->get_primitive_type()));
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ node.__set_scalar_type(scalar);
+ TTypeDesc result;
+ result.__set_types({node});
+ return result;
+}
+
+struct SourceReader {
+ std::vector<ColumnDefinition> columns;
+ bool all_null = false;
+ bool missing_first = false;
+ bool omit_first_column = false;
+ bool eof_with_rows = false;
+ int opens = 0;
+ int closes = 0;
+};
+
+class SourceFileReader final : public FileReader {
+public:
+ SourceFileReader(std::shared_ptr<io::FileSystemProperties>& properties,
+ std::unique_ptr<io::FileDescription>& file,
std::shared_ptr<io::IOContext> io,
+ RuntimeProfile* profile, SourceReader& source)
+ : FileReader(properties, file, io, profile), _source(source) {}
+ Status init(RuntimeState*) override { return Status::OK(); }
+ Status get_schema(std::vector<ColumnDefinition>* schema) const override {
+ *schema = _source.columns;
+ if (_source.omit_first_column) {
+ schema->erase(schema->begin());
+ }
+ for (size_t i = 0; i < schema->size(); ++i) {
+ (*schema)[i].local_id = cast_set<int32_t>(i);
+ (*schema)[i].default_expr.reset();
+ }
+ return Status::OK();
+ }
+ Status open(std::shared_ptr<FileScanRequest> request) override {
+ EXPECT_TRUE(request->reject_invalid_rows);
+ EXPECT_TRUE(request->conjuncts.empty());
+ ++_source.opens;
+ return FileReader::open(std::move(request));
+ }
+ Status get_block(Block* block, size_t* rows, bool* eos) override {
+ block->clear_column_data();
+ _missing_fields.clear();
+ *rows = 0;
+ *eos = _calls++ != 0;
+ if (*eos) {
+ return Status::OK();
+ }
+ auto column = IColumn::mutate(block->get_by_position(0).column);
+ auto& nullable = assert_cast<ColumnNullable&>(*column);
+ if (_source.all_null) {
+ nullable.insert_default();
+ } else {
+ nullable.get_nested_column().insert_data("value", 5);
+ nullable.get_null_map_data().push_back(0);
+ }
+ nullable.insert_default();
+ block->replace_by_position(0, std::move(column));
+ for (size_t i = 1; i < block->columns(); ++i) {
+ auto other = block->get_by_position(i).type->create_column();
+ other->insert_many_defaults(2);
+ block->replace_by_position(i, std::move(other));
+ }
+ if (_source.missing_first) {
+ _missing_fields.emplace(LocalColumnId(0), std::vector<uint8_t> {1,
0});
+ }
+ *rows = 2;
+ *eos = _source.eof_with_rows;
+ return Status::OK();
+ }
+ Status close() override {
+ ++_source.closes;
+ return Status::OK();
+ }
+
+private:
+ SourceReader& _source;
+ int _calls = 0;
+};
+
+class TestLoadTableReader final : public LoadTableReader {
+public:
+ TestLoadTableReader(std::unique_ptr<SourceReader> source_reader,
TupleDescriptor* source,
+ TupleDescriptor* destination)
+ : LoadTableReader(source, destination),
_input(std::move(source_reader)) {}
+ Status init(TableReadOptions&& options) override {
+ RETURN_IF_ERROR(LoadTableReader::init(std::move(options)));
+ _input->columns = _projected_columns;
+ return Status::OK();
+ }
+ Status prepare_split(const SplitReadOptions& options) override {
+ auto split = options;
+ split.current_range.__set_path("load-reader-test.csv");
+ split.current_range.__set_file_type(TFileType::FILE_LOCAL);
+
split.current_range.__set_format_type(TFileFormatType::FORMAT_CSV_PLAIN);
+ split.current_range.__set_size(10);
+ split.current_range.__set_file_size(10);
+ split.current_split_format = FileFormat::CSV;
+ return LoadTableReader::prepare_split(split);
+ }
+
+protected:
+ Status create_file_reader(std::unique_ptr<FileReader>* reader) override {
+ *reader = std::make_unique<SourceFileReader>(_system_properties,
_current_task->data_file,
+ _io_ctx,
_scanner_profile, *_input);
+ return Status::OK();
+ }
+
+private:
+ std::unique_ptr<SourceReader> _input;
+};
+
+class LoadReaderTest : public testing::Test {
+protected:
+ void SetUp() override {
+ TTupleDescriptor tuple;
+ tuple.__set_id(0);
+ tuple.__set_byteSize(0);
+ tuple.__set_numNullBytes(1);
+ source = std::make_unique<TupleDescriptor>(tuple);
+ tuple.__set_id(1);
+ destination = std::make_unique<TupleDescriptor>(tuple);
+ TSlotDescriptor slot;
+ slot.__set_id(10);
+ slot.__set_parent(0);
+
slot.__set_slotType(scalar_type(make_nullable(std::make_shared<DataTypeString>())));
+ slot.__set_columnPos(0);
+ slot.__set_slotIdx(0);
+ slot.__set_byteOffset(0);
+ slot.__set_nullIndicatorByte(0);
+ slot.__set_nullIndicatorBit(0);
+ slot.__set_isMaterialized(true);
+ slot.__set_colName("raw_value");
+ source_definition = slot;
+ source_slot = std::make_unique<SlotDescriptor>(slot);
+ source->add_slot(source_slot.get());
+ slot.__set_id(20);
+ slot.__set_parent(1);
+ slot.__set_colName("value");
+ destination_slot = std::make_unique<SlotDescriptor>(slot);
+ destination->add_slot(destination_slot.get());
+ params.__set_src_tuple_id(0);
+ params.__set_dest_tuple_id(1);
+ params.__set_num_of_columns_from_file(1);
+ TFileScanSlotInfo required;
+ required.__set_slot_id(10);
+ required.__set_is_file_slot(true);
+ params.__set_required_slots({required});
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::SLOT_REF);
+ node.__set_type(scalar_type(source_slot->type()));
+ node.__set_num_children(0);
+ node.__set_is_nullable(true);
+ TSlotRef ref;
+ ref.__set_slot_id(10);
+ ref.__set_tuple_id(0);
+ node.__set_slot_ref(ref);
+ TExpr expression;
+ expression.__set_nodes({node});
+ params.__set_expr_of_dest_slot({{20, expression}});
+ params.__set_dest_sid_to_src_sid_without_trans({{20, 10}});
+ }
+ void set_destination_type(DataTypePtr type) {
+ TTupleDescriptor tuple;
+ tuple.__set_id(1);
+ tuple.__set_byteSize(0);
+ tuple.__set_numNullBytes(1);
+ destination = std::make_unique<TupleDescriptor>(tuple);
+ TSlotDescriptor slot;
+ slot.__set_id(20);
+ slot.__set_parent(1);
+ slot.__set_slotType(scalar_type(type));
+ slot.__set_columnPos(0);
+ slot.__set_slotIdx(0);
+ slot.__set_byteOffset(0);
+ slot.__set_nullIndicatorByte(0);
+ slot.__set_nullIndicatorBit(type->is_nullable() ? 0 : -1);
+ slot.__set_isMaterialized(true);
+ slot.__set_colName("value");
+ destination_slot = std::make_unique<SlotDescriptor>(slot);
+ destination->add_slot(destination_slot.get());
+ }
+ void cast_destination_to_int() {
+ set_destination_type(make_nullable(std::make_shared<DataTypeInt32>()));
+ TExprNode cast;
+ cast.__set_node_type(TExprNodeType::CAST_EXPR);
+ cast.__set_type(scalar_type(destination_slot->get_data_type_ptr()));
+ cast.__set_is_nullable(true);
+ cast.__set_num_children(1);
+
params.expr_of_dest_slot[20].nodes.insert(params.expr_of_dest_slot[20].nodes.begin(),
cast);
+ }
+ Status init(LoadTableReader& reader, FileFormat format = FileFormat::CSV) {
+ TTupleDescriptor tuple;
+ tuple.__set_id(0);
+ tuple.__set_byteSize(0);
+ tuple.__set_numNullBytes(1);
+ TDescriptorTable descriptors;
+ descriptors.__set_tupleDescriptors({tuple});
+ descriptors.__set_slotDescriptors({source_definition});
+ if (source_bitmap) {
+ descriptors.slotDescriptors.push_back(bitmap_definition);
+ }
+ if (anchor_slot) {
+ descriptors.slotDescriptors.push_back(anchor_definition);
+ }
+ DescriptorTbl* table = nullptr;
+ RETURN_IF_ERROR(DescriptorTbl::create(&pool, descriptors, &table));
+ state.set_desc_tbl(table);
+ return reader.init({.projected_columns = {},
+ .conjuncts = {},
+ .format = format,
+ .scan_params = ¶ms,
+ .io_ctx = io,
+ .runtime_state = &state,
+ .scanner_profile = &profile});
+ }
+ void add_partial_update_columns(bool key) {
+ TTupleDescriptor tuple;
+ tuple.__set_id(0);
+ tuple.__set_byteSize(0);
+ tuple.__set_numNullBytes(1);
+ source = std::make_unique<TupleDescriptor>(tuple);
+ source_definition.__set_col_unique_id(7);
+ source_definition.__set_is_key(key);
+ source_slot = std::make_unique<SlotDescriptor>(source_definition);
+ source->add_slot(source_slot.get());
+ TSlotDescriptor bitmap = source_definition;
+ bitmap.__set_id(11);
+ bitmap.__set_slotIdx(1);
+ bitmap.__set_columnPos(1);
+ bitmap.__set_is_key(false);
+ bitmap.__set_col_unique_id(9);
+ bitmap.__set_colName("__DORIS_SKIP_BITMAP_COL__");
+
bitmap.__set_slotType(scalar_type(make_nullable(std::make_shared<DataTypeBitMap>())));
+ bitmap_definition = bitmap;
+ source_bitmap = std::make_unique<SlotDescriptor>(bitmap);
+ source->add_slot(source_bitmap.get());
+ TFileScanSlotInfo required;
+ required.__set_slot_id(11);
+ required.__set_is_file_slot(true);
+ params.required_slots.push_back(required);
+ params.__set_num_of_columns_from_file(2);
+ params.__set_sequence_map_col("raw_value");
+ bitmap.__set_id(21);
+ bitmap.__set_parent(1);
+ destination_bitmap = std::make_unique<SlotDescriptor>(bitmap);
+ destination->add_slot(destination_bitmap.get());
+ TExpr bitmap_expr = params.expr_of_dest_slot[20];
+ bitmap_expr.nodes[0].__set_type(bitmap.slotType);
+ bitmap_expr.nodes[0].slot_ref.__set_slot_id(11);
+ params.expr_of_dest_slot.emplace(21, bitmap_expr);
+ TSlotDescriptor sequence = source_definition;
+ sequence.__set_id(22);
+ sequence.__set_parent(1);
+ sequence.__set_slotIdx(2);
+ sequence.__set_columnPos(2);
+ sequence.__set_is_key(false);
+ sequence.__set_col_unique_id(8);
+ sequence.__set_colName("__DORIS_SEQUENCE_COL__");
+ destination_sequence = std::make_unique<SlotDescriptor>(sequence);
+ destination->add_slot(destination_sequence.get());
+ params.expr_of_dest_slot.emplace(22, params.expr_of_dest_slot[20]);
+ }
+ void add_anchor_source_column() {
+ anchor_definition = source_definition;
+ anchor_definition.__set_id(12);
+ anchor_definition.__set_slotIdx(1);
+ anchor_definition.__set_columnPos(1);
+ anchor_definition.__set_colName("anchor");
+ anchor_slot = std::make_unique<SlotDescriptor>(anchor_definition);
+ source->add_slot(anchor_slot.get());
+ TFileScanSlotInfo required;
+ required.__set_slot_id(12);
+ required.__set_is_file_slot(true);
+ params.required_slots.push_back(required);
+ params.__set_num_of_columns_from_file(2);
+ }
+ void TearDown() override {
+ if (!input_path.empty()) {
+ std::filesystem::remove(input_path);
+ }
+ }
+ std::filesystem::path input_path;
+ ObjectPool pool;
+ TSlotDescriptor source_definition, bitmap_definition, anchor_definition;
+ std::unique_ptr<SlotDescriptor> anchor_slot;
+ std::unique_ptr<SlotDescriptor> source_bitmap, destination_bitmap,
destination_sequence;
+ RuntimeState state;
+ RuntimeProfile profile {"load"};
+ std::shared_ptr<io::IOContext> io = std::make_shared<io::IOContext>();
+ TFileScanRangeParams params;
+ std::unique_ptr<TupleDescriptor> source, destination;
+ std::unique_ptr<SlotDescriptor> source_slot, destination_slot;
+};
+
+class LoadJsonPipeTest : public LoadReaderTest {
+protected:
+ void SetUp() override {
+ LoadReaderTest::SetUp();
+ auto* env = ExecEnv::GetInstance();
+ if (!env->new_load_stream_mgr()) {
+ env->_new_load_stream_mgr = std::make_unique<NewLoadStreamMgr>();
+ _owns_stream_manager = true;
+ }
+ _context = std::make_shared<StreamLoadContext>(env);
+ _pipe = std::make_shared<io::KafkaConsumerPipe>();
+ _context->pipe = _pipe;
+ ASSERT_TRUE(env->new_load_stream_mgr()->put(_context->id,
_context).ok());
+ params.__set_file_type(TFileType::FILE_STREAM);
+ params.__set_format_type(TFileFormatType::FORMAT_JSON);
+ params.__set_compress_type(TFileCompressType::PLAIN);
+ TFileAttributes attributes;
+ attributes.__set_read_json_by_line(false);
+ params.__set_file_attributes(attributes);
+ }
+ void TearDown() override {
+ ExecEnv::GetInstance()->new_load_stream_mgr()->remove(_context->id);
+ _context.reset();
+ _pipe.reset();
+ if (_owns_stream_manager) {
+ ExecEnv::GetInstance()->_new_load_stream_mgr.reset();
+ }
+ LoadReaderTest::TearDown();
+ }
+ Status prepare(LoadTableReader& reader) {
+ RETURN_IF_ERROR(init(reader, FileFormat::JSON));
+ SplitReadOptions split;
+ split.current_split_format = FileFormat::JSON;
+ split.current_range.__set_file_type(TFileType::FILE_STREAM);
+ split.current_range.__set_format_type(TFileFormatType::FORMAT_JSON);
+ split.current_range.__set_load_id(_context->id.to_thrift());
+ split.current_range.__set_size(-1);
+ return reader.prepare_split(split);
+ }
+ void append(const std::string& document) {
+ ASSERT_TRUE(_pipe->append_json(document.data(), document.size()).ok());
+ }
+ std::shared_ptr<io::KafkaConsumerPipe> _pipe;
+ std::shared_ptr<StreamLoadContext> _context;
+ bool _owns_stream_manager = false;
+};
+
+TEST_F(LoadJsonPipeTest, ReadsEveryKafkaMessageBeforePipeEof) {
+ append(R"({"raw_value":"first"})");
+ append(R"({"raw_value":"second"})");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"first");
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(1).to_string(),
"second");
+ EXPECT_EQ(_pipe->get_queue_size(), 0);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, MessagesContinueAcrossBatchBoundaries) {
+ auto options = state.query_options();
+ options.__set_batch_size(1);
+ state.set_query_options(options);
+ append(R"({"raw_value":"first"})");
+ append(R"({"raw_value":"second"})");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ for (const auto* expected : {"first", "second"}) {
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 1);
+
EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
expected);
+ }
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, FixedLengthStreamEndsAfterItsDocument) {
+ const std::string document = R"({"raw_value":"fixed"})";
+ auto pipe = std::make_shared<io::StreamLoadPipe>(1024, 1024,
document.size());
+ _context->pipe = pipe;
+ ASSERT_TRUE(pipe->append_and_flush(document.data(), document.size()).ok());
+ ASSERT_TRUE(pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"fixed");
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, RejectedMessageDoesNotHideLaterMessages) {
+ append(R"({"raw_value":"first"})");
+ append("{");
+ append(R"({"raw_value":"last"})");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(1).to_string(),
"last");
+ EXPECT_EQ(io->load_filtered_rows, 1);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, ChunkedTransferRemainsOneDocument) {
+ _pipe->set_is_chunked_transfer(true);
+ append(R"({"raw_)");
+ append(R"(value":"joined"})");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"joined");
+ EXPECT_EQ(io->load_filtered_rows, 0);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, EmptyFinishedPipeReturnsEof) {
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_EQ(io->load_filtered_rows, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, MissingPhysicalSourceAliasRequiresDeclaredDefault) {
+ add_anchor_source_column();
+ params.__set_strict_mode(true);
+ auto input = std::make_unique<SourceReader>();
+ input->omit_first_column = true;
+ auto* physical = input.get();
+ TestLoadTableReader reader(std::move(input), source.get(),
destination.get());
+ ASSERT_TRUE(init(reader).ok());
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ const auto status = reader.get_block(&output, &eos);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("raw_value"), std::string::npos);
+ EXPECT_EQ(physical->opens, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, MissingPhysicalSourceAcceptsExplicitNullDefault) {
+ add_anchor_source_column();
+ params.__set_strict_mode(true);
+ // An empty expression is FE's explicit NULL default, not an absent map
entry.
+ params.__set_default_value_of_src_slot({{10, TExpr {}}});
+ auto input = std::make_unique<SourceReader>();
+ input->omit_first_column = true;
+ TestLoadTableReader reader(std::move(input), source.get(),
destination.get());
+ ASSERT_TRUE(init(reader).ok());
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(0));
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ EXPECT_EQ(io->load_filtered_rows, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, PartitionSourceDoesNotRequireFileDefault) {
+ add_anchor_source_column();
+ params.required_slots[0].__set_is_file_slot(false);
+ params.__set_num_of_columns_from_file(1);
+ auto input = std::make_unique<SourceReader>();
+ input->omit_first_column = true;
+ TestLoadTableReader reader(std::move(input), source.get(),
destination.get());
+ ASSERT_TRUE(init(reader).ok());
+ SplitReadOptions split;
+ split.current_range.__set_columns_from_path({"partition-value"});
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"partition-value");
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, EmptyOuterArrayIsRejectedBeforeFollowingMessage) {
+ params.file_attributes.__set_strip_outer_array(true);
+ append("[]");
+ append(R"([{"raw_value":"valid"}])");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"valid");
+ EXPECT_EQ(io->load_filtered_rows, 1);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(io->load_filtered_rows, 1);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, ConfiguredRootCountsMissingAndNullElements) {
+ params.file_attributes.__set_strip_outer_array(true);
+ params.file_attributes.__set_json_root("$.payload");
+ params.file_attributes.__set_jsonpaths(R"(["$.k"])");
+ append(R"([{"other":1},{"payload":null},{"payload":{"k":2}}])");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"2");
+ EXPECT_EQ(io->load_filtered_rows, 2);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(io->load_filtered_rows, 2);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, RootRejectionPreservesEarlierAndLaterSiblings) {
+ params.file_attributes.__set_strip_outer_array(true);
+ params.file_attributes.__set_json_root("$.payload");
+ params.file_attributes.__set_jsonpaths(R"(["$.k"])");
+ append(R"([{"payload":{"k":1}},{"other":1},{"payload":null},
+ {"payload":3},{"payload":[]},{"payload":{"k":2}}])");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"1");
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(1).to_string(),
"2");
+ EXPECT_EQ(io->load_filtered_rows, 4);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadJsonPipeTest, RejectedRootsDoNotHideFollowingMessage) {
+ params.file_attributes.__set_strip_outer_array(true);
+ params.file_attributes.__set_json_root("$.payload");
+ params.file_attributes.__set_jsonpaths(R"(["$.k"])");
+ append(R"([{"other":1},{"payload":null}])");
+ append(R"([{"payload":{"k":2}}])");
+ ASSERT_TRUE(_pipe->finish().ok());
+ LoadTableReader reader(source.get(), destination.get());
+ const auto status = prepare(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"2");
+ EXPECT_EQ(io->load_filtered_rows, 2);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, SourceSlotsAreConvertedBeforeDestinationFiltering) {
+ auto child = std::make_unique<SourceReader>();
+ auto* input = child.get();
+ TestLoadTableReader reader(std::move(child), source.get(),
destination.get());
+ auto status = init(reader);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(input->columns.size(), 1);
+ EXPECT_EQ(input->columns[0].name, "raw_value");
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"value");
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, StrictCastRejectsInvalidNonNullSourceOnly) {
+ cast_destination_to_int();
+ params.__set_strict_mode(true);
+ TestLoadTableReader reader(std::make_unique<SourceReader>(), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(0));
+ EXPECT_EQ(io->load_filtered_rows, 1);
+ EXPECT_EQ(io->load_unselected_rows, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, NonStrictCastPreservesNullableRows) {
+ cast_destination_to_int();
+ params.__set_strict_mode(false);
+ TestLoadTableReader reader(std::make_unique<SourceReader>(), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(0));
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ EXPECT_EQ(io->load_filtered_rows, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, PreFilterRunsBeforeStrictConversion) {
+ cast_destination_to_int();
+ params.__set_strict_mode(true);
+ TExprNode predicate;
+ predicate.__set_node_type(TExprNodeType::BOOL_LITERAL);
+ predicate.__set_type(scalar_type(std::make_shared<DataTypeUInt8>()));
+ predicate.__set_num_children(0);
+ predicate.__set_is_nullable(false);
+ TBoolLiteral literal;
+ literal.__set_value(false);
+ predicate.__set_bool_literal(literal);
+ TExpr expression;
+ expression.__set_nodes({predicate});
+ params.__set_pre_filter_exprs_list({expression});
+ TestLoadTableReader reader(std::make_unique<SourceReader>(), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_EQ(io->load_filtered_rows, 0);
+ EXPECT_EQ(io->load_unselected_rows, 2);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, MissingDefaultDoesNotReplaceExplicitNull) {
+ TExprNode literal;
+ literal.__set_node_type(TExprNodeType::STRING_LITERAL);
+ literal.__set_type(scalar_type(source_slot->get_data_type_ptr()));
+ literal.__set_num_children(0);
+ literal.__set_is_nullable(true);
+ TStringLiteral value;
+ value.__set_value("fallback");
+ literal.__set_string_literal(value);
+ TExpr expression;
+ expression.__set_nodes({literal});
+ params.__set_default_value_of_src_slot({{10, expression}});
+ auto child = std::make_unique<SourceReader>();
+ child->missing_first = true;
+ TestLoadTableReader reader(std::move(child), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"fallback");
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, NonNullableDestinationFiltersNulls) {
+ set_destination_type(std::make_shared<DataTypeString>());
+ TestLoadTableReader reader(std::make_unique<SourceReader>(), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ ASSERT_EQ(output.rows(), 1);
+ EXPECT_FALSE(output.get_by_position(0).column->is_nullable());
+ EXPECT_EQ(io->load_filtered_rows, 1);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, FullyRejectedBatchIsNotEofAndNextSplitStillReads) {
+ set_destination_type(std::make_shared<DataTypeString>());
+ auto child = std::make_unique<SourceReader>();
+ auto* input = child.get();
+ input->all_null = true;
+ input->eof_with_rows = true;
+ TestLoadTableReader reader(std::move(child), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_EQ(io->load_filtered_rows, 2);
+ EXPECT_EQ(input->opens, 1);
+ EXPECT_EQ(input->closes, 1);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ input->all_null = false;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ eos = false;
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(output.rows(), 1);
+ EXPECT_EQ(io->load_filtered_rows, 3);
+ EXPECT_EQ(input->opens, 2);
+ EXPECT_EQ(input->closes, 2);
+ ASSERT_TRUE(reader.close().ok());
+ ASSERT_TRUE(reader.close().ok());
+ EXPECT_EQ(input->closes, 2);
+}
+
+TEST_F(LoadReaderTest, FlexibleUpdatePreservesOmittedValueAndSequenceMarks) {
+ add_partial_update_columns(false);
+ auto child = std::make_unique<SourceReader>();
+ child->missing_first = true;
+ TestLoadTableReader reader(std::move(child), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 2);
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*output.get_by_position(1).column);
+ const auto& bitmaps = assert_cast<const
ColumnBitmap&>(nullable.get_nested_column()).get_data();
+ EXPECT_TRUE(bitmaps[0].contains(7));
+ EXPECT_TRUE(bitmaps[0].contains(8));
+ EXPECT_FALSE(bitmaps[1].contains(7));
+ EXPECT_FALSE(bitmaps[1].contains(8));
+ EXPECT_EQ(io->load_filtered_rows, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, FlexibleUpdateRejectsMissingKeyBeforeExpressions) {
+ add_partial_update_columns(true);
+ auto child = std::make_unique<SourceReader>();
+ child->missing_first = true;
+ TestLoadTableReader reader(std::move(child), source.get(),
destination.get());
+ const auto init_status = init(reader);
+ ASSERT_TRUE(init_status.ok()) << init_status;
+ ASSERT_TRUE(reader.prepare_split({}).ok());
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ auto status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(output.rows(), 1);
+ EXPECT_EQ(io->load_filtered_rows, 1);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadReaderTest, NativeReadersPreserveValidRowsAroundMalformedRecords) {
+ cast_destination_to_int();
+ params.__set_strict_mode(true);
+ params.__set_file_type(TFileType::FILE_LOCAL);
+ params.__set_compress_type(TFileCompressType::PLAIN);
+ TFileTextScanRangeParams text;
+ text.__set_column_separator(",");
+ text.__set_line_delimiter("\n");
+ TFileAttributes attributes;
+ attributes.__set_text_params(text);
+ attributes.__set_read_json_by_line(true);
+ params.__set_file_attributes(attributes);
+ for (const bool json : {false, true}) {
+ SCOPED_TRACE(json ? "JSON" : "CSV");
+ params.__set_format_type(json ? TFileFormatType::FORMAT_JSON
+ : TFileFormatType::FORMAT_CSV_PLAIN);
+ input_path = std::filesystem::temp_directory_path() /
+ ("v2_load_reader_" + std::to_string(getpid()));
+ {
+ std::ofstream file(input_path);
+ file << (json ?
"{\"raw_value\":1}\n{\"raw_value\":\"bad\"}\nnot-json\n"
+ "{\"raw_value\":null}\n{\"raw_value\":3}\n"
+ : "1\nbad\n2,extra\n\\N\n3\n");
+ }
+ io->load_filtered_rows = 0;
+ const auto format = json ? FileFormat::JSON : FileFormat::CSV;
+ LoadTableReader reader(source.get(), destination.get());
+ auto status = init(reader, format);
+ ASSERT_TRUE(status.ok()) << status;
+ TFileRangeDesc range;
+ range.__set_path(input_path.string());
+ range.__set_start_offset(0);
+ range.__set_size(std::filesystem::file_size(input_path));
+ range.__set_file_size(range.size);
+ range.__set_file_type(TFileType::FILE_LOCAL);
+ range.__set_format_type(params.format_type);
+ SplitReadOptions split;
+ split.current_range = range;
+ split.current_split_format = format;
+ status = reader.prepare_split(split);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_FALSE(eos);
+ ASSERT_EQ(output.rows(), 3);
+ EXPECT_EQ(assert_cast<const ColumnInt32&>(
+ assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column)
+ .get_nested_column())
+ .get_element(0),
+ 1);
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ EXPECT_EQ(assert_cast<const ColumnInt32&>(
+ assert_cast<const
ColumnNullable&>(*output.get_by_position(0).column)
+ .get_nested_column())
+ .get_element(2),
+ 3);
+ EXPECT_EQ(io->load_filtered_rows, 2);
+ ASSERT_TRUE(reader.get_block(&output, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_TRUE(reader.close().ok());
+ }
+}
+
+TEST_F(LoadReaderTest,
NativeJsonMissingFieldUsesDefaultButExplicitNullDoesNot) {
+ add_anchor_source_column();
+ params.__set_format_type(TFileFormatType::FORMAT_JSON);
+ params.__set_file_type(TFileType::FILE_LOCAL);
+ params.__set_compress_type(TFileCompressType::PLAIN);
+ TFileTextScanRangeParams text;
+ text.__set_line_delimiter("\n");
+ TFileAttributes attributes;
+ attributes.__set_text_params(text);
+ attributes.__set_read_json_by_line(true);
+ params.__set_file_attributes(attributes);
+ TExprNode literal;
+ literal.__set_node_type(TExprNodeType::STRING_LITERAL);
+ literal.__set_type(scalar_type(source_slot->get_data_type_ptr()));
+ literal.__set_num_children(0);
+ literal.__set_is_nullable(true);
+ TStringLiteral value;
+ value.__set_value("fallback");
+ literal.__set_string_literal(value);
+ TExpr expression;
+ expression.__set_nodes({literal});
+ params.__set_default_value_of_src_slot({{10, expression}});
+ input_path = std::filesystem::temp_directory_path() /
+ ("v2_load_defaults_" + std::to_string(getpid()));
+ {
+ std::ofstream file(input_path);
+ file << "{\"anchor\":\"a\"}\n{\"anchor\":\"b\",\"raw_value\":null}\n";
+ }
+ LoadTableReader reader(source.get(), destination.get());
+ auto status = init(reader, FileFormat::JSON);
+ ASSERT_TRUE(status.ok()) << status;
+ SplitReadOptions split;
+ split.current_split_format = FileFormat::JSON;
+ split.current_range.__set_path(input_path.string());
+ split.current_range.__set_start_offset(0);
+ split.current_range.__set_size(std::filesystem::file_size(input_path));
+ split.current_range.__set_file_size(split.current_range.size);
+ split.current_range.__set_file_type(TFileType::FILE_LOCAL);
+ split.current_range.__set_format_type(params.format_type);
+ status = reader.prepare_split(split);
+ ASSERT_TRUE(status.ok()) << status;
+ Block output(destination->slots(), 0);
+ bool eos = false;
+ status = reader.get_block(&output, &eos);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"fallback");
+ EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1));
+ EXPECT_EQ(io->load_filtered_rows, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+} // namespace
+} // namespace doris::format
diff --git a/docs/file-scanner-v2-design.md b/docs/file-scanner-v2-design.md
index b2fe43ae76f..833dc9a94f2 100644
--- a/docs/file-scanner-v2-design.md
+++ b/docs/file-scanner-v2-design.md
@@ -27,7 +27,8 @@ boundaries rather than isolated acceleration for a particular
format.
- Do not force every optimization onto every reader.
- Do not expose file-local column positions to the query layer.
- Do not sacrifice error semantics in order to continue execution.
-- Full support for the Load path is currently out of scope.
+- Do not embed ingestion conversion and validation in Scanner or physical
decoders.
+ `LoadTableReader` adapts the existing source/destination tuple protocol at
the table-reader layer.
> **Design placement rule:** The correct layer for a capability depends on
> whether it manages the
> query, manages a Split, restores table semantics, or interprets a physical
> file. Layer boundaries
@@ -337,6 +338,40 @@ flowchart LR
> restore table semantics, and how to account for cost, allowing correctness,
> performance, and
> extensibility to evolve independently.
+## File Load Adapter
+
+Traditional `FileLoadScanNode` plans keep their existing FE and Thrift
representation.
+`LoadTableReader` derives from `TableReader` and consumes `src_tuple_id`,
`required_slots`,
+source defaults, pre-filters, destination expressions and strict-mode
metadata. It returns the
+original destination tuple, so Scanner continues to apply destination
conjuncts before its
+ordinary output projections.
+
+The base reader receives no destination predicates, aggregate pushdown or
condition-cache
+snapshot. `LoadTableReader::finalize_chunk()` invokes base finalization to
materialize and validate
+the source schema, including path columns, then applies per-record defaults or
flexible-update
+missing-column marks, source pre-filters, destination expressions and
strict/nullability checks.
+The base read loop owns file opening, batch progression and EOF, without any
load-specific state
+or branches. Source schema checks run before the finalization hook changes
columns or row counts.
+`LoadTableReader::customize_file_scan_request()` enables malformed-record
rejection before the
+physical reader opens. Rejected and unselected rows are counted separately
through scanner-local
+IOContext counters. Physical text decoders report malformed records through
the same rejected-row
+channel.
+
+JSON field-presence metadata is separate from NULL values and from the
immutable read request.
+FileReader reports masks keyed by physical column identity; LoadTableReader
translates them using
+TableColumnMapper during finalization, before the physical reader closes.
Missing keys in a
+flexible update are rejected, while omitted values and an omitted sequence-map
column retain the
+corresponding skip-bitmap marks. A fully rejected batch is not EOF.
+Non-line JSON pipes read successive messages until pipe EOF; chunked HTTP
buffers are still
+combined into one document. File ranges in non-line mode remain
single-document inputs.
+
+Supported traditional file loads select V2 even when their planner omits query
session options.
+`enable_file_scanner_v2` continues to control query scans; Thrift does not
distinguish an omitted
+option from its default false value for load plans. Formats outside the load
adapter's selection
+remain on their existing path, including Avro and Arrow IPC loads. V2 Arrow
support is currently
+limited to the remote-Doris table reader. Load ranges retain FE split
boundaries rather than
+publishing additional physical children.
+
## Further Reading
- [FileScannerV2 profiling and pruning
PR](https://github.com/apache/doris/pull/65449)
diff --git a/regression-test/data/load_p0/stream_load/file_scanner_v2_load.csv
b/regression-test/data/load_p0/stream_load/file_scanner_v2_load.csv
new file mode 100644
index 00000000000..ba0383b9fb9
--- /dev/null
+++ b/regression-test/data/load_p0/stream_load/file_scanner_v2_load.csv
@@ -0,0 +1,5 @@
+1,10
+2,bad
+3,30
+4,40
+5
diff --git a/regression-test/data/load_p0/stream_load/file_scanner_v2_load.json
b/regression-test/data/load_p0/stream_load/file_scanner_v2_load.json
new file mode 100644
index 00000000000..d6197ecf5e8
--- /dev/null
+++ b/regression-test/data/load_p0/stream_load/file_scanner_v2_load.json
@@ -0,0 +1,5 @@
+{"k":1}
+{"k":2,"v":null}
+{"k":3,"v":"bad"}
+not-json
+{"k":4,"v":40}
diff --git
a/regression-test/data/load_p0/stream_load/file_scanner_v2_load_empty_array.json
b/regression-test/data/load_p0/stream_load/file_scanner_v2_load_empty_array.json
new file mode 100644
index 00000000000..fb73f316f70
--- /dev/null
+++
b/regression-test/data/load_p0/stream_load/file_scanner_v2_load_empty_array.json
@@ -0,0 +1,2 @@
+[]
+[{"k":1}]
diff --git
a/regression-test/data/load_p0/stream_load/file_scanner_v2_load_invalid_roots.json
b/regression-test/data/load_p0/stream_load/file_scanner_v2_load_invalid_roots.json
new file mode 100644
index 00000000000..6520cb83153
--- /dev/null
+++
b/regression-test/data/load_p0/stream_load/file_scanner_v2_load_invalid_roots.json
@@ -0,0 +1 @@
+[{"other":1},{"payload":null},{"payload":{"k":2}}]
diff --git
a/regression-test/data/load_p0/stream_load/file_scanner_v2_load_partial.json
b/regression-test/data/load_p0/stream_load/file_scanner_v2_load_partial.json
new file mode 100644
index 00000000000..9769562ab14
--- /dev/null
+++ b/regression-test/data/load_p0/stream_load/file_scanner_v2_load_partial.json
@@ -0,0 +1,3 @@
+{"k":1}
+{"k":2,"v":null}
+{"v":9}
diff --git
a/regression-test/suites/load_p0/stream_load/test_csv_with_none_utf8_data.groovy
b/regression-test/suites/load_p0/stream_load/test_csv_with_none_utf8_data.groovy
index 736399e5800..9950e2b8df7 100644
---
a/regression-test/suites/load_p0/stream_load/test_csv_with_none_utf8_data.groovy
+++
b/regression-test/suites/load_p0/stream_load/test_csv_with_none_utf8_data.groovy
@@ -55,7 +55,8 @@ suite("test_csv_with_none_utf8_data", "p0") {
def json = parseJson(result)
def (code, out, err) = curl("GET", json.ErrorURL)
log.info("error result: " + out)
- def checkError = out.contains("Invalid file encoding: all CSV
files must be UTF-8 encoded")
+ // Supported loads use V2 diagnostics; keep the rejection-count
checks below.
+ def checkError = out.contains("[DATA_QUALITY_ERROR]Only support
csv data in utf8 codec")
assertTrue(checkError)
assertEquals("fail", json.Status.toLowerCase())
assertTrue(json.Message.contains("too many filtered rows"))
diff --git
a/regression-test/suites/load_p0/stream_load/test_file_scanner_v2_load.groovy
b/regression-test/suites/load_p0/stream_load/test_file_scanner_v2_load.groovy
new file mode 100644
index 00000000000..c6fea8fc8dd
--- /dev/null
+++
b/regression-test/suites/load_p0/stream_load/test_file_scanner_v2_load.groovy
@@ -0,0 +1,157 @@
+// 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.
+
+suite("test_file_scanner_v2_load", "p0") {
+ def tableName = "test_file_scanner_v2_load"
+ def checkRows = { result, exception, loaded, filtered, unselected ->
+ if (exception != null) {
+ throw exception
+ }
+ def response = parseJson(result)
+ assertEquals("Success", response.Status)
+ assertEquals(loaded, response.NumberLoadedRows as int)
+ assertEquals(filtered, response.NumberFilteredRows as int)
+ assertEquals(unselected, response.NumberUnselectedRows as int)
+ }
+
+ try {
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql """CREATE TABLE ${tableName} (k INT NOT NULL, v INT NULL DEFAULT
'7')
+ DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES("replication_num" = "1")"""
+ streamLoad {
+ table tableName
+ set 'column_separator', ','
+ set 'columns', 'raw_k,v,k=cast(raw_k as int)+10'
+ set 'where', 'k = 13'
+ set 'strict_mode', 'true'
+ set 'max_filter_ratio', '1'
+ file 'file_scanner_v2_load.csv'
+ check { result, exception, startTime, endTime ->
+ // Conversion errors must be counted even if the target WHERE
rejects that row.
+ checkRows(result, exception, 1, 2, 2)
+ }
+ }
+ sql "sync"
+ assertEquals([[13, 30]], sql("SELECT k, v FROM ${tableName} ORDER BY
k"))
+
+ sql "TRUNCATE TABLE ${tableName}"
+ streamLoad {
+ table tableName
+ set 'format', 'json'
+ set 'read_json_by_line', 'true'
+ set 'strict_mode', 'true'
+ set 'max_filter_ratio', '1'
+ file 'file_scanner_v2_load.json'
+ check { result, exception, startTime, endTime ->
+ checkRows(result, exception, 3, 2, 0)
+ }
+ }
+ sql "sync"
+ assertEquals([[1, 7], [2, null], [4, 40]], sql("SELECT k, v FROM
${tableName} ORDER BY k"))
+
+ ["0", "1"].each { ratio ->
+ sql "TRUNCATE TABLE ${tableName}"
+ streamLoad {
+ table tableName
+ set 'format', 'json'
+ set 'read_json_by_line', 'true'
+ set 'strip_outer_array', 'true'
+ set 'max_filter_ratio', ratio
+ file 'file_scanner_v2_load_empty_array.json'
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ def response = parseJson(result)
+ assertEquals(2, response.NumberTotalRows as int)
+ assertEquals(1, response.NumberFilteredRows as int)
+ assertEquals(0, response.NumberUnselectedRows as int)
+ if (ratio == "0") {
+ // Empty arrays must reach the quality gate even when
the next row is valid.
+ assertEquals("Fail", response.Status)
+ assertEquals(0, response.NumberLoadedRows as int)
+ assertTrue(response.Message.contains("too many
filtered rows"))
+ } else {
+ checkRows(result, exception, 1, 1, 0)
+ }
+ }
+ }
+ sql "sync"
+ assertEquals(ratio == "0" ? [] : [[1, 7]],
+ sql("SELECT k, v FROM ${tableName} ORDER BY k"))
+ }
+
+ ["0", "1"].each { ratio ->
+ sql "TRUNCATE TABLE ${tableName}"
+ streamLoad {
+ table tableName
+ set 'format', 'json'
+ set 'read_json_by_line', 'true'
+ set 'strip_outer_array', 'true'
+ set 'json_root', '$.payload'
+ set 'jsonpaths', '["$.k"]'
+ set 'columns', 'k'
+ set 'max_filter_ratio', ratio
+ file 'file_scanner_v2_load_invalid_roots.json'
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ def response = parseJson(result)
+ assertEquals(3, response.NumberTotalRows as int)
+ assertEquals(2, response.NumberFilteredRows as int)
+ assertEquals(0, response.NumberUnselectedRows as int)
+ if (ratio == "0") {
+ // Missing and non-object roots must each count
against the quality gate.
+ assertEquals("Fail", response.Status)
+ assertEquals(0, response.NumberLoadedRows as int)
+ assertTrue(response.Message.contains("too many
filtered rows"))
+ } else {
+ checkRows(result, exception, 1, 2, 0)
+ }
+ }
+ }
+ sql "sync"
+ assertEquals(ratio == "0" ? [] : [[2, 7]],
+ sql("SELECT k, v FROM ${tableName} ORDER BY k"))
+ }
+
+ sql "DROP TABLE ${tableName}"
+ sql """CREATE TABLE ${tableName} (k INT NOT NULL, v INT NULL)
+ UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES("replication_num" = "1",
"enable_unique_key_merge_on_write" = "true",
+ "enable_unique_key_skip_bitmap_column" = "true")"""
+ sql "INSERT INTO ${tableName} VALUES (1, 10), (2, 20)"
+ streamLoad {
+ table tableName
+ set 'format', 'json'
+ set 'read_json_by_line', 'true'
+ set 'strict_mode', 'false'
+ set 'max_filter_ratio', '1'
+ set 'unique_key_update_mode', 'UPDATE_FLEXIBLE_COLUMNS'
+ file 'file_scanner_v2_load_partial.json'
+ check { result, exception, startTime, endTime ->
+ checkRows(result, exception, 2, 1, 0)
+ }
+ }
+ sql "sync"
+ assertEquals([[1, 10], [2, null]], sql("SELECT k, v FROM ${tableName}
ORDER BY k"))
+ } finally {
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ }
+}
diff --git a/regression-test/suites/load_p0/stream_load/test_json_load.groovy
b/regression-test/suites/load_p0/stream_load/test_json_load.groovy
index 79ed17b16bb..f57e63b849f 100644
--- a/regression-test/suites/load_p0/stream_load/test_json_load.groovy
+++ b/regression-test/suites/load_p0/stream_load/test_json_load.groovy
@@ -766,7 +766,8 @@ suite("test_json_load", "p0,nonConcurrent") {
def code = process.waitFor()
def out = process.text
log.info("result: ${out}".toString())
- def reason = "Reason: There is no column matching jsonpaths in
the json file, columns:[name, age, agent_id, ], please check columns and
jsonpaths:[\"\$.Name\", \"\$.Age\", \"\$.Agent_id\"]. src line
[{\"name\":\"Name1\",\"age\":21,\"agent_id\":\"1\"}]; \n"
+ // V2 distinguishes unmatched JSON paths from unmatched object
fields.
+ def reason = "Reason: [DATA_QUALITY_ERROR]No source column
matches the JSON paths. src line
[{\"name\":\"Name1\",\"age\":21,\"agent_id\":\"1\"}]; \n"
assertEquals("${reason}", "${out}")
}
}
@@ -802,7 +803,8 @@ suite("test_json_load", "p0,nonConcurrent") {
def code = process.waitFor()
def out = process.text
log.info("result: ${out}".toString())
- def reason = "Reason: There is no column matching jsonpaths in
the json file, columns:[Name, Age, Agent_id, ], please check columns and
jsonpaths:. src line [{\"name\":\"Name1\",\"age\":21,\"agent_id\":\"1\"}]; \n"
+ // Case-sensitive source aliases take V2's unmatched-object
path.
+ def reason = "Reason: [DATA_QUALITY_ERROR]No source column
matches this JSON object. src line
[{\"name\":\"Name1\",\"age\":21,\"agent_id\":\"1\"}]; \n"
assertEquals("${reason}", "${out}")
}
}
diff --git
a/regression-test/suites/load_p0/stream_load/test_stream_load_error_url.groovy
b/regression-test/suites/load_p0/stream_load/test_stream_load_error_url.groovy
index 270b494b1f0..a9cf96cc113 100644
---
a/regression-test/suites/load_p0/stream_load/test_stream_load_error_url.groovy
+++
b/regression-test/suites/load_p0/stream_load/test_stream_load_error_url.groovy
@@ -66,7 +66,8 @@ suite("test_stream_load_error_url", "p0") {
assertTrue(json.Message.contains("[DATA_QUALITY_ERROR]too many
filtered rows"))
def (code, out, err) = curl("GET", json.ErrorURL)
log.info("error result: " + out)
- assertTrue(out.contains("Column count mismatch: expected"))
+ // V2 reports physical field counts before destination
conversion.
+ assertTrue(out.contains("[DATA_QUALITY_ERROR]Expected 3
fields, found 9"))
log.info("url: " + json.ErrorURL)
if (isCloudMode()) {
assertTrue(json.ErrorURL.contains("X-Amz-Signature=") ||
json.ErrorURL.contains("error_log"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
