github-actions[bot] commented on code in PR #69320:
URL: https://github.com/apache/doris/pull/69320#discussion_r4321638593
##########
be/src/format_v2/json/json_reader.cpp:
##########
@@ -702,7 +714,15 @@ Status JsonReader::_append_simple_json_rows(Block* block,
bool* is_empty_row, bo
_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) {
Review Comment:
[P2] Count empty outer arrays as rejected load records
With `strip_outer_array=true`, no `jsonpaths`, and JSON Lines containing
`[]` followed by `[{"k":1}]`, the empty-array early return at lines 710-712
bypasses load rejection accounting. V1 reports `Empty json line` and increments
the filtered-row count; V2 silently skips that document and loads the next one
with zero filtered rows. Since `StreamLoadExecutor` uses that count to enforce
`max_filter_ratio`, a load with `max_filter_ratio=0` now succeeds instead of
failing its data-quality check. Route empty arrays through the rejection
handler when `reject_invalid_rows` is enabled and cover both the count and
quality-gate result.
##########
be/src/format_v2/table/load_reader.cpp:
##########
@@ -0,0 +1,368 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/table/load_reader.h"
+
+#include "core/column/column_complex.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "exprs/vexpr.h"
+#include "exprs/vexpr_context.h"
+#include "format_v2/wal/wal_reader.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format {
+
+LoadTableReader::LoadTableReader(TupleDescriptor* source, const
TupleDescriptor* destination)
+ : _source_tuple(source), _destination_tuple(destination),
_source_row(source, false) {}
+
+LoadTableReader::~LoadTableReader() = default;
+
+Status LoadTableReader::init(TableReadOptions&& options) {
+ _scan_params = options.scan_params;
+ _runtime_state = options.runtime_state;
+ _format = options.format;
+ RETURN_IF_ERROR(_init_source_columns());
+ if (_scan_params->__isset.pre_filter_exprs_list) {
+ RETURN_IF_ERROR(
+ VExpr::create_expr_trees(_scan_params->pre_filter_exprs_list,
_pre_filters));
+ } else if (_scan_params->__isset.pre_filter_exprs) {
+ VExprContextSPtr expr;
+
RETURN_IF_ERROR(VExpr::create_expr_tree(_scan_params->pre_filter_exprs, expr));
+ _pre_filters.push_back(std::move(expr));
+ }
+ RETURN_IF_ERROR(VExpr::prepare(_pre_filters, _runtime_state, _source_row));
+ RETURN_IF_ERROR(VExpr::open(_pre_filters, _runtime_state));
+ for (auto* slot : _destination_tuple->slots()) {
+ const auto it = _scan_params->expr_of_dest_slot.find(slot->id());
+ if (it == _scan_params->expr_of_dest_slot.end() ||
it->second.nodes.empty()) {
+ return Status::InvalidArgument("Missing load expression for
destination slot {}",
+ slot->id());
+ }
+ VExprContextSPtr expr;
+ RETURN_IF_ERROR(VExpr::create_expr_tree(it->second, expr));
+ RETURN_IF_ERROR(expr->prepare(_runtime_state, _source_row));
+ RETURN_IF_ERROR(expr->open(_runtime_state));
+ _destination_exprs.push_back(std::move(expr));
+ int source_position = -1;
+ const auto direct =
_scan_params->dest_sid_to_src_sid_without_trans.find(slot->id());
+ if (direct != _scan_params->dest_sid_to_src_sid_without_trans.end()) {
+ for (size_t i = 0; i < _source_tuple->slots().size(); ++i) {
+ if (_source_tuple->slots()[i]->id() == direct->second) {
+ source_position = cast_set<int>(i);
+ break;
+ }
+ }
+ if (source_position < 0) {
+ return Status::InvalidArgument("Unknown load source slot {}",
direct->second);
+ }
+ }
+ _direct_source_positions.push_back(source_position);
+ if (slot->is_sequence_col()) {
+ _sequence_uid = slot->col_unique_id();
+ }
+ }
+ // Destination predicates must run after conversions and error accounting.
Passing them to
+ // the source reader would both bind the wrong slots and change
rejected-row statistics.
+ return TableReader::init({.projected_columns = std::move(_source_columns),
+ .conjuncts = {},
+ .format = options.format,
+ .scan_params = options.scan_params,
+ .io_ctx = options.io_ctx,
+ .runtime_state = options.runtime_state,
+ .scanner_profile = options.scanner_profile,
+ .file_slot_descs = &_source_file_slots,
+ .push_down_agg_type = TPushAggOp::NONE,
+ .condition_cache_digest = 0});
+}
+
+Status LoadTableReader::_init_source_columns() {
+ const auto& slots = _source_tuple->slots();
+ for (size_t i = 0; i < slots.size(); ++i) {
+ auto* slot = slots[i];
+ ColumnDefinition column;
+ column.identifier = Field::create_field<TYPE_STRING>(slot->col_name());
+ column.name = slot->col_name();
+ column.type = slot->get_data_type_ptr();
+ const auto required =
+ std::ranges::find_if(_scan_params->required_slots,
+ [&](const auto& info) { return
info.slot_id == slot->id(); });
+ if (required == _scan_params->required_slots.end()) {
+ return Status::InvalidArgument("Missing load source slot {} in
required slots",
+ slot->id());
+ }
+ if (required->is_file_slot) {
+ _source_file_slots.push_back(slot);
+ } else {
+ column.is_partition_key = true;
+ _partition_positions.push_back(cast_set<int>(i));
+ }
+ const auto default_it =
_scan_params->default_value_of_src_slot.find(slot->id());
+ if (default_it != _scan_params->default_value_of_src_slot.end() &&
+ !default_it->second.nodes.empty()) {
+ RETURN_IF_ERROR(VExpr::create_expr_tree(default_it->second,
column.default_expr));
+ VExprContextSPtr expr;
+ RETURN_IF_ERROR(VExpr::create_expr_tree(default_it->second, expr));
+ RETURN_IF_ERROR(expr->prepare(_runtime_state, _source_row));
+ RETURN_IF_ERROR(expr->open(_runtime_state));
+ _source_defaults.emplace(i, std::move(expr));
+ }
+ if (slot->is_skip_bitmap_col()) {
+ _skip_bitmap_position = cast_set<int>(i);
+ }
+ if (_scan_params->__isset.sequence_map_col &&
+ slot->col_name() == _scan_params->sequence_map_col) {
+ _sequence_map_uid = slot->col_unique_id();
+ }
+ ProjectedColumnBuildContext context {
+ .scan_params = _scan_params, .runtime_state = _runtime_state,
.slot_desc = slot};
+ if (_format == FileFormat::WAL) {
+ if (slot->col_unique_id() < 0) {
+ return Status::InternalError("WAL projected column {} has no
valid unique id",
+ column.name);
+ }
+ // WAL headers identify columns by stable unique id, including
after a rename.
+ column.identifier =
Field::create_field<TYPE_INT>(slot->col_unique_id());
+ } else {
+ RETURN_IF_ERROR(annotate_projected_column(*required, &context,
&column));
+ }
+ _source_columns.push_back(std::move(column));
+ }
+ return Status::OK();
+}
+
+Status LoadTableReader::prepare_split(const SplitReadOptions& options) {
+ auto source_options = options;
+ source_options.conjuncts = VExprContextSPtrs {};
+ source_options.partition_prune_conjuncts.clear();
+ source_options.condition_cache_digest = 0;
+ source_options.condition_cache_split_context.reset();
+ source_options.all_runtime_filters_applied = true;
+ const auto& range = options.current_range;
+ if (_partition_positions.size() > range.columns_from_path.size()) {
+ return Status::InvalidArgument("Load range has {} partition values for
{} source slots",
+ range.columns_from_path.size(),
_partition_positions.size());
+ }
+ for (size_t i = 0; i < _partition_positions.size(); ++i) {
+ const auto* slot = _source_tuple->slots()[_partition_positions[i]];
+ auto column = slot->get_data_type_ptr()->create_column();
+ const bool is_null = range.__isset.columns_from_path_is_null &&
+ i < range.columns_from_path_is_null.size() &&
+ range.columns_from_path_is_null[i];
+ if (is_null) {
+ column->insert_default();
+ } else {
+ Slice value(range.columns_from_path[i]);
+ DataTypeSerDe::FormatOptions serde_options;
+ RETURN_IF_ERROR(
+
slot->get_data_type_ptr()->get_serde()->deserialize_one_cell_from_hive_text(
+ *column, value, serde_options));
+ }
+ Field field;
+ column->get(0, field);
+ source_options.partition_values.insert_or_assign(slot->col_name(),
std::move(field));
+ }
+ return TableReader::prepare_split(source_options);
+}
+
+Status LoadTableReader::refresh_conjuncts(VExprContextSPtrs,
std::optional<uint64_t>, bool) {
+ return Status::OK();
+}
+
+Status LoadTableReader::build_physical_splits(const FileScanSplit&,
std::vector<FileScanSplit>*,
+ bool* was_split) {
+ // Keep ingestion records and their error accounting in the original FE
source range.
+ *was_split = false;
+ return Status::OK();
+}
+
+Status LoadTableReader::_reject_row(size_t row, const std::string& reason) {
+ RETURN_IF_ERROR(_runtime_state->append_error_msg_to_file(
+ [&] { return _source.dump_one_line(row,
_scan_params->num_of_columns_from_file); },
+ [&] { return reason; }));
+ ++_io_ctx->load_filtered_rows;
+ return Status::OK();
+}
+
+Status LoadTableReader::_materialize_missing_values() {
+ const size_t rows = _source.rows();
+ std::map<size_t, const std::vector<uint8_t>*> missing;
+ // Translate presence metadata only through the mapper, while the physical
reader is alive.
+ const auto& fields = _data_reader.reader->missing_fields();
+ for (const auto& mapping : _data_reader.column_mapper->mappings()) {
+ if (mapping.file_local_id.has_value()) {
+ const auto it = fields.find(LocalColumnId(*mapping.file_local_id));
+ if (it != fields.end()) {
+ missing.emplace(mapping.global_index.value(), &it->second);
+ }
+ }
+ }
+ if (_skip_bitmap_position >= 0) {
+ auto bitmap =
+
_source_tuple->slots()[_skip_bitmap_position]->get_data_type_ptr()->create_column();
+ bitmap->insert_many_defaults(rows);
+ auto& nullable = assert_cast<ColumnNullable&>(*bitmap);
+ std::fill(nullable.get_null_map_data().begin(),
nullable.get_null_map_data().end(), 0);
+ auto& values =
assert_cast<ColumnBitmap&>(nullable.get_nested_column()).get_data();
+ auto keep = ColumnUInt8::create(rows, 1);
+ for (const auto& [position, mask] : missing) {
+ const auto* slot = _source_tuple->slots()[position];
+ if (slot->is_skip_bitmap_col()) {
+ continue;
+ }
+ for (size_t row = 0; row < rows; ++row) {
+ if (!(*mask)[row]) {
+ continue;
+ }
+ if (slot->is_key() && !slot->is_auto_increment()) {
+ if (keep->get_data()[row]) {
+ RETURN_IF_ERROR(_reject_row(
+ row, fmt::format("Missing key column {} in
flexible partial update",
+ slot->col_name())));
+ keep->get_data()[row] = 0;
+ }
+ } else {
+ values[row].add(slot->col_unique_id());
+ if (_sequence_map_uid == slot->col_unique_id() &&
_sequence_uid >= 0) {
+ values[row].add(_sequence_uid);
+ }
+ }
+ }
+ }
+ _source.replace_by_position(_skip_bitmap_position, std::move(bitmap));
+ const auto size = _source.columns();
+ _source.insert({std::move(keep), std::make_shared<DataTypeUInt8>(),
"load_valid"});
+ RETURN_IF_ERROR(Block::filter_block(&_source, size, size));
+ } else {
+ for (const auto& [position, mask] : missing) {
+ const auto expr = _source_defaults.find(position);
+ if (expr == _source_defaults.end()) {
+ continue;
+ }
+ ColumnPtr default_values;
+ RETURN_IF_ERROR(expr->second->execute(&_source, default_values));
+ default_values = default_values->convert_to_full_column_if_const();
+ auto& input = _source.get_by_position(position);
+ default_values = input.type->is_nullable() ?
make_nullable(default_values)
+ :
remove_nullable(default_values);
+ auto result = input.type->create_column();
+ for (size_t row = 0; row < rows; ++row) {
+ result->insert_from((*mask)[row] ? *default_values :
*input.column, row);
+ }
+ _source.replace_by_position(position, std::move(result));
+ }
+ }
+ return Status::OK();
+}
+
+Status LoadTableReader::_convert(Block* output) {
+ const size_t rows = _source.rows();
+ auto keep = ColumnUInt8::create(rows, 1);
+ const std::vector<BitmapValue>* skip = nullptr;
+ if (_skip_bitmap_position >= 0) {
+ const auto& nullable = assert_cast<const ColumnNullable&>(
+ *_source.get_by_position(_skip_bitmap_position).column);
+ skip = &assert_cast<const
ColumnBitmap&>(nullable.get_nested_column()).get_data();
+ }
+ for (size_t i = 0; i < _destination_exprs.size(); ++i) {
+ const auto* slot = _destination_tuple->slots()[i];
+ ColumnPtr column;
+ RETURN_IF_ERROR(_destination_exprs[i]->execute(&_source, column));
+ column = column->convert_to_full_column_if_const();
+ for (size_t row = 0; row < rows; ++row) {
+ if (!keep->get_data()[row] || !column->is_null_at(row) ||
+ (skip != nullptr &&
(*skip)[row].contains(slot->col_unique_id()))) {
+ continue;
+ }
+ const int source_position = _direct_source_positions[i];
+ const bool invalid_cast =
+ _scan_params->strict_mode && source_position >= 0 &&
+
!_source.get_by_position(source_position).column->is_null_at(row);
+ if (invalid_cast || !slot->is_nullable()) {
+ RETURN_IF_ERROR(_reject_row(
+ row, fmt::format("Column {} {}", slot->col_name(),
+ invalid_cast ? "cannot be converted
in strict mode"
+ : "is NULL but is not
nullable")));
+ keep->get_data()[row] = 0;
+ }
+ }
+ column = slot->is_nullable() ? make_nullable(column) :
remove_nullable(column);
+ output->replace_by_position(i, std::move(column));
+ }
+ const auto size = output->columns();
+ output->insert({std::move(keep), std::make_shared<DataTypeUInt8>(),
"load_valid"});
+ RETURN_IF_ERROR(Block::filter_block(output, size, size));
+ return Status::OK();
+}
+
+Status LoadTableReader::finalize_chunk(Block* block, size_t physical_rows) {
+ _source = Block(_source_tuple->slots(), 0);
+ RETURN_IF_ERROR(TableReader::finalize_chunk(&_source, physical_rows));
Review Comment:
[P2] Reject absent file-source aliases without declared defaults
With target `t(id INT NOT NULL, v INT NULL)`, a Parquet/ORC file containing
only `id`, and `COLUMNS(id,raw_v,v=raw_v)`, this base finalization synthesizes
NULL for the absent `raw_v` instead of rejecting the source mapping. FE
supplies no `default_value_of_src_slot` entry for this alias because it is not
a table column; V1 rejects that case in
`FileScanner::_generate_missing_columns`. Here the nullable destination accepts
the synthesized NULL even with `strict_mode=true`, so the load succeeds with
NULL values. Validate missing physical source mappings against the original
default-map membership before generic materialization, preserving the
distinction between no entry and an explicit NULL default.
##########
be/test/exec/scan/file_scanner_v2_test.cpp:
##########
@@ -437,7 +437,7 @@ TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
query_options.__set_enable_file_scanner_v2(true);
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
-
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
Review Comment:
[P2] Update the remaining ORC-load selector expectation
The same selector change also affects
`FileScannerV2Test.IcebergOrcDefaultCannotHideDeferredParquetRanges`: its
unchanged `EXPECT_FALSE(TEST_should_use_file_scanner_v2(options, true,
params))` at line 505 sets `FORMAT_ORC`, for which the new load branch always
returns true. That test is included in `doris_be_test`, so the unit suite will
fail deterministically despite the updated assertions here. Update that
remaining load expectation while retaining its query-path checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
