MCOL-5142, Implementation of Recursive CTE

cmake/cpack_manage.cmake

@@ -1,3 +1,27 @@
+if(NOT RPM)
+    return()
+endif()
+
+# Define %ignore as a comment so RPM doesn't claim ownership of shared directories
+set(CPACK_RPM_SPEC_MORE_DEFINE "${CPACK_RPM_SPEC_MORE_DEFINE}
+%define ignore \#
+")
+
+set(ignored
+    "%ignore /usr"
+    "%ignore /usr/local"
+    "%ignore /bin"
+    "%ignore /lib"
+    "%ignore /usr/sbin"
+    "%ignore /usr/lib64/mysql"
+    "%ignore /usr/lib64/mysql/plugin"
+    "%ignore /etc/my.cnf.d"
+    "%ignore /var/lib"
+    "%ignore /var"
+)
+
+set(CPACK_RPM_columnstore-engine_USER_FILELIST ${ignored})
+
 macro(columnstore_append_for_cpack var_name)
     # Get current value from parent scope or use empty string
     if(DEFINED ${var_name})
@@ -28,31 +52,29 @@ macro(columnstore_add_rpm_deps)
     columnstore_append_for_cpack(CPACK_RPM_columnstore-engine_PACKAGE_REQUIRES ${ARGN})
 endmacro()
 
-if(NOT COLUMNSTORE_MAINTAINER)
-    return()
-endif()
-
-# Columnstore-specific RPM packaging overrides 1) Use fast compression to speed up packaging
-set(CPACK_RPM_COMPRESSION_TYPE
-    "zstd"
-    CACHE STRING "RPM payload compression" FORCE
-)
-# 2) Disable debuginfo/debugsource to avoid slow packaging and duplicate file warnings
-set(CPACK_RPM_DEBUGINFO_PACKAGE
-    OFF
-    CACHE BOOL "Disable debuginfo package" FORCE
-)
-set(CPACK_RPM_PACKAGE_DEBUG
-    0
-    CACHE STRING "Disable RPM debug package" FORCE
-)
-unset(CPACK_RPM_BUILD_SOURCE_DIRS_PREFIX CACHE)
+columnstore_add_rpm_deps("snappy" "jemalloc" "procps-ng" "gawk")
 
-# Ensure our overrides are applied by CPack at packaging time CPACK_PROJECT_CONFIG_FILE is included by cpack after
-# CPackConfig.cmake is loaded
-set(CPACK_PROJECT_CONFIG_FILE
-    "${CMAKE_CURRENT_LIST_DIR}/cpack_overrides.cmake"
-    CACHE FILEPATH "Columnstore CPack overrides" FORCE
-)
+if(COLUMNSTORE_MAINTAINER)
+    # Columnstore-specific RPM packaging overrides 1) Use fast compression to speed up packaging
+    set(CPACK_RPM_COMPRESSION_TYPE
+        "zstd"
+        CACHE STRING "RPM payload compression" FORCE
+    )
+    # 2) Disable debuginfo/debugsource to avoid slow packaging and duplicate file warnings
+    set(CPACK_RPM_DEBUGINFO_PACKAGE
+        OFF
+        CACHE BOOL "Disable debuginfo package" FORCE
+    )
+    set(CPACK_RPM_PACKAGE_DEBUG
+        0
+        CACHE STRING "Disable RPM debug package" FORCE
+    )
+    unset(CPACK_RPM_BUILD_SOURCE_DIRS_PREFIX CACHE)
 
-columnstore_add_rpm_deps("snappy" "jemalloc" "procps-ng" "gawk")
+    # Ensure our overrides are applied by CPack at packaging time CPACK_PROJECT_CONFIG_FILE is included by cpack after
+    # CPackConfig.cmake is loaded
+    set(CPACK_PROJECT_CONFIG_FILE
+        "${CMAKE_CURRENT_LIST_DIR}/cpack_overrides.cmake"
+        CACHE FILEPATH "Columnstore CPack overrides" FORCE
+    )
+endif() # COLUMNSTORE_MAINTAINER

dbcon/execplan/calpontselectexecutionplan.cpp

@@ -628,6 +628,11 @@ void CalpontSelectExecutionPlan::serialize(messageqcpp::ByteStream& b) const
   b << timeZone;
   b << fPron;
   b << (uint8_t)fWithRollup;
+  b << (uint8_t)fIsRecursiveWithTable;
+  b << (uint8_t)fIsRecursiveQuery;
+  b << (uint8_t)fContainsRecursiveQuery;
+
+  b << fMaxRecursiveDepth;
 }
 
 void CalpontSelectExecutionPlan::unserialize(messageqcpp::ByteStream& b)
@@ -832,6 +837,13 @@ void CalpontSelectExecutionPlan::unserialize(messageqcpp::ByteStream& b)
   utils::Pron::instance().pron(fPron);
   b >> tmp8;
   fWithRollup = tmp8;
+  b >> tmp8;
+  fIsRecursiveWithTable = tmp8;
+  b >> tmp8;
+  fIsRecursiveQuery = tmp8;
+  b >> tmp8;
+  fContainsRecursiveQuery = tmp8;
+  b >> fMaxRecursiveDepth;
 }
 
 bool CalpontSelectExecutionPlan::operator==(const CalpontSelectExecutionPlan& t) const

dbcon/execplan/calpontselectexecutionplan.h

@@ -23,6 +23,7 @@
 /** @file */
 
 #pragma once
+#include <cstdint>
 #include <vector>
 #include <map>
 #include <iosfwd>
@@ -496,7 +497,12 @@ class CalpontSelectExecutionPlan : public CalpontExecutionPlan
   {
     return fDerivedTableList;
   }
-  void derivedTableList(const SelectList& derivedTableList)
+
+  SelectList& derivedTableList()
+  {
+    return fDerivedTableList;
+  }
+  void derivedTableList(SelectList& derivedTableList)
   {
     fDerivedTableList = derivedTableList;
   }
@@ -523,10 +529,12 @@ class CalpontSelectExecutionPlan : public CalpontExecutionPlan
   {
     fUnionVec = unionVec;
   }
+
   const SelectList& unionVec() const
   {
     return fUnionVec;
   }
+
   SelectList& unionVec()
   {
     return fUnionVec;
@@ -765,6 +773,46 @@ class CalpontSelectExecutionPlan : public CalpontExecutionPlan
     return fTimeZone;
   }
 
+  void isRecursiveWithTable(bool b)
+  {
+    fIsRecursiveWithTable = b;
+  }
+
+  bool isRecursiveWithTable()
+  {
+    return fIsRecursiveWithTable;
+  }
+
+  void isRecursiveQuery(bool b)
+  {
+    fIsRecursiveQuery = b;
+  }
+
+  bool isRecursiveQuery()
+  {
+    return fIsRecursiveQuery;
+  }
+
+  void containsRecursiveQuery(bool b)
+  {
+    fContainsRecursiveQuery = b;
+  }
+
+  bool containsRecursiveQuery()
+  {
+    return fContainsRecursiveQuery;
+  }
+
+  void maxRecursiveDepth(uint32_t i)
+  {
+    fMaxRecursiveDepth = i;
+  }
+
+  int maxRecursiveDepth()
+  {
+    return fMaxRecursiveDepth;
+  }
+
   /**
    * The serialization interface
    */
@@ -985,6 +1033,11 @@ class CalpontSelectExecutionPlan : public CalpontExecutionPlan
    * A flag to compute subtotals, related to GROUP BY operation.
    */
   bool fWithRollup;
+  bool fIsRecursiveWithTable = false;
+  bool fIsRecursiveQuery = false;
+  bool fContainsRecursiveQuery = false;
+
+  uint32_t fMaxRecursiveDepth;
 };
 
 /**

dbcon/joblist/fifo.h

@@ -83,7 +83,7 @@ class FIFO : public DataListImpl<std::vector<element_t>, element_t>
   }
 
   inline void dropToken() {};
-  inline void dropToken(uint32_t){};
+  inline void dropToken(uint32_t) {};
 
   // Counters that reflect how many many times this FIFO blocked on reads/writes
   uint64_t blockedWriteCount() const;

dbcon/joblist/jlf_subquery.cpp

@@ -21,7 +21,7 @@
 #include <iostream>
 #include <stack>
 #include <iterator>
-//#define NDEBUG
+// #define NDEBUG
 #include <cassert>
 #include <vector>
 using namespace std;
@@ -748,6 +748,10 @@ int doFromSubquery(CalpontExecutionPlan* ep, const string& alias, const string&
   SJSTEP subQueryStep = transformer.makeSubQueryStep(csep, true);
   subQueryStep->view(view);
   SJSTEP subAd(new SubAdapterStep(subQueryStep, jobInfo));
+  if (csep->isRecursiveQuery())
+  {
+    dynamic_cast<SubAdapterStep*>(subAd.get())->isRecursiveStep(true);
+  }
   jobInfo.selectAndFromSubs.push_back(subAd);
 
   return CNX_VTABLE_ID;
@@ -870,6 +874,10 @@ SJSTEP doUnionSub(CalpontExecutionPlan* ep, JobInfo& jobInfo)
   transformer.setVarbinaryOK();
   SJSTEP subQueryStep = transformer.makeSubQueryStep(csep, false);
   SJSTEP subAd(new SubAdapterStep(subQueryStep, jobInfo));
+  if (csep->isRecursiveQuery())
+  {
+    dynamic_cast<SubAdapterStep*>(subAd.get())->isRecursiveStep(true);
+  }
   return subAd;
 }