[fix](be) Fix json contains duplicate array candidates

be/src/util/jsonb_document.h

@@ -1002,6 +1002,30 @@ struct ArrayVal : public ContainerVal {
     const_iterator end() const { return const_iterator((pointer)(payload + size)); }
 };
 
+namespace jsonb_detail {
+
+inline bool array_contains_value(const ArrayVal* target_array, const JsonbValue* candidate) {
+    const int target_num = target_array->numElem();
+    for (int i = 0; i < target_num; ++i) {
+        if (target_array->get(i)->contains(candidate)) {
+            return true;
+        }
+    }
+    return false;
+}
+
+inline bool array_contains_array(const ArrayVal* target_array, const ArrayVal* candidate_array) {
+    const int candidate_num = candidate_array->numElem();
+    for (int i = 0; i < candidate_num; ++i) {
+        if (!array_contains_value(target_array, candidate_array->get(i))) {
+            return false;
+        }
+    }
+    return true;
+}
+
+} // namespace jsonb_detail
+
 inline const JsonbValue* JsonbDocument::createValue(const char* pb, size_t size) {
     if (!pb || size < sizeof(JsonbHeader) + sizeof(JsonbValue)) {
         return nullptr;
@@ -1153,29 +1177,11 @@ inline bool JsonbValue::contains(const JsonbValue* rhs) const {
         return false;
     }
     case JsonbType::T_Array: {
-        int lhs_num = unpack<ArrayVal>()->numElem();
+        const auto* lhs_array = unpack<ArrayVal>();
         if (rhs->isArray()) {
-            int rhs_num = rhs->unpack<ArrayVal>()->numElem();
-            if (rhs_num > lhs_num) {
-                return false;
-            }
-            int contains_num = 0;
-            for (int i = 0; i < lhs_num; ++i) {
-                for (int j = 0; j < rhs_num; ++j) {
-                    if (unpack<ArrayVal>()->get(i)->contains(rhs->unpack<ArrayVal>()->get(j))) {
-                        contains_num++;
-                        break;
-                    }
-                }
-            }
-            return contains_num == rhs_num;
+            return jsonb_detail::array_contains_array(lhs_array, rhs->unpack<ArrayVal>());
         }
-        for (int i = 0; i < lhs_num; ++i) {
-            if (unpack<ArrayVal>()->get(i)->contains(rhs)) {
-                return true;
-            }
-        }
-        return false;
+        return jsonb_detail::array_contains_value(lhs_array, rhs);
     }
     case JsonbType::T_Object: {
         if (rhs->isObject()) {

be/test/util/jsonb_contains_test.cpp

@@ -0,0 +1,64 @@
+// 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 <gtest/gtest.h>
+
+#include <string_view>
+
+#include "core/value/jsonb_value.h"
+#include "util/jsonb_document.h"
+
+namespace doris {
+namespace {
+
+void expect_jsonb_contains(std::string_view target_json, std::string_view candidate_json,
+                           bool expected) {
+    JsonBinaryValue target;
+    auto st = target.from_json_string(target_json.data(), target_json.size());
+    ASSERT_TRUE(st.ok()) << st.to_string();
+
+    JsonBinaryValue candidate;
+    st = candidate.from_json_string(candidate_json.data(), candidate_json.size());
+    ASSERT_TRUE(st.ok()) << st.to_string();
+
+    const JsonbDocument* target_doc = nullptr;
+    st = JsonbDocument::checkAndCreateDocument(target.value(), target.size(), &target_doc);
+    ASSERT_TRUE(st.ok()) << st.to_string();
+
+    const JsonbDocument* candidate_doc = nullptr;
+    st = JsonbDocument::checkAndCreateDocument(candidate.value(), candidate.size(), &candidate_doc);
+    ASSERT_TRUE(st.ok()) << st.to_string();
+
+    EXPECT_EQ(target_doc->getValue()->contains(candidate_doc->getValue()), expected);
+}
+
+} // namespace
+
+TEST(JsonbContainsTest, ArrayCandidateDoesNotConsumeTargetElements) {
+    expect_jsonb_contains("[1,1,1]", "[1,1]", true);
+    expect_jsonb_contains("[1]", "[1,1]", true);
+    expect_jsonb_contains("[1,2,3]", "[2,1]", true);
+
+    expect_jsonb_contains("[1,2,3]", "[2,4]", false);
+}
+
+TEST(JsonbContainsTest, ArrayCandidateUsesRecursiveContains) {
+    expect_jsonb_contains(R"([{"a":1,"b":2},[3,4]])", R"([{"a":1},[4]])", true);
+    expect_jsonb_contains(R"([{"a":1},[3,4]])", R"([{"a":2}])", false);
+}
+
+} // namespace doris