MDEV-27677: Implement JSON_OVERLAPS()

1) When at least one of the two json documents is of scalar type:
     1.a) If value and json document both are scalar, then return true
          if they have same type and value.
     1.b) If json document is scalar but other is array (or vice versa),
          then return true if array has at least one element of same type
          and value as scalar.
     1.c) If one is scalar and other is object, then return false because
          it can't be compared.

  2) When both arguments are of non-scalar type and below conditons
      are satisfied then return true:
      2.a) When both arguments are arrays:
           Iterate over the value and json document. If there exists at
           least one element in other array of same type and value as
           that of element in value.
      2.b) If both arguments are objects:
           Iterate over value and json document and if there exists at least
           one key-value pair common between two objects.
      2.c) If either of json document or value is array and other is object:
           Iterate over the array, if an element of type object is found,
           then compare it with the object (which is the other arguemnt).
           If the entire object matches i.e all they key value pairs match.

1 files changed, 367 insertions, 0 deletions

diff --git a/sql/item_jsonfunc.cc b/sql/item_jsonfunc.cc
index 89ac59098ac..a0dfcaae6e3 100644
--- a/sql/item_jsonfunc.cc
+++ b/sql/item_jsonfunc.cc
@@ -4031,3 +4031,370 @@ bool Item_func_json_normalize::fix_length_and_dec()
   set_maybe_null();
   return FALSE;
 }
+
+
+/*
+  When the two values match or don't match we need to return true or false.
+  But we can have some more elements in the array left or some more keys
+  left in the object that we no longer want to compare. In this case,
+  we want to skip the current item.
+*/
+void json_skip_current_level(json_engine_t *js, json_engine_t *value)
+{
+  json_skip_level(js);
+  json_skip_level(value);
+}
+
+
+/* At least one of the two arguments is a scalar. */
+bool json_find_overlap_with_scalar(json_engine_t *js, json_engine_t *value)
+{
+  if (json_value_scalar(value))
+  {
+    if (js->value_type == value->value_type)
+    {
+      if (js->value_type == JSON_VALUE_NUMBER)
+      {
+        double d_j, d_v;
+        char *end;
+        int err;
+
+        d_j= js->s.cs->strntod((char *) js->value, js->value_len, &end, &err);
+        d_v= value->s.cs->strntod((char *) value->value, value->value_len,
+                                   &end, &err);
+
+        return (fabs(d_j - d_v) < 1e-12);
+      }
+      else if (js->value_type == JSON_VALUE_STRING)
+      {
+        return value->value_len == js->value_len &&
+               memcmp(value->value, js->value, value->value_len) == 0;
+      }
+    }
+    return value->value_type == js->value_type;
+  }
+  else if (value->value_type == JSON_VALUE_ARRAY)
+  {
+    while (json_scan_next(value) == 0 && value->state == JST_VALUE)
+    {
+      if (json_read_value(value))
+        return FALSE;
+      if (js->value_type == value->value_type)
+      {
+        int res1= json_find_overlap_with_scalar(js, value);
+        if (res1)
+          return TRUE;
+      }
+      if (!json_value_scalar(value))
+        json_skip_level(value);
+    }
+  }
+  return FALSE;
+}
+
+
+/*
+  Compare when one is object and other is array. This means we are looking
+  for the object in the array. Hence, when value type of an element of the
+  array is object, then compare the two objects entirely. If they are
+  equal return true else return false.
+*/
+bool json_compare_arr_and_obj(json_engine_t *js, json_engine_t *value)
+{
+  st_json_engine_t loc_val= *value;
+  while (json_scan_next(js) == 0 && js->state == JST_VALUE)
+  {
+    if (json_read_value(js))
+      return FALSE;
+    if (js->value_type == JSON_VALUE_OBJECT)
+    {
+      int res1= json_find_overlap_with_object(js, value, true);
+      if (res1)
+        return TRUE;
+      *value= loc_val;
+    }
+    if (!json_value_scalar(js))
+      json_skip_level(js);
+  }
+  return FALSE;
+}
+
+
+bool json_compare_arrays_in_order(json_engine_t *js, json_engine_t *value)
+{
+  bool res= false;
+  while (json_scan_next(js) == 0 && json_scan_next(value) == 0 &&
+         js->state == JST_VALUE && value->state == JST_VALUE)
+  {
+    if (json_read_value(js) || json_read_value(value))
+      return FALSE;
+    if (js->value_type != value->value_type)
+    {
+      json_skip_current_level(js, value);
+      return FALSE;
+    }
+    res= check_overlaps(js, value, true);
+    if (!res)
+    {
+      json_skip_current_level(js, value);
+      return FALSE;
+    }
+  }
+  res= (value->state == JST_ARRAY_END || value->state == JST_OBJ_END ?
+        TRUE : FALSE);
+  json_skip_current_level(js, value);
+  return res;
+}
+
+
+int json_find_overlap_with_array(json_engine_t *js, json_engine_t *value,
+                                 bool compare_whole)
+{
+  if (value->value_type == JSON_VALUE_ARRAY)
+  {
+    if (compare_whole)
+      return json_compare_arrays_in_order(js, value);
+
+    json_engine_t loc_value= *value, current_js= *js;
+
+    while (json_scan_next(js) == 0 && js->state == JST_VALUE)
+    {
+      if (json_read_value(js))
+        return FALSE;
+      current_js= *js;
+      while (json_scan_next(value) == 0 && value->state == JST_VALUE)
+      {
+        if (json_read_value(value))
+          return FALSE;
+        if (js->value_type == value->value_type)
+        {
+          int res1= check_overlaps(js, value, true);
+          if (res1)
+            return TRUE;
+        }
+        else
+        {
+          if (!json_value_scalar(value))
+            json_skip_level(value);
+        }
+        *js= current_js;
+      }
+      *value= loc_value;
+      if (!json_value_scalar(js))
+        json_skip_level(js);
+    }
+    return FALSE;
+  }
+  else if (value->value_type == JSON_VALUE_OBJECT)
+  {
+    if (compare_whole)
+    {
+      json_skip_current_level(js, value);
+      return FALSE;
+    }
+    return json_compare_arr_and_obj(js, value);
+  }
+  else
+    return json_find_overlap_with_scalar(value, js);
+}
+
+
+int json_find_overlap_with_object(json_engine_t *js, json_engine_t *value,
+                                  bool compare_whole)
+{
+  if (value->value_type == JSON_VALUE_OBJECT)
+  {
+    /* Find at least one common key-value pair */
+    json_string_t key_name;
+    bool found_key= false, found_value= false;
+    json_engine_t loc_js= *js;
+    const uchar *k_start, *k_end;
+
+    json_string_set_cs(&key_name, value->s.cs);
+
+    while (json_scan_next(value) == 0 && value->state == JST_KEY)
+    {
+      k_start= value->s.c_str;
+      do
+      {
+        k_end= value->s.c_str;
+      } while (json_read_keyname_chr(value) == 0);
+
+      if (unlikely(value->s.error))
+        return FALSE;
+
+      json_string_set_str(&key_name, k_start, k_end);
+      found_key= find_key_in_object(js, &key_name);
+      found_value= 0;
+
+      if (found_key)
+      {
+        if (json_read_value(js) || json_read_value(value))
+          return FALSE;
+
+        /*
+          The value of key-value pair can be an be anything. If it is an object
+          then we need to compare the whole value and if it is an array then
+          we need to compare the elements in that order. So set compare_whole
+          to true.
+        */
+        if (js->value_type == value->value_type)
+          found_value= check_overlaps(js, value, true);
+        if (found_value)
+        {
+          if (!compare_whole)
+            return TRUE;
+          *js= loc_js;
+        }
+        else
+        {
+          if (compare_whole)
+          {
+            json_skip_current_level(js, value);
+            return FALSE;
+          }
+          *js= loc_js;
+        }
+      }
+      else
+      {
+        if (compare_whole)
+        {
+          json_skip_current_level(js, value);
+          return FALSE;
+        }
+        json_skip_key(value);
+        *js= loc_js;
+      }
+    }
+    json_skip_current_level(js, value);
+    return compare_whole ? TRUE : FALSE;
+  }
+  else if (value->value_type == JSON_VALUE_ARRAY)
+  {
+    if (compare_whole)
+    {
+      json_skip_current_level(js, value);
+      return FALSE;
+    }
+    return json_compare_arr_and_obj(value, js);
+  }
+  return FALSE;
+}
+
+
+/*
+  Find if two json documents overlap
+
+  SYNOPSIS
+    check_overlaps()
+    js     - json document
+    value  - value
+    compare_whole - If true then find full overlap with the document in case of
+                    object and comparing in-order in case of array.
+                    Else find at least one match between two objects or array.
+
+  IMPLEMENTATION
+  We can compare two json datatypes if they are of same type to check if
+  they are equal. When comparing between a json document and json value,
+  there can be following cases:
+  1) When at least one of the two json documents is of scalar type:
+     1.a) If value and json document both are scalar, then return true
+          if they have same type and value.
+     1.b) If json document is scalar but other is array (or vice versa),
+          then return true if array has at least one element of same type
+          and value as scalar.
+     1.c) If one is scalar and other is object, then return false because
+          it can't be compared.
+
+  2) When both arguments are of non-scalar type:
+      2.a) If both arguments are arrays:
+           Iterate over the value and json document. If there exists at least
+           one element in other array of same type and value as that of
+           element in value, then return true else return false.
+      2.b) If both arguments are objects:
+           Iterate over value and json document and if there exists at least
+           one key-value pair common between two objects, then return true,
+           else return false.
+      2.c) If either of json document or value is array and other is object:
+           Iterate over the array, if an element of type object is found,
+           then compare it with the object (which is the other arguemnt).
+           If the entire object matches i.e all they key value pairs match,
+           then return true else return false.
+
+  When we are comparing an object which is nested in other object or nested
+  in an array, we need to compare all the key-value pairs, irrespective of
+  what order they are in as opposed to non-nested where we return true if
+  at least one match is found. However, if we have an array nested in another
+  array, then we compare two arrays in that order i.e we compare
+  i-th element of array 1 with i-th element of array 2.
+
+  RETURN
+    FALSE - If two json documents do not overlap
+    TRUE  - if two json documents overlap
+*/
+int check_overlaps(json_engine_t *js, json_engine_t *value, bool compare_whole)
+{
+  switch (js->value_type)
+  {
+  case JSON_VALUE_OBJECT:
+    return json_find_overlap_with_object(js, value, compare_whole);
+  case JSON_VALUE_ARRAY:
+    return json_find_overlap_with_array(js, value, compare_whole);
+  default:
+    return json_find_overlap_with_scalar(js, value);
+  }
+}
+
+longlong Item_func_json_overlaps::val_int()
+{
+  String *js= args[0]->val_json(&tmp_js);
+  json_engine_t je, ve;
+  int result;
+
+  if ((null_value= args[0]->null_value))
+    return 0;
+
+  if (!a2_parsed)
+  {
+    val= args[1]->val_json(&tmp_val);
+    a2_parsed= a2_constant;
+  }
+
+  if (val == 0)
+  {
+    null_value= 1;
+    return 0;
+  }
+
+  json_scan_start(&je, js->charset(), (const uchar *) js->ptr(),
+                  (const uchar *) js->ptr() + js->length());
+
+  json_scan_start(&ve, val->charset(), (const uchar *) val->ptr(),
+                  (const uchar *) val->end());
+
+  if (json_read_value(&je) || json_read_value(&ve))
+    goto error;
+
+  result= check_overlaps(&je, &ve, false);
+  if (unlikely(je.s.error || ve.s.error))
+    goto error;
+
+  return result;
+
+error:
+  if (je.s.error)
+    report_json_error(js, &je, 0);
+  if (ve.s.error)
+    report_json_error(val, &ve, 1);
+  return 1;
+}
+
+bool Item_func_json_overlaps::fix_length_and_dec()
+{
+  a2_constant= args[1]->const_item();
+  a2_parsed= FALSE;
+  set_maybe_null();
+
+  return Item_bool_func::fix_length_and_dec();
+}