[fix](be) Correct null-aware mark join results for empty correlated subqueries on 4.1

be/src/exec/common/hash_table/join_hash_table.h

@@ -464,6 +464,8 @@ class JoinHashTable {
             if (!build_idx) {
                 probe_idxs[matched_cnt] = probe_idx;
                 build_idxs[matched_cnt] = 0;
+                // A previous NULL-key probe may leave 1 here; the no-match sentinel is not NULL.
+                null_flags[matched_cnt] = 0;
                 picking_null_keys = false;
                 matched_cnt++;
             }
@@ -481,7 +483,8 @@ class JoinHashTable {
             /// If the probe key is null
             if constexpr (has_null_map) {
                 if (null_map[probe_idx]) {
-                    probe_idx++;
+                    build_idx = 0;
+                    picking_null_keys = false;
                     break;
                 }
             }
@@ -514,4 +517,4 @@ class JoinHashTable {
 template <typename Key, typename Hash, bool DirectMapping>
 using JoinHashMap = JoinHashTable<Key, Hash, DirectMapping>;
 #include "common/compile_check_end.h"
-} // namespace doris
+} // namespace doris

be/test/exec/common/hash_table/join_hash_table_test.cpp

@@ -0,0 +1,84 @@
+// 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.
+
+#include "exec/common/hash_table/join_hash_table.h"
+
+#include <gtest/gtest.h>
+
+#include <array>
+#include <cstdint>
+#include <tuple>
+
+#include "exec/common/hash_table/hash_crc32_return32.h"
+
+namespace doris {
+
+TEST(JoinHashTableTest, NullAwareOtherConjunctSentinelResetsNullFlag) {
+    JoinHashTable<uint32_t, HashCRC32Return32<UInt32>, false> table;
+    table.prepare_build<TJoinOp::NULL_AWARE_LEFT_SEMI_JOIN>(2, 8, false, 0);
+
+    const std::array<uint32_t, 2> build_keys = {0, 7};
+    const std::array<uint32_t, 2> build_buckets = {
+            0, table.hash(build_keys[1]) & (table.get_bucket_size() - 1)};
+    table.build(build_keys.data(), build_buckets.data(), build_keys.size(), false);
+
+    const std::array<uint32_t, 1> probe_keys = {42};
+    const uint32_t probe_bucket = table.hash(probe_keys[0]) & (table.get_bucket_size() - 1);
+    const std::array<uint32_t, 1> build_idx_map = {table.get_first()[probe_bucket]};
+    std::array<uint32_t, 9> probe_idxs = {};
+    std::array<uint32_t, 9> build_idxs = {};
+    std::array<uint8_t, 9> null_flags = {};
+    null_flags[0] = 1; // A previous NULL-key probe left a NULL mark at this output position.
+
+    const auto [probe_idx, build_idx, matched_cnt, picking_null_keys] =
+            table.find_null_aware_with_other_conjuncts(
+                    probe_keys.data(), build_idx_map.data(), 0, 0, probe_keys.size(),
+                    probe_idxs.data(), build_idxs.data(), null_flags.data(), false, nullptr);
+
+    EXPECT_EQ(1, probe_idx);
+    EXPECT_EQ(0, build_idx);
+    ASSERT_EQ(1, matched_cnt);
+    EXPECT_FALSE(picking_null_keys);
+    EXPECT_EQ(0, probe_idxs[0]);
+    EXPECT_EQ(0, build_idxs[0]);
+    EXPECT_EQ(0, null_flags[0]);
+}
+
+TEST(JoinHashTableTest, NullAwareOtherConjunctNullProbeIsNotSkipped) {
+    JoinHashTable<uint32_t, HashCRC32Return32<UInt32>, false> table;
+    table.prepare_build<TJoinOp::NULL_AWARE_LEFT_SEMI_JOIN>(2, 8, false, 0);
+
+    const std::array<uint32_t, 2> probe_keys = {0, 42};
+    const std::array<uint32_t, 2> build_idx_map = {0, 0};
+    const std::array<uint8_t, 2> null_map = {1, 0};
+    std::array<uint32_t, 9> probe_idxs = {};
+    std::array<uint32_t, 9> build_idxs = {};
+    std::array<uint8_t, 9> null_flags = {};
+
+    const auto [probe_idx, build_idx, matched_cnt, picking_null_keys] =
+            table.find_null_aware_with_other_conjuncts(probe_keys.data(), build_idx_map.data(), 0,
+                                                       0, probe_keys.size(), probe_idxs.data(),
+                                                       build_idxs.data(), null_flags.data(), false,
+                                                       null_map.data());
+
+    EXPECT_EQ(0, probe_idx);
+    EXPECT_EQ(0, build_idx);
+    EXPECT_EQ(0, matched_cnt);
+    EXPECT_FALSE(picking_null_keys);
+}
+
+} // namespace doris

regression-test/data/nereids_syntax_p0/mark_join_correlated_empty.out

@@ -0,0 +1,11 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !in_empty --
+1	false
+2	false
+
+-- !not_in_empty --
+1	true
+2	true
+
+-- !in_empty_single_row --
+2	false

regression-test/suites/nereids_syntax_p0/mark_join_correlated_empty.groovy

@@ -0,0 +1,70 @@
+// 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("mark_join_correlated_empty") {
+    sql "DROP TABLE IF EXISTS mark_join_correlated_empty_l"
+    sql "DROP TABLE IF EXISTS mark_join_correlated_empty_r"
+
+    sql """
+        CREATE TABLE mark_join_correlated_empty_l (
+            id INT NOT NULL, k INT NULL, v INT NOT NULL
+        ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES ('replication_num' = '1')
+    """
+    sql """
+        CREATE TABLE mark_join_correlated_empty_r (
+            id INT NOT NULL, k INT NOT NULL, v INT NOT NULL
+        ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES ('replication_num' = '1')
+    """
+
+    sql "INSERT INTO mark_join_correlated_empty_l VALUES (1, NULL, 0), (2, 42, -1)"
+    sql "INSERT INTO mark_join_correlated_empty_r VALUES (1, 7, -2)"
+
+    explain {
+        sql """
+            SELECT l.id, l.k IN (SELECT r.k FROM mark_join_correlated_empty_r r
+                                  WHERE r.v > l.v) AS m
+            FROM (SELECT * FROM mark_join_correlated_empty_l ORDER BY id LIMIT 2) l
+            ORDER BY l.id
+        """
+        contains "NULL AWARE LEFT SEMI JOIN"
+        contains "mark join predicates"
+        contains "other join predicates"
+    }
+
+    order_qt_in_empty """
+        SELECT l.id, l.k IN (SELECT r.k FROM mark_join_correlated_empty_r r
+                              WHERE r.v > l.v) AS m
+        FROM (SELECT * FROM mark_join_correlated_empty_l ORDER BY id LIMIT 2) l
+        ORDER BY l.id
+    """
+
+    order_qt_not_in_empty """
+        SELECT l.id, l.k NOT IN (SELECT r.k FROM mark_join_correlated_empty_r r
+                                  WHERE r.v > l.v) AS m
+        FROM (SELECT * FROM mark_join_correlated_empty_l ORDER BY id LIMIT 2) l
+        ORDER BY l.id
+    """
+
+    order_qt_in_empty_single_row """
+        SELECT l.id, l.k IN (SELECT r.k FROM mark_join_correlated_empty_r r
+                              WHERE r.v > l.v) AS m
+        FROM (SELECT * FROM mark_join_correlated_empty_l WHERE id = 2 ORDER BY id LIMIT 1) l
+        ORDER BY l.id
+    """
+}