[opt](assert_num_rows) support filter in AssertNumRows operator and f…

…ix some explain (apache#28935)

* NEED

* Update pipeline x

* fix pipelinex compile

be/src/pipeline/exec/assert_num_rows_operator.cpp

@@ -17,6 +17,8 @@
 
 #include "assert_num_rows_operator.h"
 
+#include "vec/exprs/vexpr_context.h"
+
 namespace doris::pipeline {
 
 OperatorPtr AssertNumRowsOperatorBuilder::build_operator() {
@@ -83,6 +85,7 @@ Status AssertNumRowsOperatorX::pull(doris::RuntimeState* state, vectorized::Bloc
     }
     COUNTER_SET(local_state.rows_returned_counter(), local_state.num_rows_returned());
     COUNTER_UPDATE(local_state.blocks_returned_counter(), 1);
+    RETURN_IF_ERROR(vectorized::VExprContext::filter_block(_conjuncts, block, block->columns()));
     return Status::OK();
 }
 

be/src/vec/exec/vassert_num_rows_node.cpp

@@ -29,6 +29,7 @@
 
 #include "runtime/runtime_state.h"
 #include "vec/core/block.h"
+#include "vec/exprs/vexpr_context.h"
 
 namespace doris {
 class DescriptorTbl;
@@ -100,6 +101,7 @@ Status VAssertNumRowsNode::pull(doris::RuntimeState* state, vectorized::Block* b
                                  to_string_lambda(_assertion), _desired_num_rows, _subquery_string);
     }
     COUNTER_SET(_rows_returned_counter, _num_rows_returned);
+    RETURN_IF_ERROR(VExprContext::filter_block(_conjuncts, block, block->columns()));
     return Status::OK();
 }
 

fe/fe-core/src/main/java/org/apache/doris/planner/AssertNumRowsNode.java

@@ -79,6 +79,11 @@ public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
         StringBuilder output = new StringBuilder()
                 .append(prefix).append("assert number of rows: ")
                 .append(assertion).append(" ").append(desiredNumOfRows).append("\n");
+
+        if (!conjuncts.isEmpty()) {
+            output.append(prefix).append("predicates: ").append(getExplainString(conjuncts)).append("\n");
+        }
+
         return output.toString();
     }
 

fe/fe-core/src/main/java/org/apache/doris/planner/DataGenScanNode.java

@@ -26,6 +26,7 @@
 import org.apache.doris.tablefunction.DataGenTableValuedFunction;
 import org.apache.doris.tablefunction.TableValuedFunctionTask;
 import org.apache.doris.thrift.TDataGenScanNode;
+import org.apache.doris.thrift.TExplainLevel;
 import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TPlanNode;
 import org.apache.doris.thrift.TPlanNodeType;
@@ -117,4 +118,23 @@ public boolean needToCheckColumnPriv() {
     public int getNumInstances() {
         return 1;
     }
+
+    @Override
+    public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
+        if (detailLevel == TExplainLevel.BRIEF) {
+            return "";
+        }
+
+        StringBuilder output = new StringBuilder();
+
+        if (!conjuncts.isEmpty()) {
+            output.append(prefix).append("predicates: ").append(getExplainString(conjuncts)).append("\n");
+        }
+
+        output.append(prefix).append("table value function: ").append(tvf.getDataGenFunctionName()).append("\n");
+
+
+
+        return output.toString();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/planner/PartitionSortNode.java

@@ -120,6 +120,10 @@ public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
         }
         output.append("\n");
 
+        if (!conjuncts.isEmpty()) {
+            output.append(prefix).append("predicates: ").append(getExplainString(conjuncts)).append("\n");
+        }
+
         // Add the limit information;
         output.append(prefix).append("has global limit: ").append(hasGlobalLimit).append("\n");
         output.append(prefix).append("partition limit: ").append(partitionLimit).append("\n");

regression-test/data/correctness_p0/test_assert_row_num.out

@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql_1 --
+
+-- !sql_2 --
+0
+1
+2
+3
+4
+5
+6
+7
+8
+9
+
+-- !sql_3 --
+

regression-test/suites/correctness_p0/test_assert_row_num.groovy

@@ -0,0 +1,91 @@
+// 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_assert_num_rows") {
+    qt_sql_1 """
+        SELECT * from numbers("number"="10") WHERE  ( SELECT * FROM (SELECT 3) __DORIS_DUAL__ ) IS  NULL
+    """
+
+    qt_sql_2 """
+        SELECT * from numbers("number"="10") WHERE  ( SELECT * FROM (SELECT 3) __DORIS_DUAL__ ) IS NOT NULL
+    """
+    sql """
+        DROP TABLE IF EXISTS table_9_undef_undef;
+    """
+    sql """
+        DROP TABLE IF EXISTS table_10_undef_undef;
+    """
+    sql """
+    CREATE TABLE table_10_undef_undef (
+        `pk` int, `col_int_undef_signed` int ,
+        `col_varchar_10__undef_signed` varchar(10),
+        `col_varchar_1024__undef_signed` varchar(1024))
+    ENGINE=olap distributed BY hash(pk) buckets 10 properties('replication_num'='1');
+    """
+
+    sql """
+    CREATE TABLE table_9_undef_undef (
+        `pk` int,`col_int_undef_signed` int ,
+        `col_varchar_10__undef_signed` varchar(10) ,
+        `col_varchar_1024__undef_signed` varchar(1024))
+    ENGINE=olap distributed BY hash(pk) buckets 10 properties('replication_num' = '1');
+    """
+
+    sql """
+        INSERT INTO table_9_undef_undef
+        VALUES (0,NULL,"get",'r'),
+            (1,NULL,'q','i'),
+            (2,2,"about","yes"),
+            (3,NULL,"see","see"),
+            (4,NULL,"was","been"),
+            (5,NULL,"yes",'p'),
+            (6,6,"you",'u'),
+            (7,0,"me",'v'),
+            (8,5,"something",'f');
+    """
+    sql """
+        INSERT INTO table_10_undef_undef
+        VALUES (0,NULL,"it's","time"),
+            (1,NULL,"right",'o'),
+            (2,5,'y','k'),
+            (3,1,'r',"I'll"),
+            (4,2,'e',"time"),
+            (5,8,'v',"from"),
+            (6,NULL,"you",'v'),
+            (7,NULL,'r','a'),
+            (8,1,'d',"didn't"),
+            (9,NULL,'r',"go");
+    """
+
+    qt_sql_3 """
+        SELECT alias1 . `pk` AS field1,
+            alias1 . `col_int_undef_signed` AS field2
+        FROM table_10_undef_undef AS alias1,
+            table_9_undef_undef AS alias2
+        WHERE
+            (SELECT *
+            FROM
+            (SELECT 3) __DORIS_DUAL__) IS NULL
+        HAVING field2 < 2
+        ORDER BY alias1 . `pk`,
+                alias1 .`pk` ASC,
+                field1,
+                field2
+        LIMIT 2
+        OFFSET 6;
+    """
+}