diff --git a/be/src/exec/scan/file_scanner.cpp b/be/src/exec/scan/file_scanner.cpp index 5f1d248c1e1f4d..44b7b9d284e4cc 100644 --- a/be/src/exec/scan/file_scanner.cpp +++ b/be/src/exec/scan/file_scanner.cpp @@ -2065,6 +2065,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 cd4066ec987ad8..be1f68faf9f06e 100644 --- a/be/src/exec/scan/file_scanner.h +++ b/be/src/exec/scan/file_scanner.h @@ -119,6 +119,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; } diff --git a/be/src/exec/scan/scanner.cpp b/be/src/exec/scan/scanner.cpp index ae1ed96e000768..7015a28d8d3a23 100644 --- a/be/src/exec/scan/scanner.cpp +++ b/be/src/exec/scan/scanner.cpp @@ -288,9 +288,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 c14f6ee2048a7b..182b9b6a60cd1c 100644 --- a/be/src/exec/scan/scanner.h +++ b/be/src/exec/scan/scanner.h @@ -124,6 +124,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; } + // 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 00000000000000..553b5529180aba --- /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 00000000000000..e0f4e587cefb4a --- /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" +}