Repository navigation
[refactor](pipeline) Delete non-pipeline exec nodes - #35803
Conversation
|
Thank you for your contribution to Apache Doris. Since 2024-03-18, the Document has been moved to doris-website. |
|
run buildall |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| template <typename ParentType> | ||
| Status ParallelScannerBuilder<ParentType>::_build_scanners_by_rowid( | ||
| std::list<VScannerSPtr>& scanners) { | ||
| Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<VScannerSPtr>& scanners) { |
There was a problem hiding this comment.
warning: function '_build_scanners_by_rowid' exceeds recommended size/complexity thresholds [readability-function-size]
Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<VScannerSPtr>& scanners) {
^Additional context
be/src/olap/parallel_scanner_builder.cpp:40: 126 lines including whitespace and comments (threshold 80)
Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<VScannerSPtr>& scanners) {
^| struct BlockRowPos { | ||
| BlockRowPos() : block_num(0), row_num(0), pos(0) {} | ||
| int64_t block_num; //the pos at which block | ||
| int64_t row_num; //the pos at which row |
There was a problem hiding this comment.
warning: use default member initializer for 'row_num' [modernize-use-default-member-init]
be/src/pipeline/dependency.h:508:
- BlockRowPos() : block_num(0), row_num(0), pos(0) {}
+ BlockRowPos() : block_num(0), , pos(0) {}| int64_t row_num; //the pos at which row | |
| int64_t row_num{0}; //the pos at which row |
| BlockRowPos() : block_num(0), row_num(0), pos(0) {} | ||
| int64_t block_num; //the pos at which block | ||
| int64_t row_num; //the pos at which row | ||
| int64_t pos; //pos = all blocks size + row_num |
There was a problem hiding this comment.
warning: use default member initializer for 'pos' [modernize-use-default-member-init]
be/src/pipeline/dependency.h:508:
- BlockRowPos() : block_num(0), row_num(0), pos(0) {}
+ BlockRowPos() : block_num(0), row_num(0), {}| int64_t pos; //pos = all blocks size + row_num | |
| int64_t pos{0}; //pos = all blocks size + row_num |
| Status ScanLocalState<Derived>::_normalize_in_and_eq_predicate( | ||
| vectorized::VExpr* expr, vectorized::VExprContext* expr_ctx, SlotDescriptor* slot, | ||
| ColumnValueRange<T>& range, vectorized::VScanNode::PushDownType* pdt) { | ||
| Status ScanLocalState<Derived>::_normalize_in_and_eq_predicate(vectorized::VExpr* expr, |
There was a problem hiding this comment.
warning: function '_normalize_in_and_eq_predicate' has cognitive complexity of 63 (threshold 50) [readability-function-cognitive-complexity]
Status ScanLocalState<Derived>::_normalize_in_and_eq_predicate(vectorized::VExpr* expr,
^Additional context
be/src/pipeline/exec/scan_operator.cpp:593: +1, including nesting penalty of 0, nesting level increased to 1
if (TExprNodeType::IN_PRED == expr->node_type()) {
^be/src/pipeline/exec/scan_operator.cpp:597: +2, including nesting penalty of 1, nesting level increased to 2
if (hybrid_set != nullptr) {
^be/src/pipeline/exec/scan_operator.cpp:599: +3, including nesting penalty of 2, nesting level increased to 3
if (hybrid_set->size() <=
^be/src/pipeline/exec/scan_operator.cpp:602: +1, nesting level increased to 3
} else {
^be/src/pipeline/exec/scan_operator.cpp:607: +1, nesting level increased to 2
} else {
^be/src/pipeline/exec/scan_operator.cpp:611: +3, including nesting penalty of 2, nesting level increased to 3
if (temp_pdt == PushDownType::UNACCEPTABLE) {
^be/src/pipeline/exec/scan_operator.cpp:621: +3, including nesting penalty of 2, nesting level increased to 3
if (!state->use_set) {
^be/src/pipeline/exec/scan_operator.cpp:628: +2, including nesting penalty of 1, nesting level increased to 2
while (iter->has_next()) {
^be/src/pipeline/exec/scan_operator.cpp:631: +3, including nesting penalty of 2, nesting level increased to 3
if (nullptr == iter->get_value()) {
^be/src/pipeline/exec/scan_operator.cpp:636: +3, including nesting penalty of 2, nesting level increased to 3
RETURN_IF_ERROR(_change_value_range<true>(
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/scan_operator.cpp:636: +4, including nesting penalty of 3, nesting level increased to 4
RETURN_IF_ERROR(_change_value_range<true>(
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/pipeline/exec/scan_operator.cpp:642: +1, nesting level increased to 1
} else if (TExprNodeType::BINARY_PRED == expr->node_type()) {
^be/src/pipeline/exec/scan_operator.cpp:644: nesting level increased to 2
auto eq_checker = [](const std::string& fn_name) { return fn_name == "eq"; };
^be/src/pipeline/exec/scan_operator.cpp:650: +2, including nesting penalty of 1, nesting level increased to 2
RETURN_IF_ERROR(_should_push_down_binary_predicate(
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/scan_operator.cpp:650: +3, including nesting penalty of 2, nesting level increased to 3
RETURN_IF_ERROR(_should_push_down_binary_predicate(
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/pipeline/exec/scan_operator.cpp:653: +2, including nesting penalty of 1, nesting level increased to 2
if (temp_pdt == PushDownType::UNACCEPTABLE) {
^be/src/pipeline/exec/scan_operator.cpp:659: +2, including nesting penalty of 1, nesting level increased to 2
if (value.data != nullptr) {
^be/src/pipeline/exec/scan_operator.cpp:660: +3, including nesting penalty of 2, nesting level increased to 3
if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
^be/src/pipeline/exec/scan_operator.cpp:660: +1
if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
^be/src/pipeline/exec/scan_operator.cpp:663: +4, including nesting penalty of 3, nesting level increased to 4
RETURN_IF_ERROR(_change_value_range<true>(
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/scan_operator.cpp:663: +5, including nesting penalty of 4, nesting level increased to 5
RETURN_IF_ERROR(_change_value_range<true>(
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/pipeline/exec/scan_operator.cpp:666: +1, nesting level increased to 3
} else {
^be/src/pipeline/exec/scan_operator.cpp:667: +4, including nesting penalty of 3, nesting level increased to 4
if (sizeof(typename PrimitiveTypeTraits<T>::CppType) != value.size) {
^be/src/pipeline/exec/scan_operator.cpp:672: +4, including nesting penalty of 3, nesting level increased to 4
RETURN_IF_ERROR(_change_value_range<true>(
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/pipeline/exec/scan_operator.cpp:672: +5, including nesting penalty of 4, nesting level increased to 5
RETURN_IF_ERROR(_change_value_range<true>(
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^| template <bool with_other_conjuncts> | ||
| Status ProcessHashTableProbe<JoinOpType, Parent>::do_mark_join_conjuncts( | ||
| Block* output_block, size_t hash_table_bucket_size) { | ||
| Status ProcessHashTableProbe<JoinOpType>::do_mark_join_conjuncts(Block* output_block, |
There was a problem hiding this comment.
warning: function 'do_mark_join_conjuncts' has cognitive complexity of 56 (threshold 50) [readability-function-cognitive-complexity]
_conjuncts>
^Additional context
be/src/vec/exec/join/process_hash_table_probe_impl.h:376: +1, including nesting penalty of 0, nesting level increased to 1
ck->rows();
^be/src/vec/exec/join/process_hash_table_probe_impl.h:385: +1, including nesting penalty of 0, nesting level increased to 1
et_data();
^be/src/vec/exec/join/process_hash_table_probe_impl.h:397: +2, including nesting penalty of 1, nesting level increased to 2
tched = -1;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:399: +3, including nesting penalty of 2, nesting level increased to 3
ucket_size;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:400: +4, including nesting penalty of 3, nesting level increased to 4
are_join) {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:403: +1, nesting level increased to 4
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:403: +5, including nesting penalty of 4, nesting level increased to 5
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:406: +1, nesting level increased to 5
i] = false;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:409: +1, nesting level increased to 5
_indexs[i];
^be/src/vec/exec/join/process_hash_table_probe_impl.h:415: +2, including nesting penalty of 1, nesting level increased to 2
}
^be/src/vec/exec/join/process_hash_table_probe_impl.h:419: +1, nesting level increased to 1
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:419: +2, including nesting penalty of 1, nesting level increased to 2
} else {
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/vec/exec/join/process_hash_table_probe_impl.h:419: +3, including nesting penalty of 2, nesting level increased to 3
} else {
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/vec/exec/join/process_hash_table_probe_impl.h:426: +1, including nesting penalty of 0, nesting level increased to 1
er.data();
^be/src/vec/exec/join/process_hash_table_probe_impl.h:430: +2, including nesting penalty of 1, nesting level increased to 2
ll = false;
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/vec/exec/join/process_hash_table_probe_impl.h:430: +3, including nesting penalty of 2, nesting level increased to 3
ll = false;
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/vec/exec/join/process_hash_table_probe_impl.h:436: +2, including nesting penalty of 1, nesting level increased to 2
ter.data();
^be/src/vec/exec/join/process_hash_table_probe_impl.h:455: +1
_has_null =
^be/src/vec/exec/join/process_hash_table_probe_impl.h:456: +1, including nesting penalty of 0, nesting level increased to 1
_conjuncts;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:458: +2, including nesting penalty of 1, nesting level increased to 2
_indexs[i];
^be/src/vec/exec/join/process_hash_table_probe_impl.h:460: +3, including nesting penalty of 2, nesting level increased to 3
_indexs[i];
^be/src/vec/exec/join/process_hash_table_probe_impl.h:462: +1
[i] = true;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:466: +1, nesting level increased to 2
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:466: +3, including nesting penalty of 2, nesting level increased to 3
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:469: +1, nesting level increased to 3
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:469: +4, including nesting penalty of 3, nesting level increased to 4
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:469: +1
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:477: +1, including nesting penalty of 0, nesting level increased to 1
}
^be/src/vec/exec/join/process_hash_table_probe_impl.h:479: +2, including nesting penalty of 1, nesting level increased to 2
mark column
^| template <bool with_other_conjuncts> | ||
| Status ProcessHashTableProbe<JoinOpType, Parent>::do_mark_join_conjuncts( | ||
| Block* output_block, size_t hash_table_bucket_size) { | ||
| Status ProcessHashTableProbe<JoinOpType>::do_mark_join_conjuncts(Block* output_block, |
There was a problem hiding this comment.
warning: function 'do_mark_join_conjuncts' exceeds recommended size/complexity thresholds [readability-function-size]
_conjuncts>
^Additional context
be/src/vec/exec/join/process_hash_table_probe_impl.h:363: 123 lines including whitespace and comments (threshold 80)
_conjuncts>
^| Status ProcessHashTableProbe<JoinOpType, Parent>::do_other_join_conjuncts( | ||
| Block* output_block, std::vector<uint8_t>& visited, bool has_null_in_build_side) { | ||
| template <int JoinOpType> | ||
| Status ProcessHashTableProbe<JoinOpType>::do_other_join_conjuncts(Block* output_block, |
There was a problem hiding this comment.
warning: function 'do_other_join_conjuncts' has cognitive complexity of 61 (threshold 50) [readability-function-cognitive-complexity]
JoinOpType>
^Additional context
be/src/vec/exec/join/process_hash_table_probe_impl.h:495: +1, including nesting penalty of 0, nesting level increased to 1
ck->rows();
^be/src/vec/exec/join/process_hash_table_probe_impl.h:504: +1, including nesting penalty of 0, nesting level increased to 1
ll = false;
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/vec/exec/join/process_hash_table_probe_impl.h:504: +2, including nesting penalty of 1, nesting level increased to 2
ll = false;
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/vec/exec/join/process_hash_table_probe_impl.h:519: +1, including nesting penalty of 0, nesting level increased to 1
.data();
^be/src/vec/exec/join/process_hash_table_probe_impl.h:526: +2, including nesting penalty of 1, nesting level increased to 2
re are any.
^be/src/vec/exec/join/process_hash_table_probe_impl.h:530: +3, including nesting penalty of 2, nesting level increased to 3
mn_ptr[i];
^be/src/vec/exec/join/process_hash_table_probe_impl.h:533: +1, nesting level increased to 3
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:535: +3, including nesting penalty of 2, nesting level increased to 3
}
^be/src/vec/exec/join/process_hash_table_probe_impl.h:540: +2, including nesting penalty of 1, nesting level increased to 2
^be/src/vec/exec/join/process_hash_table_probe_impl.h:541: +3, including nesting penalty of 2, nesting level increased to 3
unt; ++i) {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:544: +4, including nesting penalty of 3, nesting level increased to 4
mn_ptr[i]);
^be/src/vec/exec/join/process_hash_table_probe_impl.h:550: +1, nesting level increased to 1
er_column);
^be/src/vec/exec/join/process_hash_table_probe_impl.h:556: +2, including nesting penalty of 1, nesting level increased to 2
().data();
^be/src/vec/exec/join/process_hash_table_probe_impl.h:559: +3, including nesting penalty of 2, nesting level increased to 3
indexs[i];
^be/src/vec/exec/join/process_hash_table_probe_impl.h:560: +4, including nesting penalty of 3, nesting level increased to 4
EMI_JOIN) {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:562: +1, nesting level increased to 4
i] = false;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:566: +1, nesting level increased to 4
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:569: +1, nesting level increased to 3
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:569: +4, including nesting penalty of 3, nesting level increased to 4
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:572: +1, nesting level increased to 4
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:573: +5, including nesting penalty of 4, nesting level increased to 5
i] = false;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:581: +1, nesting level increased to 1
er_column);
^be/src/vec/exec/join/process_hash_table_probe_impl.h:583: +2, including nesting penalty of 1, nesting level increased to 2
NTI_JOIN) {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:586: +1, nesting level increased to 1
;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:588: +2, including nesting penalty of 1, nesting level increased to 2
r_size = 0;
^be/src/vec/exec/join/process_hash_table_probe_impl.h:595: +1, including nesting penalty of 0, nesting level increased to 1
0);
^be/src/vec/exec/join/process_hash_table_probe_impl.h:599: +1, nesting level increased to 1
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:599: +2, including nesting penalty of 1, nesting level increased to 2
} else {
^be/src/vec/exec/join/process_hash_table_probe_impl.h:604: +2, including nesting penalty of 1, nesting level increased to 2
;
^be/src/common/status.h:614: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/vec/exec/join/process_hash_table_probe_impl.h:604: +3, including nesting penalty of 2, nesting level increased to 3
;
^be/src/common/status.h:616: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^| Status ProcessHashTableProbe<JoinOpType, Parent>::do_other_join_conjuncts( | ||
| Block* output_block, std::vector<uint8_t>& visited, bool has_null_in_build_side) { | ||
| template <int JoinOpType> | ||
| Status ProcessHashTableProbe<JoinOpType>::do_other_join_conjuncts(Block* output_block, |
There was a problem hiding this comment.
warning: function 'do_other_join_conjuncts' exceeds recommended size/complexity thresholds [readability-function-size]
JoinOpType>
^Additional context
be/src/vec/exec/join/process_hash_table_probe_impl.h:490: 116 lines including whitespace and comments (threshold 80)
JoinOpType>
^| pipeline::OlapScanLocalState* local_state = (pipeline::OlapScanLocalState*)_local_state; | ||
| INCR_COUNTER(local_state); | ||
| } | ||
| pipeline::OlapScanLocalState* local_state = (pipeline::OlapScanLocalState*)_local_state; |
There was a problem hiding this comment.
warning: use auto when initializing with a cast to avoid duplicating the type name [modernize-use-auto]
| pipeline::OlapScanLocalState* local_state = (pipeline::OlapScanLocalState*)_local_state; | |
| auto* local_state = (pipeline::OlapScanLocalState*)_local_state; |
TPC-H: Total hot run time: 40236 ms |
TPC-DS: Total hot run time: 172575 ms |
ClickBench: Total hot run time: 30.73 s |
|
run buildall |
github-actions
Bot
added
the
approved
|
PR approved by at least one committer and no changes requested. |
|
PR approved by anyone and no changes requested. |
TPC-H: Total hot run time: 40396 ms |
TPC-DS: Total hot run time: 170587 ms |
ClickBench: Total hot run time: 31.02 s |
|
TeamCity be ut coverage result: |
### What problem does this PR solve? Issue Number: None Related PR: apache#69771 Problem Summary: The Spark and Flink Doris connectors read a table through FE's _query_plan API and the BE Thrift scanner (open_scanner / get_next), which drains the queue that MemoryScratchSink fills with Arrow batches. When the fragment failed, the sink's close() still put the nullptr sentinel and nothing recorded the failure: no code has called ResultQueueMgr::update_queue_status since the non-pipeline engine was removed (apache#35803), so RecordBatchQueue::status() stayed OK and get_next reported eos. A scan error, a cancellation or, with this PR, a batch the Arrow Flight convertor rejects (invalid UTF-8, a DATETIME before year 1) ended the read normally with the rows fetched so far: a table with one invalid UTF-8 row read zero rows and no error. MemoryScratchSinkLocalState::close now records a failed exec_status in the queue before the sentinel (update_status ignores OK), and fetch_result checks the queue status again when it takes the sentinel, because a fetch already waiting for data passed the first check before the failure. get_next then returns the error, for example "[INVALID_ARGUMENT]Invalid Arrow Flight result in column 2 (''): Invalid: Invalid UTF8 sequence at string index 1". ### Release note A failed read through the _query_plan scan API (Spark / Flink Doris connectors) now returns the error instead of ending as a complete result. ### Check List (For Author) - Test: Unit Test / Regression test - BE UT ResultQueueMgrTest: a fetch waiting for data reports a failure recorded before the sentinel; an OK status keeps a normal end. - Regression arrow_flight_sql_p0/test_query_plan_scan_validation reads _query_plan scans over the BE Thrift scanner: before this change the invalid UTF-8 scan read 0 rows and no error; with it, the UTF-8 and timestamp scans get the errors and a valid scan reads all rows. - Behavior changed: Yes. Such scans fail instead of returning partial data. - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…ht results (#69771) ### What problem does this PR solve? Issue Number: None Related PR: #68596, #68604 Problem Summary: #### In short This ports branch-4.x's #68596 and #68604 to master. An Arrow Flight result can carry values that Arrow clients cannot decode: a `STRING` value that is not valid UTF-8 (for example `unhex('84')`), and a `DATETIME` outside 0001-01-01..9999-12-31 (Doris's own range starts at `0000-01-01 00:00:00`). The Arrow writers accept both, so the BE publishes the batch and the failure surfaces later in the client: PyArrow rejects the string, and raises `OverflowError` when it converts the timestamp. This PR checks every batch in the Flight convertor before it is handed out and fails the query with a server-side error that names the column, the nested path and the row. It also makes a failed `_query_plan` scan of the Spark and Flink connectors report its error: the BE used to end such a scan as a complete result, so a rejected batch would have become silently missing rows. #### Background - `ArrowFlightArrowBlockConvertor` (`be/src/format/arrow/arrow_block_convertor.h`) builds the Arrow batches of an Arrow Flight result: `ArrowFlightBatchReader` uses it for Flight SQL and ADBC clients, which also covers a remote Doris catalog with `use_arrow_flight = true`, and the memory scratch sink uses it for the `_query_plan` scan API of the Spark and Flink Doris connectors. - The STRING serde appends the bytes of each value to an Arrow `utf8` (or `large_utf8`) builder without looking at them. Arrow requires these values to be valid UTF-8, but only `Array::ValidateFull()` checks it; building and publishing the array does not. - The memory scratch sink queues the batches that `get_next` of the BE Thrift scanner hands to those connectors, and ends the queue with a `nullptr` sentinel. Its `close()` put the sentinel whatever the execution status, and nothing has recorded a status in the queue since the non-pipeline engine was removed (#35803), so a fragment that failed (a scan error, a cancellation) ended the scan as if it were complete. - `DATETIME(p)` is published as `timestamp[s]`, `timestamp[ms]` or `timestamp[us]` by its scale. Arrow accepts any `int64` there, so a year-0 value such as `0000-12-31 00:00:00` (`-62135683200000000` us) is published as it is, while Python's `datetime`, which PyArrow converts to, only covers years 1..9999. #### The problem, and what it cost | Result value | Before this PR | |---|---| | `unhex('84')`, `unhex('C0AF')`, a surrogate or a code point above U+10FFFF, also inside an ARRAY / STRUCT / MAP key or value | Published as is; the client fails decoding the batch, after earlier batches were already delivered | | Two rows `unhex('C2')` and `unhex('A2')` | Published; each row is invalid UTF-8 on its own, though the concatenated buffer is valid | | `CAST('0000-12-31 00:00:00' AS DATETIME(6))`, or a zero date read from a table or an INT96 Parquet file | Published as `-62135683200000000`; PyArrow raises `OverflowError` converting it | | A `_query_plan` scan (Spark / Flink connector) whose fragment fails | `get_next` reports the end of the scan after the rows fetched so far; with the validation below, a table with one invalid UTF-8 row would read zero rows and no error | The query looks successful on the server, so neither the error nor the offending column shows up in Doris; the Flight client gets a failure that does not say which value caused it. #### How this PR fixes it - `ArrowBlockConvertor::convert_to_arrow` becomes virtual, and `ArrowFlightArrowBlockConvertor` overrides it: after the base conversion, it runs `ValidateFull()` on every column and returns `InvalidArgument("Invalid Arrow Flight result in column <n> ('<name>'): ...")` on failure. `ValidateFull()` checks each value, nested children included, so the row boundaries of the `C2` / `A2` case are respected. Binary columns keep their bytes; NULL values are skipped through the validity bitmap. - A new `FlightTimestampValidator` walks each column recursively (list values, map keys and values, struct fields), reads the actual timestamp unit, and checks each value against 0001-01-01..9999-12-31, also against the local calendar when the Arrow type has a time zone. Values under a NULL list / map / struct parent are not observable and are not checked. A value outside the range returns `InvalidArgument("Arrow Flight timestamp in column '<path>' at row <n> is outside the supported 0001-9999 range (type: ...)")`, with the row counted in the block. - For a timestamp with a time zone, the local bounds are computed with the exact offsets (cctz, Python's zoneinfo) and with the minute-rounded offsets of pytz, which PyArrow uses for zone names when it is installed (`(offset + 30) // 60 * 60`), and the stricter bound wins. pytz turns America/Chicago's local mean time -05:50:36 into -05:51, so 0001-01-01 05:50:36Z, local 00:00:00 for cctz, would be 0000-12-31 23:59:36 in PyArrow; the lower bound there is 05:51:00Z. In such zones up to 30 seconds at the start of year 1 are rejected. - `MemoryScratchSinkLocalState::close` records a failed execution status in the result queue before it puts the sentinel, and `ResultQueueMgr::fetch_result` checks the queue status again when it takes the sentinel, because a fetch already waiting for data passed the first check before the failure. `get_next` then returns the error to the connector. - Of the Arrow convertors, only the Flight convertor changes. The ordinary Doris Arrow convertor (Python UDFs) and the table-format convertors (Parquet, Iceberg, Paimon writers) keep their current behavior; `DorisArrowBlockConvertor` still publishes the same values in the new tests. - The code is branch-4.2's final form of the two PRs; master's Flight convertor already overrides `init()` and `write_column()`, and the new override sits next to them. #### Results | Query over Arrow Flight | Before | After | |---|---|---| | `SELECT unhex(encoded) AS payload` with an invalid UTF-8 row | Batch published, client fails decoding | Query fails: `Invalid Arrow Flight result in column 1 ('payload'): Invalid: Invalid UTF8 sequence at string index 0` | | `SELECT array(unhex(encoded)) AS nested` ... | Batch published | `... ('nested'): Invalid: List child array invalid: Invalid: Invalid UTF8 sequence at string index 0` | | `SELECT CAST(value AS DATETIME(6)) AS event_time` over `0000-12-31 00:00:00` | Batch published, client fails converting | Query fails: `Arrow Flight timestamp in column 'event_time' at row 1 is outside the supported 0001-9999 range (type: timestamp[us])`; nested values report paths such as `nested[]`, `nested.key`, `nested.child` | | Remote Doris catalog over Flight reading a `0000-01-01 00:00:00` DATETIME | Value returned | `Arrow Flight timestamp in column 'datetime_0' at row 1 is outside the supported 0001-9999 range (type: timestamp[s])` | | `_query_plan` scan through `open_scanner` / `get_next` over a table with an invalid UTF-8 row | 0 rows, then the end of the scan | `get_next` returns `[INVALID_ARGUMENT]Invalid Arrow Flight result in column 2 (''): Invalid: Invalid UTF8 sequence at string index 1` | | The same scan over a `0000-12-31 00:00:00` DATETIME | 0 rows, then the end of the scan | `Arrow Flight timestamp in column '' at row 2 is outside the supported 0001-9999 range (type: timestamp[us])` | | Valid Unicode text, NULLs, binary bytes, `0001-01-01`, `9999-12-31 23:59:59.999999` | Returned | Returned unchanged | Behavior change: a Flight query (Flight SQL, ADBC, a remote Doris catalog with `use_arrow_flight = true`, or a `_query_plan` read of the Spark / Flink connector) whose result contains a `STRING` value that is not valid UTF-8, or a `DATETIME` outside years 0001..9999, now fails with the error above instead of returning data. A remote Doris catalog that used to round-trip the zero date now reports this error; the three remote Doris suites that inserted `0000-01-01 00:00:00` as their minimum value now use `0001-01-01 00:00:00`, and `test_remote_doris_all_types_select` asserts the zero-date error separately. A `_query_plan` read whose fragment fails for any reason, also a scan error or a cancellation, now gets the error from `get_next` instead of ending as a complete result. Tests: - BE UT `ArrowBlockConvertorTest`: six new cases for invalid UTF-8 (four malformed encodings, per-row boundaries across a slice and the next batch, nested array / struct / map key / map value, large strings, bytes under a NULL), and valid text, NULLs and binary. - BE UT `ArrowFlightTimestampTest` (new file): zero date, year 0 and year 10000 at scales 0 / 3 / 6, calendar boundaries and pre-epoch fractions, slices and later batches, nested values and map keys, UTC and zoned bounds, the America/Chicago lower bound with pytz's rounded offset, naive bounds that do not depend on the session time zone, and values under NULL parents; the ordinary Doris convertor still publishes the same values. - BE UT `ResultQueueMgrTest`: a fetch already waiting for data reports a failure recorded before the sentinel; an OK status keeps a normal end. - Regression `arrow_flight_sql_p0/test_query_plan_scan_validation` reads `_query_plan` scans the way the connectors do (FE `_query_plan`, then BE `open_scanner` / `get_next` over Thrift): a valid scan reads all rows, and the invalid UTF-8 and year-0 scans get the errors above. - Regression `arrow_flight_sql_p0/test_flight_utf8_validation`, `test_flight_timestamp_range` (both red before this PR on master, green with it), `test_flight_int96_timestamp_range` (reads the INT96 fixtures of `group4` over HDFS; needs the Hive docker environment), and the three remote Doris suites.
Proposed changes
Issue Number: close #xxx