This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 9af1840dcd0 [improvement](load) Add LoadTableReader for FileScannerV2
ingestion (#69333)
9af1840dcd0 is described below
commit 9af1840dcd07e73f322bcd7320282ef1dc900bbb
Author: Gabriel <[email protected]>
AuthorDate: Mon Oct 26 10:23:38 2026 +0800
[improvement](load) Add LoadTableReader for FileScannerV2 ingestion (#69333)
### What problem does this PR solve?
Add `LoadTableReader` so supported traditional file loads use
FileScannerV2 with the existing FE/Thrift source and destination tuples.
Source defaults, pre-filters, destination conversion, strict/nullability
checks, and flexible partial-update bitmaps run in the table-reader
finalization hook. CSV/JSON malformed records are rejected with rollback
and load counters. JSON pipes consume every message until EOF; empty
arrays and invalid configured roots contribute to the load quality gate
without losing valid siblings.
Master-specific adaptations:
- Preserve master's ADBC/Fluss V2 query selection and Lance exclusion.
- Match master's single-argument `TableReader::refresh_conjuncts()` and
`RowDescriptor` constructors, and use `SlotDescriptor::type()`.
- Omit the physical-split override and shared split-cache field, which
are absent on master. Loads retain the existing FE split boundaries.
Avro and Arrow IPC loads keep their existing path. No FE or Thrift
changes. This port is independent of #69312.
### Validation
- Passed the repository clang-format 16 check, covering all 17 changed
C++ source/header files.
- Passed `build-support/check-build-hygiene.sh` and `git diff --cached
--check`.
- Passed focused Clang syntax checks for all 11 changed C++ translation
units and `table_reader.cpp`, using this master's generated headers and
UT build configuration. The standalone scanner check suppresses two
pre-existing unused-constant warnings from master.
- Carried over the source PR's BE unit tests, JSON pipe tests, and
Stream Load regression fixtures.
- `run-be-ut.sh` completed configuration and started the full build. The
fresh build requires about 14,750 targets, so it was stopped and
replaced with the focused syntax checks above. No master runtime
unit-test or cluster-regression pass is claimed; full build and runtime
validation remain for CI.
### Release note
Route supported traditional file loads through FileScannerV2 with a
table reader for load conversion and validation.
---
be/src/exec/operator/file_scan_operator.cpp | 30 +
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 | 376 ++++++++
be/src/format_v2/table/load_reader.h | 69 ++
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, 1985 insertions(+), 35 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index de0cc940ad5..e439f308ef4 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -155,6 +155,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;
+ }
+ }
// ADBC only has a FileScannerV2 reader, and enable_file_scanner_v2 is a
session variable marked
// fuzzy=true, so the regression harness flips it to false at random.
Without letting adbc
// through unconditionally, those queries land in v1, which has no "adbc"
branch, and come back
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index c8aaab44f7a..15a71f2c2d1 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -65,6 +65,7 @@
#include "format_v2/table/hudi_reader.h"
#include "format_v2/table/iceberg_position_delete_sys_table_reader.h"
#include "format_v2/table/iceberg_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"
@@ -414,6 +415,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;
@@ -466,7 +468,13 @@ Status FileScannerV2::_open_impl(RuntimeState* state) {
RETURN_IF_ERROR(Scanner::_open_impl(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));
@@ -927,7 +935,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.
+ // LoadTableReader 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();
}
@@ -1023,6 +1042,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;
@@ -1364,6 +1386,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 0c029838d67..d7d4ea48000 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -243,6 +243,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 bc4026138f1..d497b790a1b 100644
--- a/be/src/format_v2/delimited_text/csv_reader.cpp
+++ b/be/src/format_v2/delimited_text/csv_reader.cpp
@@ -174,6 +174,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 f3c068ec252..31164d8ba37 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 8576477d9f8..98258447a03 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -141,6 +141,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.
@@ -350,6 +353,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);
@@ -466,6 +475,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 7e10aec5e03..d384067171f 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..f24bbfb2f1f
--- /dev/null
+++ b/be/src/format_v2/table/load_reader.cpp
@@ -0,0 +1,376 @@
+// 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) {}
+
+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->type();
+ 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.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->type()->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->type()->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) {
+ 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]->type()->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..73c9b194451
--- /dev/null
+++ b/be/src/format_v2/table/load_reader.h
@@ -0,0 +1,69 @@
+// 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) 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 7ae0d5ca835..154ba5e6431 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -274,8 +274,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) {
@@ -345,12 +344,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());
@@ -812,9 +806,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) {
@@ -828,6 +825,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();
}
@@ -1062,11 +1063,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 11d252394c0..b38caa0a7bc 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -462,7 +462,7 @@ TEST(FileScannerV2Test,
ConditionCacheDigestIncludesRuntimeFilterPayload) {
0);
}
-TEST(FileScannerV2Test, FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
+TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesAndLoads) {
TQueryOptions query_options;
TFileScanRangeParams params;
params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
@@ -471,7 +471,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));
@@ -505,7 +505,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");
@@ -535,7 +535,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 f2f48ba6367..bbd8db29444 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..21d378e40f3
--- /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->type()));
+ 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->type()));
+ 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->type()));
+ 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 a3390ca0d58..397fa822157 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
@@ -659,7 +659,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}")
}
}
@@ -695,7 +696,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]
