[branch-4.1][fix](load) only load source scanners update load counters

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 &&

be/src/exec/scan/file_scanner.h

@@ -112,6 +112,8 @@ class FileScanner : public Scanner {
 
     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; }

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() {

be/src/exec/scan/scanner.h

@@ -117,6 +117,13 @@ class Scanner {
     // 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);
 

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

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"
+}