[fix](leftjoin) fix bug of left and full join with other conjuncts (#20946)

Fix bug of left and full outer join with other conjuncts. When equal matched row count of a probe row exceed batch_size, some times the _join_node->_is_any_probe_match_row_output flag is not set correcty, which result in outputing extra rows for the probe row.

be/src/vec/exec/join/process_hash_table_probe_impl.h

@@ -1036,7 +1036,8 @@ void ProcessHashTableProbe<JoinOpType>::_process_splited_equal_matched_tuples(
             *visited_map[i] |= other_hit;
         }
     }
-    _join_node->_is_any_probe_match_row_output |= simd::contain_byte(filter_map, row_count, 1);
+    _join_node->_is_any_probe_match_row_output |=
+            simd::contain_byte(filter_map + start_row_idx, row_count, 1);
 }
 
 template <int JoinOpType>

regression-test/data/query_p0/join/test_full_join_batch_size.out

@@ -0,0 +1,9 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql1 --
+\N	\N	1	211
+\N	\N	1	311
+\N	\N	1	411
+1	11	1	\N
+1	111	1	\N
+1	1111	1	\N
+

regression-test/data/query_p0/join/test_left_join_batch_size.out

@@ -0,0 +1,6 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql1 --
+1	11	1	\N
+1	111	1	\N
+1	1111	1	\N
+

regression-test/suites/query_p0/join/test_full_join_batch_size.groovy

@@ -0,0 +1,54 @@
+// 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_full_join_batch_size", "query,p0") {
+    sql " drop table if exists test_left_join_batch_size_l; ";
+    sql " drop table if exists test_left_join_batch_size_r; ";
+    sql """
+        create table test_left_join_batch_size_l (
+            k1 int,
+            v1 int
+        ) distributed by hash(k1) buckets 3
+        properties("replication_num" = "1");
+    """
+    sql """
+        create table test_left_join_batch_size_r (
+            k1 int,
+            v1 int
+        ) distributed by hash(k1) buckets 3
+        properties("replication_num" = "1");
+    """
+
+    sql """ insert into test_left_join_batch_size_l values (1, 11), (1, 111), (1, 1111) """
+    sql """ insert into test_left_join_batch_size_r values (1, null), (1, 211), (1, 311), (1, 411) """
+
+    qt_sql1 """
+        select /*+SET_VAR(batch_size=3)*/
+               l.k1,
+               l.v1,
+               r.k1,
+               r.v1
+        from
+               test_left_join_batch_size_l l
+               full join test_left_join_batch_size_r r on (
+                      r.v1 = 0
+                      or r.v1 is null
+               )
+               and l.k1 = r.k1
+                order by l.k1, l.v1, r.k1, r.v1;
+    """
+}

regression-test/suites/query_p0/join/test_left_join_batch_size.groovy

@@ -0,0 +1,54 @@
+// 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_left_join_batch_size", "query,p0") {
+    sql " drop table if exists test_left_join_batch_size_l; ";
+    sql " drop table if exists test_left_join_batch_size_r; ";
+    sql """
+        create table test_left_join_batch_size_l (
+            k1 int,
+            v1 int
+        ) distributed by hash(k1) buckets 3
+        properties("replication_num" = "1");
+    """
+    sql """
+        create table test_left_join_batch_size_r (
+            k1 int,
+            v1 int
+        ) distributed by hash(k1) buckets 3
+        properties("replication_num" = "1");
+    """
+
+    sql """ insert into test_left_join_batch_size_l values (1, 11), (1, 111), (1, 1111) """
+    sql """ insert into test_left_join_batch_size_r values (1, null), (1, 211), (1, 311), (1, 411) """
+
+    qt_sql1 """
+        select /*+SET_VAR(batch_size=3)*/
+               l.k1,
+               l.v1,
+               r.k1,
+               r.v1
+        from
+               test_left_join_batch_size_l l
+               left join test_left_join_batch_size_r r on (
+                      r.v1 = 0
+                      or r.v1 is null
+               )
+               and l.k1 = r.k1
+                order by l.k1, l.v1, r.k1, r.v1;
+    """
+}