This is an automated email from the ASF dual-hosted git repository.
liaoxin01 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 a786cba9824 [fix](load) only load source scanners update load counters
(#63781)
a786cba9824 is described below
commit a786cba9824bebe5decd4d70a7ae7dd905609a38
Author: Xin Liao <[email protected]>
AuthorDate: Thu Jun 25 16:02:26 2026 +0800
[fix](load) only load source scanners update load counters (#63781)
### What
- Gate the load counter update in
`Scanner::_collect_profile_before_close()` on a new virtual
`_should_update_load_counters()`:
- Base `Scanner` reports only when `_is_load` (classic
stream/broker/routine load scanners with src tuple desc).
- `FileScanner` additionally reports for `FILE_STREAM` scans: TVF based
loads (`http_stream`, group commit) plan the load source as a tvf query
scan without src tuple desc (`_is_load` is false), but their WHERE
clause filtered rows must still be reported as `NumberUnselectedRows` /
counted into `NumberTotalRows`.
- Add a deterministic regression case covering INSERT-SELECT / DELETE /
UPDATE whose scans filter out all rows.
### Why
For DELETE/UPDATE/INSERT INTO ... SELECT executed through the insert
path, rows filtered by query scan predicates (including runtime filters)
were added to the RuntimeState load counters. When all scanned rows are
filtered, `num_rows_load_success()` (total - filtered - unselected) goes
negative, BE reports a negative `dpp.norm.ALL`, and FE fails the
`insert_max_filter_ratio` check with errors like:
```
Insert has too many filtered data 0/-2 insert_max_filter_ratio is 1.000000
```
This only triggers with `enable_insert_strict=false` and
`insert_max_filter_ratio > 0` (the strict branch only checks
`filteredRows > 0`). The intermittency in the field comes from runtime
filter arrival timing: rows are only counted when the RF arrives in time
to filter inside the scanner.
This was historically gated by `if (!enable_profile && !_is_load)
return;` (already buggy with `enable_profile=true`), and became
unconditional after #57314 removed the early return.
Compared to the previous revision of this PR (thrift
`skip_query_scan_load_counters` option set by FE for DELETE/UPDATE):
- No thrift / FE changes needed.
- Also fixes plain INSERT INTO ... SELECT: `AbstractInsertExecutor` sets
`query_type=LOAD` for all insert-path commands, so the query-type based
gate still let OlapScanner predicate-filtered rows pollute the counters
there.
- `FILE_STREAM` is a precise discriminator: only the `http_stream` /
`group_commit` TVFs use it, and they require a backend id on the
ConnectContext (only present for stream-load style HTTP requests), so
they can never appear in a normal query/DELETE/UPDATE.
`FileScanner::_counter.num_rows_filtered` is only accumulated in
`_convert_to_output_block` (load-only path), so for `http_stream` the
`NumberFilteredRows` reported to clients comes from sink validation and
is unaffected by this gate.
### Test
- New regression case
`regression-test/suites/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.groovy`:
uses value-column predicates on an AGGREGATE KEY table (cannot be pushed
down to storage, evaluated by scanner conjuncts) to deterministically
reproduce; before this fix the INSERT-SELECT/DELETE/UPDATE statements
fail with `0/-10`.
- `test_group_commit_http_stream` semantics preserved: `insert into ...
select from http_stream(...) where ...` still reports
`NumberUnselectedRows`/`NumberTotalRows` (the previous `_is_load`-only
attempt broke this with `expected: <6> but was: <5>`).
---
be/src/exec/scan/file_scanner.cpp | 13 +++
be/src/exec/scan/file_scanner.h | 2 +
be/src/exec/scan/scanner.cpp | 8 +-
be/src/exec/scan/scanner.h | 7 ++
...scan_filtered_rows_not_pollute_load_counter.out | 15 ++++
...n_filtered_rows_not_pollute_load_counter.groovy | 94 ++++++++++++++++++++++
6 files changed, 136 insertions(+), 3 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index d5c18a38b85..6419ce4f65c 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -2079,6 +2079,19 @@ void FileScanner::update_realtime_counters() {
_last_bytes_read_from_remote =
_file_cache_statistics->bytes_read_from_remote;
}
+bool FileScanner::_should_update_load_counters() const {
+ if (_is_load) {
+ return true;
+ }
+ // TVF based loads (e.g. http_stream, group commit relay) plan the load
source as a
+ // tvf query scan without src tuple desc, so _is_load is false. But rows
filtered by
+ // the load's WHERE clause still need to be reported as unselected rows.
FILE_STREAM
+ // is only reachable from such load entries, never from normal queries, so
use it to
+ // identify these scanners.
+ return (_params->__isset.file_type && _params->file_type ==
TFileType::FILE_STREAM) ||
+ (_current_range.__isset.file_type && _current_range.file_type ==
TFileType::FILE_STREAM);
+}
+
void FileScanner::_collect_profile_before_close() {
Scanner::_collect_profile_before_close();
if (config::enable_file_cache && _state->query_options().enable_file_cache
&&
diff --git a/be/src/exec/scan/file_scanner.h b/be/src/exec/scan/file_scanner.h
index 2113cdaf475..fbcbca464a5 100644
--- a/be/src/exec/scan/file_scanner.h
+++ b/be/src/exec/scan/file_scanner.h
@@ -119,6 +119,8 @@ protected:
void _collect_profile_before_close() override;
+ bool _should_update_load_counters() const override;
+
// fe will add skip_bitmap_col to _input_tuple_desc iff the target
olaptable has skip_bitmap_col
// and the current load is a flexible partial update
bool _should_process_skip_bitmap_col() const { return _skip_bitmap_col_idx
!= -1; }
diff --git a/be/src/exec/scan/scanner.cpp b/be/src/exec/scan/scanner.cpp
index 7d72d5838d1..8788331c245 100644
--- a/be/src/exec/scan/scanner.cpp
+++ b/be/src/exec/scan/scanner.cpp
@@ -286,9 +286,11 @@ void Scanner::_collect_profile_before_close() {
COUNTER_UPDATE(_local_state->_scan_cpu_timer, _scan_cpu_timer);
COUNTER_UPDATE(_local_state->_rows_read_counter, _num_rows_read);
- // Update stats for load
- _state->update_num_rows_load_filtered(_counter.num_rows_filtered);
- _state->update_num_rows_load_unselected(_counter.num_rows_unselected);
+ // Update stats for load. See _should_update_load_counters() for why this
is gated.
+ if (_should_update_load_counters()) {
+ _state->update_num_rows_load_filtered(_counter.num_rows_filtered);
+ _state->update_num_rows_load_unselected(_counter.num_rows_unselected);
+ }
}
void Scanner::_update_scan_cpu_timer() {
diff --git a/be/src/exec/scan/scanner.h b/be/src/exec/scan/scanner.h
index 38d31cd26c5..a91395b911e 100644
--- a/be/src/exec/scan/scanner.h
+++ b/be/src/exec/scan/scanner.h
@@ -125,6 +125,13 @@ protected:
// Update the counters before closing this scanner
virtual void _collect_profile_before_close();
+ // Whether rows filtered/unselected by this scanner should be reported to
the load
+ // counters in RuntimeState. Only the scanner reading the load source data
should
+ // report, otherwise rows filtered by query predicates (e.g. in INSERT
INTO ... SELECT
+ // or DELETE FROM ... WHERE) would be mixed into load counters and make
+ // num_rows_load_success() negative.
+ virtual bool _should_update_load_counters() const { return _is_load; }
+
// Check if scanner is already closed, if not, mark it as closed.
// Returns true if the scanner was successfully marked as closed (first
time).
// Returns false if the scanner was already closed.
diff --git
a/regression-test/data/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.out
b/regression-test/data/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.out
new file mode 100644
index 00000000000..553b5529180
--- /dev/null
+++
b/regression-test/data/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.out
@@ -0,0 +1,15 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !insert_empty --
+0
+
+-- !insert_empty_profile --
+0
+
+-- !delete_noop --
+3
+
+-- !update_noop --
+1 1
+2 2
+3 3
+
diff --git
a/regression-test/suites/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.groovy
b/regression-test/suites/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.groovy
new file mode 100644
index 00000000000..e0f4e587cef
--- /dev/null
+++
b/regression-test/suites/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.groovy
@@ -0,0 +1,94 @@
+// 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_scan_filtered_rows_not_pollute_load_counter", "p0") {
+ // Rows filtered by scan predicates of a query must not be counted as load
+ // "unselected" rows. Otherwise loadedRows reported to FE becomes negative
+ // (total 0 - unselected N) and the insert fails with errors like
+ // "Insert has too many filtered data 0/-10 insert_max_filter_ratio is
1.000000".
+ def srcTable = "test_scan_filter_load_counter_src"
+ def dstTable = "test_scan_filter_load_counter_dst"
+ def uniqTable = "test_scan_filter_load_counter_uniq"
+
+ sql """ DROP TABLE IF EXISTS ${srcTable} """
+ // Predicates on value columns of an AGGREGATE KEY table can neither be
pushed
+ // down as column predicates nor as common expressions, so they are
evaluated
+ // by the scanner conjuncts and counted into
ScannerCounter.num_rows_unselected.
+ sql """
+ CREATE TABLE ${srcTable} (
+ k1 INT,
+ v1 INT REPLACE
+ ) AGGREGATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """
+ INSERT INTO ${srcTable} VALUES
+ (1,1),(2,2),(3,3),(4,4),(5,5),(6,6),(7,7),(8,8),(9,9),(10,10)
+ """
+
+ sql """ DROP TABLE IF EXISTS ${dstTable} """
+ sql """
+ CREATE TABLE ${dstTable} (
+ k1 INT
+ ) DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES ("replication_num" = "1");
+ """
+
+ sql "set enable_insert_strict=false"
+ sql "set insert_max_filter_ratio=1"
+
+ // All 10 scanned rows are filtered inside the scanner; the insert must
+ // succeed as a no-op instead of failing the filter ratio check.
+ sql """ INSERT INTO ${dstTable} SELECT k1 FROM ${srcTable} WHERE v1 > 1000
"""
+ qt_insert_empty "select count(*) from ${dstTable}"
+
+ // Same with profile enabled, which was the original trigger of this issue.
+ sql "set enable_profile=true"
+ sql """ INSERT INTO ${dstTable} SELECT k1 FROM ${srcTable} WHERE v1 > 1000
"""
+ qt_insert_empty_profile "select count(*) from ${dstTable}"
+ sql "set enable_profile=false"
+
+ // DELETE ... WHERE EXISTS executes through the insert path (delete sign).
+ // A no-op delete whose source scan filters out all rows must succeed.
+ sql """ DROP TABLE IF EXISTS ${uniqTable} """
+ sql """
+ CREATE TABLE ${uniqTable} (
+ k1 INT,
+ v1 INT
+ ) UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES ("replication_num" = "1");
+ """
+ sql """ INSERT INTO ${uniqTable} VALUES (1,1),(2,2),(3,3) """
+ sql """
+ DELETE FROM ${uniqTable} t WHERE EXISTS (
+ SELECT 1 FROM ${srcTable} s WHERE s.k1 = t.k1 AND s.v1 > 1000
+ )
+ """
+ qt_delete_noop "select count(*) from ${uniqTable}"
+
+ // UPDATE also executes through the insert path; a no-op update whose
+ // subquery scan filters out all rows must succeed.
+ sql """
+ UPDATE ${uniqTable} SET v1 = 100 WHERE k1 IN (
+ SELECT k1 FROM ${srcTable} WHERE v1 > 1000
+ )
+ """
+ qt_update_noop "select * from ${uniqTable} order by k1"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
