feat(shannon): parallel query processing

include/mem_root_deque.h

@@ -1,4 +1,5 @@
 /* Copyright (c) 2019, 2023, Oracle and/or its affiliates.
+   Copyright (c) 2021, Huawei Technologies Co., Ltd.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License, version 2.0,
@@ -18,7 +19,9 @@
 
    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+   Copyright (c) 2023, Shannon Data AI and/or its affiliates. */
 
 #ifndef MEM_ROOT_DEQUE_H
 #define MEM_ROOT_DEQUE_H
@@ -166,7 +169,7 @@ class mem_root_deque {
   ~mem_root_deque() { clear(); }
 
   Element_type &operator[](size_t idx) const { return get(idx + m_begin_idx); }
-
+  Element_type &at(size_t idx) const { return get(idx + m_begin_idx); }
   /**
     Adds the given element to the end of the deque.
     The element is a copy of the given one.

include/my_alloc.h

@@ -1,4 +1,5 @@
 /* Copyright (c) 2000, 2023, Oracle and/or its affiliates.
+   Copyright (c) 2021, Huawei Technologies Co., Ltd.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License, version 2.0,
@@ -18,7 +19,9 @@
 
    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+   Copyright (c) 2023, Shannon Data AI and/or its affiliates. */
 
 /**
  * @file include/my_alloc.h
@@ -36,6 +39,7 @@
 #include <new>
 #include <type_traits>
 #include <utility>
+#include <functional>
 
 #include "memory_debugging.h"
 #include "my_compiler.h"
@@ -48,6 +52,10 @@
 extern "C" void sql_alloc_error_handler();
 #endif
 
+
+using CallBackFunc = std::function<void (PSI_memory_key, size_t, uint)>;
+constexpr uint PQ_MEMORY_USED_BUCKET = 16;
+
 /**
  * The MEM_ROOT is a simple arena, where allocations are carved out of
  * larger blocks. Using an arena over plain malloc gives you two main
@@ -359,6 +367,8 @@ struct MEM_ROOT {
     return false;
   }
 
+  void set_alloc_func(CallBackFunc func) { allocCBFunc = func; }
+  void set_free_func(CallBackFunc func) { freeCBFunc = func; }
   /// @}
 
  private:
@@ -416,6 +426,10 @@ struct MEM_ROOT {
   void (*m_error_handler)(void) = nullptr;
 
   PSI_memory_key m_psi_key = 0;
+
+  CallBackFunc allocCBFunc = nullptr;
+
+  CallBackFunc freeCBFunc = nullptr;
 };
 
 /**

include/my_dbug.h

@@ -1,4 +1,5 @@
 /* Copyright (c) 2000, 2023, Oracle and/or its affiliates.
+   Copyright (c) 2021, Huawei Technologies Co., Ltd.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License, version 2.0,
@@ -18,7 +19,9 @@
 
    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+   Copyright (c) 2023, Shannon Data AI and/or its affiliates. */
 
 #ifndef MY_DBUG_INCLUDED
 #define MY_DBUG_INCLUDED
@@ -91,6 +94,9 @@ extern void _db_unlock_file_(void);
 extern FILE *_db_fp_(void);
 extern void _db_flush_();
 
+extern void db_stack_copy(CODE_STATE *leader_cs);
+extern void db_stack_reset();
+
 #ifdef __cplusplus
 
 #if defined(__GNUC__)

include/sql_string.h

@@ -606,6 +606,12 @@ class String {
     @return allocated string or NULL
   */
   char *dup(MEM_ROOT *root) const;
+
+  void ltrim();
+
+  void rtrim();
+
+  void rtrim_zero();  
 };
 
 /**

mysys/dbug.cc

@@ -926,6 +926,19 @@ void _db_pop_() {
   }
 }
 
+void db_stack_copy(CODE_STATE *leader_cs) {
+  CODE_STATE *cs;
+  get_code_state_or_return;
+  assert(cs->stack == &init_settings);
+  cs->stack = leader_cs->stack;
+}
+
+void db_stack_reset() {
+  CODE_STATE *cs;
+  get_code_state_or_return;
+  cs->stack = &init_settings;
+}
+
 /*
  *  FUNCTION
  *

sql-common/sql_string.cc

@@ -1001,6 +1001,69 @@ char *String::dup(MEM_ROOT *root) const {
   return strmake_root(root, m_ptr, m_length);
 }
 
+void String::ltrim() {
+  if (m_ptr) {
+    auto pos = m_ptr;
+    while (m_length > 0 && *pos && my_isspace(m_charset, *pos)) {
+      pos++;
+      m_length--;
+    }
+
+    if (pos != m_ptr) {
+      memmove(m_ptr, pos, m_length);
+      if (m_length < m_alloced_length)
+        m_ptr[m_length] = 0;
+      else
+        mem_realloc_exp(m_length);
+    }
+  }
+}
+
+void String::rtrim() {
+  if (m_ptr) {
+    while ((m_length > 0) && my_isspace(m_charset, *(m_ptr + m_length - 1))) {
+      m_length--;
+    }
+    if (m_length < m_alloced_length)
+      m_ptr[m_length] = 0;
+    else
+      mem_realloc_exp(m_length);
+  }
+}
+
+void String::rtrim_zero() {
+  bool proint_flag = false;
+  size_t length = m_length;
+  if (m_ptr) {
+    while (length > 0) {
+      if (my_toupper(m_charset, *(m_ptr + length - 1)) == '.') {
+        proint_flag = true;
+        break;
+      }
+      length--;
+    }
+    // trim right last 0, when after '.'
+    while ((m_length > 0) &&
+           (my_toupper(m_charset, *(m_ptr + m_length - 1)) == '0') &&
+           proint_flag) {
+      m_length--;
+    }
+    if (my_toupper(m_charset, *(m_ptr + m_length - 1)) == '.') {
+      m_length--;
+      // convert string '.' to '0'
+      if (m_length == 0) {
+        m_length = 1;
+        *m_ptr = '0';
+      }
+    }
+
+    if (m_length < m_alloced_length)
+      m_ptr[m_length] = 0;
+    else
+      mem_realloc_exp(m_length);
+  }
+}
+
 /**
   Convert string to printable ASCII string
 

sql/CMakeLists.txt

@@ -392,6 +392,7 @@ SET(SQL_SHARED_SOURCES
   debug_sync.cc
   default_values.cc
   derror.cc
+  exchange.cc
   error_handler.cc
   field.cc
   field_conv.cc
@@ -569,6 +570,8 @@ SET(SQL_SHARED_SOURCES
   sql_error.cc
   sql_exception_handler.cc
   sql_executor.cc
+  sql_exchange_nosort.cc
+  sql_exchange_sort.cc
   sql_get_diagnostics.cc
   sql_gipk.cc
   sql_handler.cc
@@ -582,7 +585,9 @@ SET(SQL_SHARED_SOURCES
   sql_load.cc
   sql_locale.cc
   sql_manager.cc
+  sql_message_queue.cc
   sql_optimizer.cc
+  sql_parallel.cc
   sql_parse.cc
   sql_partition.cc
   sql_partition_admin.cc
@@ -592,6 +597,7 @@ SET(SQL_SHARED_SOURCES
   sql_prepare.cc
   sql_profile.cc
   sql_pq_condition.cc
+  sql_pq_clone.cc
   sql_query_rewrite.cc
   sql_reload.cc
   sql_rename.cc

sql/cmp_varlen_keys.h

@@ -1,4 +1,5 @@
 /* Copyright (c) 2016, 2023, Oracle and/or its affiliates.
+   Copyright (c) 2021, Huawei Technologies Co., Ltd.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License, version 2.0,
@@ -18,7 +19,9 @@
 
    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+   Copyright (c) 2023, Shannon Data AI and/or its affiliates. */
 
 #ifndef CMP_VARLEN_KEYS_INCLUDED
 #define CMP_VARLEN_KEYS_INCLUDED
@@ -44,7 +47,7 @@
  */
 inline bool cmp_varlen_keys(
     Bounds_checked_array<st_sort_field> sort_field_array, bool use_hash,
-    const uchar *s1, const uchar *s2) {
+    const uchar *s1, const uchar *s2, int *pq_cmp_value = nullptr) {
   const uchar *kp1 = s1 + Sort_param::size_of_varlength_field;
   const uchar *kp2 = s2 + Sort_param::size_of_varlength_field;
   int res;
@@ -53,7 +56,11 @@ inline bool cmp_varlen_keys(
     if (sort_field.maybe_null) {
       const int k1_nullbyte = *kp1++;
       const int k2_nullbyte = *kp2++;
-      if (k1_nullbyte != k2_nullbyte) return k1_nullbyte < k2_nullbyte;
+
+      if (k1_nullbyte != k2_nullbyte) {
+        if (pq_cmp_value) *pq_cmp_value = k1_nullbyte - k2_nullbyte;
+        return k1_nullbyte < k2_nullbyte;
+      }
       if (k1_nullbyte == 0 || k1_nullbyte == 0xff) {
         if (!sort_field.is_varlen) {
           kp1 += sort_field.length;
@@ -79,12 +86,19 @@ inline bool cmp_varlen_keys(
 
     res = memcmp(kp1, kp2, kp_len);
 
-    if (res) return res < 0;
+    if (res) {
+      if (pq_cmp_value) *pq_cmp_value = res;
+      return res < 0;
+    }
+
     if (kp1_len != kp2_len) {
-      if (sort_field.reverse)
+      if (sort_field.reverse) {
+        if (pq_cmp_value) *pq_cmp_value = kp2_len - kp1_len;
         return kp2_len < kp1_len;
-      else
+      } else {
+        if (pq_cmp_value) *pq_cmp_value = kp1_len - kp2_len;
         return kp1_len < kp2_len;
+      }
     }
 
     kp1 += kp1_len;
@@ -93,8 +107,11 @@ inline bool cmp_varlen_keys(
 
   if (use_hash) {
     // Compare hashes at the end of sort keys
-    return memcmp(kp1, kp2, 8) < 0;
+    int cmp_value = memcmp(kp1, kp2, 8);
+    if (pq_cmp_value) *pq_cmp_value = cmp_value;
+    return cmp_value < 0;
   } else {
+    if (pq_cmp_value) *pq_cmp_value = 1;
     return false;
   }
 }