[fix](mark join) fix bug of mark join with other conjuncts (#16655)

Fix bug that probe_index is not increased for mark hash join with other conjuncts.

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

@@ -578,6 +578,7 @@ Status ProcessHashTableProbe<JoinOpType>::do_process_with_other_join_conjuncts(
                                 ++current_offset;
                                 visited_map.emplace_back(&it->visited);
                             }
+                            ++probe_index;
                         } else {
                             auto multi_match_last_offset = current_offset;
                             auto it = mapped.begin();

regression-test/data/correctness/test_subquery_in_disjunction.out

@@ -66,3 +66,19 @@
 1	2	3
 10	20	30
 
+-- !mark_join_with_other_conjuncts1 --
+1	2
+1	3
+2	4
+2	5
+3	3
+3	4
+
+-- !mark_join_with_other_conjuncts2 --
+1	2
+1	3
+2	4
+2	5
+3	3
+3	4
+

regression-test/suites/correctness/test_subquery_in_disjunction.groovy

@@ -120,4 +120,83 @@ suite("test_subquery_in_disjunction") {
     qt_two_subquery_in_one_conjuncts """
         SELECT * FROM test_sq_dj1 WHERE c1 IN (SELECT c1 FROM test_sq_dj2) OR c1 IN (SELECT c2 FROM test_sq_dj2) OR c1 < 10 ORDER BY c1;
     """
+
+    // test mark join that one probe row matches multiple build rows
+    sql """drop table if exists sub_query_correlated_subquery1;"""
+    sql """create table if not exists sub_query_correlated_subquery1
+        (k1 bigint, k2 bigint)
+        duplicate key(k1)
+        distributed by hash(k2) buckets 1
+        properties('replication_num' = '1');"""
+
+    sql """drop table if exists sub_query_correlated_subquery3;"""
+    sql """create table if not exists sub_query_correlated_subquery3
+        (kk1 int not null, k2 varchar(128), k3 bigint, v1 bigint, v2 bigint)
+        distributed by hash(k2) buckets 1
+        properties('replication_num' = '1');"""
+
+    sql """
+        insert into
+            sub_query_correlated_subquery1
+        values
+            (1, 2),
+            (1, 3),
+            (2, 4),
+            (2, 5),
+            (3, 3),
+            (3, 4),
+            (20, 2),
+            (22, 3),
+            (24, 4);
+    """
+    sql """
+        insert into
+            sub_query_correlated_subquery3
+        values
+            (1, "abc", 2, 3, 4),
+            (1, "abcd", 3, 3, 4),
+            (2, "xyz", 2, 4, 2),
+            (2, "uvw", 3, 4, 2),
+            (2, "uvw", 3, 4, 2),
+            (3, "abc", 4, 5, 3),
+            (3, "abc", 4, 5, 3);
+    """
+
+    qt_mark_join_with_other_conjuncts1 """
+        SELECT
+            *
+        FROM
+            sub_query_correlated_subquery1
+        WHERE
+            k1 IN (
+                SELECT
+                    kk1
+                FROM
+                    sub_query_correlated_subquery3
+                WHERE
+                    sub_query_correlated_subquery1.k1 > sub_query_correlated_subquery3.k3
+            )
+            OR k1 < 10
+        ORDER BY
+            k1, k2;
+    """
+
+    qt_mark_join_with_other_conjuncts2 """
+        SELECT
+            *
+        FROM
+            sub_query_correlated_subquery1
+        WHERE
+            k1 IN (
+                SELECT
+                    kk1
+                FROM
+                    sub_query_correlated_subquery3
+                WHERE
+                    sub_query_correlated_subquery1.k1 != sub_query_correlated_subquery3.k3
+            )
+            OR k1 < 10
+        ORDER BY
+            k1, k2;
+    """
 }