This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 00a946dc431 [branch-4.1][fix](load) only load source scanners update
load counters (#65276)
00a946dc431 is described below
commit 00a946dc431cf01442850612393ee46144f40eb5
Author: Xin Liao <[email protected]>
AuthorDate: Tue Jul 7 13:41:33 2026 +0800
[branch-4.1][fix](load) only load source scanners update load counters
(#65276)
Pick apache/doris#63781
---
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 | 14 ++++
...n_filtered_rows_not_pollute_load_counter.groovy | 94 ++++++++++++++++++++++
6 files changed, 135 insertions(+), 3 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index fdc53fc8231..e6cd425bd58 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -2009,6 +2009,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 8a32d154695..833ca9419bb 100644
--- a/be/src/exec/scan/file_scanner.h
+++ b/be/src/exec/scan/file_scanner.h
@@ -112,6 +112,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 65dcb46c3b1..c8c959942d0 100644
--- a/be/src/exec/scan/scanner.cpp
+++ b/be/src/exec/scan/scanner.cpp
@@ -257,9 +257,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 241dd99107a..48f6f7ca3ab 100644
--- a/be/src/exec/scan/scanner.h
+++ b/be/src/exec/scan/scanner.h
@@ -117,6 +117,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; }
+
// Filter the output block finally.
Status _filter_output_block(Block* block);
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..8a943e71beb
--- /dev/null
+++
b/regression-test/data/load_p0/insert/test_scan_filtered_rows_not_pollute_load_counter.out
@@ -0,0 +1,14 @@
+-- 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]
