fix(logging): Add setddldebuglevel command (#3312)

dbcon/ddlpackage/CMakeLists.txt

@@ -32,6 +32,7 @@ ADD_LIBRARY(ddlpackage SHARED
     markpartition.cpp
     restorepartition.cpp
     droppartition.cpp
+    debugstatement.cpp
     ${BISON_ddl_gram_OUTPUTS}
     ${FLEX_ddl_scan_OUTPUTS}
 )

dbcon/ddlpackage/ddlpkg.h

@@ -332,7 +332,8 @@ enum DDL_SERIAL_TYPE
   DDL_TRUNC_TABLE_STATEMENT,
   DDL_MARK_PARTITION_STATEMENT,
   DDL_RESTORE_PARTITION_STATEMENT,
-  DDL_DROP_PARTITION_STATEMENT
+  DDL_DROP_PARTITION_STATEMENT,
+  DDL_DEBUG_STATEMENT
 };
 
 /** @brief An abstract base for TableDef, ColumnDef, ...
@@ -1319,8 +1320,10 @@ struct AlterTableStatement : public SqlStatement
 
   QualifiedName* fTableName;
   AlterTableActionList fActions;
+
  private:
   long fTimeZone;
+
  public:
 };
 
@@ -1441,6 +1444,30 @@ struct DropTableStatement : public SqlStatement
   bool fCascade;
 };
 
+/** @brief DebugStatement
+ */
+struct DebugDDLStatement : public SqlStatement
+{
+  /** @brief Deserialize from ByteStream */
+  EXPORT virtual int unserialize(messageqcpp::ByteStream& bs);
+
+  /** @brief Serialize to ByteStream */
+  EXPORT virtual int serialize(messageqcpp::ByteStream& bs);
+
+  DebugDDLStatement(uint32_t debugLevel);
+
+  EXPORT DebugDDLStatement();
+
+  /** @brief Dump to stdout. */
+  EXPORT std::ostream& put(std::ostream& os) const;
+
+  virtual ~DebugDDLStatement()
+  {
+  }
+
+  uint32_t fDebugLevel;
+};
+
 /** @brief TruncTableStatement represents the drop table operation
  */
 struct TruncTableStatement : public SqlStatement

dbcon/ddlpackage/debugstatement.cpp

@@ -0,0 +1,40 @@
+/* Copyright (C) 2024 MariaDB Corporation
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License
+   as published by the Free Software Foundation; version 2 of
+   the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   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 Street, Fifth Floor, Boston,
+   MA 02110-1301, USA. */
+
+#define DDLPKG_DLLEXPORT
+#include "ddlpkg.h"
+#undef DDLPKG_DLLEXPORT
+
+using namespace std;
+
+namespace ddlpackage
+{
+DebugDDLStatement::DebugDDLStatement(uint32_t debugLevel)
+: fDebugLevel(debugLevel)
+{
+}
+
+DebugDDLStatement::DebugDDLStatement()
+: fDebugLevel(0)
+{
+}
+
+ostream& DebugDDLStatement::put(ostream& os) const
+{
+  os << "DDL debug level: " << fDebugLevel << endl;
+  return os;
+}
+}  // namespace ddlpackage

dbcon/ddlpackage/serialize.cpp

@@ -429,6 +429,21 @@ int DropTableStatement::serialize(ByteStream& bytestream)
   return ret;
 }
 
+/** @brief Construct from Bytestream */
+int DebugDDLStatement::unserialize(ByteStream& bytestream)
+{
+  bytestream >> fDebugLevel;
+  return 1;
+}
+
+/** @brief Serialize to ByteStream */
+int DebugDDLStatement::serialize(ByteStream& bytestream)
+{
+  bytestream << (quadbyte)DDL_DEBUG_STATEMENT;
+  bytestream << fDebugLevel;
+  return 1;
+}
+
 ///////////////////////////////////////
 /// TruncTableStatement Serialization
 ///////////////////////////////////////

dbcon/ddlpackageproc/droppartitionprocessor.cpp

@@ -122,6 +122,7 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
   auto stmt = formatStatementString(dropPartitionStmt->fSql, dropPartitionStmt->fTableName->fSchema,
                                     dropPartitionStmt->fTableName->fName, dropPartitionStmt->fPartitions);
   SQLLogger logger(stmt, fDDLLoggingId, sessionID, txnID.id);
+  ostringstream debugLog;
 
   try
   {
@@ -226,6 +227,8 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
         return result;
       }
     }
+    debugLog << "DropPartitionProcessor: got table lock (id) " << uniqueID << " for table (OID)"
+             << roPair.objnum << endl;
 
     // 1. Get the OIDs for the columns
     // 2. Get the OIDs for the dictionaries
@@ -241,9 +244,18 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
 
     tableColRidList = systemCatalogPtr->columnRIDs(userTableName);
     tableAuxColOid = systemCatalogPtr->tableAUXColumnOID(userTableName);
-
     dictOIDList = systemCatalogPtr->dictOIDs(userTableName);
 
+    debugLog << "DropPartitionProcessor:" << endl;
+    debugLog << "column RIDS: ";
+    for (const auto& rid : tableColRidList)
+      debugLog << rid.objnum << " ";
+    debugLog << endl;
+    debugLog << "dict OIDS: ";
+    for (const auto& dictOid : dictOIDList)
+      debugLog << dictOid.dictOID << " ";
+    debugLog << endl;
+
     // Save qualified tablename, all column, dictionary OIDs, and transaction ID into a file in ASCII format
     for (unsigned i = 0; i < tableColRidList.size(); i++)
     {
@@ -271,6 +283,12 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
     {
       throw std::runtime_error(emsg);
     }
+    debugLog << "DropPartitionProcessor: marked partitions for deletion:" << endl;
+    for (const auto& partition : dropPartitionStmt->fPartitions)
+      debugLog << "dbroot: " << partition.dbroot << " physical parition: " << partition.pp
+               << " segment: " << partition.seg << endl;
+    VERBOSE_INFO(debugLog.str());
+    debugLog.clear();
 
     set<BRM::LogicalPartition> markedPartitions;
     set<BRM::LogicalPartition> outOfServicePartitions;
@@ -298,14 +316,16 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
     // Save the oids to a file
     createWritePartitionLogFile(roPair.objnum, markedPartitions, oidList, uniqueId);
 
-    VERBOSE_INFO("Removing files");
+    VERBOSE_INFO("Removing parition files");
     removePartitionFiles(oidList, markedPartitions, uniqueId);
     // Flush PrimProc cache for those lbids
+    VERBOSE_INFO("Flushing paritions");
     rc = cacheutils::flushPartition(oidList, markedPartitions);
 
     // Remove the partition from extent map
     emsg.clear();
     rc = fDbrm->deletePartition(oidList, dropPartitionStmt->fPartitions, emsg);
+    VERBOSE_INFO("DropPartitionProcessor: partitions removed");
 
     if (rc != 0)
       throw std::runtime_error(emsg);
@@ -390,6 +410,7 @@ DropPartitionProcessor::DDLResult DropPartitionProcessor::processPackageInternal
   }
 
   fSessionManager.committed(txnID);
+  VERBOSE_INFO("DropPartitionProcessor:: commited");
   return result;
 }
 

dbcon/mysql/ha_mcs_client_udfs.cpp

@@ -38,6 +38,9 @@ using namespace oam;
 #include "errorids.h"
 using namespace logging;
 
+#include "ddlpkg.h"
+#include "calpontsystemcatalog.h"
+
 // #include "resourcemanager.h"
 
 #include "columnstoreversion.h"
@@ -180,7 +183,6 @@ extern "C"
     return result;
   }
 
-
   my_bool mcssetparms_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
   {
     return setparms_init(initid, args, message, "MCSSETPARMS");
@@ -190,7 +192,6 @@ extern "C"
   {
   }
 
-
   const char* calsetparms(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
                           char* is_null, char* error)
   {
@@ -247,7 +248,6 @@ extern "C"
     return result;
   }
 
-
   my_bool mcsgetstats_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
   {
     return getstats_init(initid, args, message, "MCSGETSTATS");
@@ -555,6 +555,80 @@ extern "C"
     return 0;
   }
 
+  my_bool mcs_set_ddldebug_level_init(UDF_INIT* initid, UDF_ARGS* args, char* message, const char* funcname)
+  {
+    if ((args->arg_count != 1) || (args->arg_type[0] != INT_RESULT) ||
+        *reinterpret_cast<uint32_t*>(args->args[0]) > 3)
+    {
+      sprintf(message, "%s() requires one integer `debug level` ", funcname);
+      return 1;
+    }
+
+    initid->maybe_null = 0;
+    initid->max_length = 5;
+
+    return 0;
+  }
+
+  const char* mcs_set_ddldebug_level(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length)
+  {
+    uint32_t level = *reinterpret_cast<uint32_t*>(args->args[0]);
+    std::unique_ptr<ddlpackage::DebugDDLStatement> stmt =
+        std::make_unique<ddlpackage::DebugDDLStatement>(level);
+    stmt->fSessionID = execplan::CalpontSystemCatalog::idb_tid2sid(current_thd->thread_id);
+    MessageQueueClient mq("DDLProc");
+
+    ByteStream bytestream;
+    bytestream << (uint32_t)stmt->fSessionID;
+    stmt->serialize(bytestream);
+    ByteStream::byte b = 0;
+    THD* thd = current_thd;
+    string emsg;
+    mq.write(bytestream);
+
+    try
+    {
+      bytestream = mq.read();
+
+      if (bytestream.length() == 0)
+      {
+        thd->get_stmt_da()->set_overwrite_status(true);
+        thd->raise_error_printf(ER_INTERNAL_ERROR, "Lost connection to DDLProc");
+        std::memcpy(result, "Error", 5);
+        *length = 5;
+        return result;
+      }
+      else
+      {
+        bytestream >> b;
+        bytestream >> emsg;
+      }
+    }
+    catch (runtime_error&)
+    {
+      thd->get_stmt_da()->set_overwrite_status(true);
+      thd->raise_error_printf(ER_INTERNAL_ERROR, "Lost connection to DDLProc");
+      std::memcpy(result, "Error", 5);
+      *length = 5;
+      return result;
+    }
+    catch (...)
+    {
+      thd->get_stmt_da()->set_overwrite_status(true);
+      thd->raise_error_printf(ER_INTERNAL_ERROR, "Unknown error caught");
+      std::memcpy(result, "Error", 5);
+      *length = 5;
+      return result;
+    }
+
+    *length = 2;
+    std::memcpy(result, "Ok", 2);
+    return result;
+  }
+
+  void mcs_set_ddldebug_level_deinit(UDF_INIT* initid)
+  {
+  }
 
   my_bool mcscleartablelock_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
   {
@@ -636,7 +710,6 @@ extern "C"
     return 0;
   }
 
-
   my_bool mcslastinsertid_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
   {
     return lastinsertid_init(initid, args, message, "MCSLASTINSERTID");
@@ -761,9 +834,9 @@ extern "C"
 
   static const unsigned long TraceSize = 16 * 1024;
 
-// mysqld will call this with only 766 bytes available in result no matter what we asked for in
-// calgettrace_init()
-// if we return a pointer that is not result, mysqld will take our pointer and use it, freeing up result
+  // mysqld will call this with only 766 bytes available in result no matter what we asked for in
+  // calgettrace_init()
+  // if we return a pointer that is not result, mysqld will take our pointer and use it, freeing up result
   const char* mcsgettrace(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
                           char* is_null, char* error)
   {

dbcon/mysql/ha_mcs_partition.cpp

@@ -979,8 +979,8 @@ extern "C"
     return 0;
   }
 
-      const char* caldroppartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result,
-                                           unsigned long* length, char* is_null, char* error)
+  const char* caldroppartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
+                                       char* is_null, char* error)
   {
     string msg;
     CalpontSystemCatalog::TableName tableName;
@@ -1167,8 +1167,8 @@ extern "C"
     delete[] initid->ptr;
   }
 
-      const char* calshowpartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result,
-                                           unsigned long* length, char* is_null, char* error)
+  const char* calshowpartitionsbyvalue(UDF_INIT* initid, UDF_ARGS* args, char* result, unsigned long* length,
+                                       char* is_null, char* error)
   {
     BRM::DBRM::refreshShm();
     DBRM em;

dbcon/mysql/install_mcs_mysql.sh.in

@@ -79,6 +79,7 @@ CREATE OR REPLACE FUNCTION caldisablepartitionsbyvalue RETURNS STRING SONAME 'ha
 CREATE OR REPLACE FUNCTION calenablepartitionsbyvalue RETURNS STRING SONAME 'ha_columnstore.so';
 CREATE OR REPLACE FUNCTION calshowpartitionsbyvalue RETURNS STRING SONAME 'ha_columnstore.so';
 CREATE OR REPLACE AGGREGATE FUNCTION moda RETURNS STRING SONAME 'libregr_mysql.so';
+CREATE OR REPLACE FUNCTION mcs_set_ddldebug_level RETURNS STRING SONAME 'ha_columnstore.so';
 
 CREATE DATABASE IF NOT EXISTS infinidb_querystats;
 CREATE TABLE IF NOT EXISTS infinidb_querystats.querystats

ddlproc/ddlprocessor.cpp

@@ -75,8 +75,9 @@ void cleanPMSysCache()
 class PackageHandler
 {
  public:
-  PackageHandler(QueryTeleClient qtc, DBRM* dbrm, messageqcpp::IOSocket& ios, bool concurrentSupport)
-   : fIos(ios), fDbrm(dbrm), fQtc(qtc), fConcurrentSupport(concurrentSupport)
+  PackageHandler(QueryTeleClient qtc, DBRM* dbrm, messageqcpp::IOSocket& ios, bool concurrentSupport,
+                 uint32_t* debugLevel)
+   : fIos(ios), fDbrm(dbrm), fQtc(qtc), fConcurrentSupport(concurrentSupport), fDebugLevel(debugLevel)
   {
   }
 
@@ -490,6 +491,8 @@ class PackageHandler
           boost::scoped_ptr<CreateTableProcessor> processor(new CreateTableProcessor(fDbrm));
           processor->fTxnid.id = fTxnid.id;
           processor->fTxnid.valid = true;
+          if (fDebugLevel)
+            processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
           // cout << "create table using txnid " << fTxnid.id << endl;
 
           QueryTeleStats qts;
@@ -524,6 +527,8 @@ class PackageHandler
           boost::scoped_ptr<AlterTableProcessor> processor(new AlterTableProcessor(fDbrm));
           processor->fTxnid.id = fTxnid.id;
           processor->fTxnid.valid = true;
+          if (fDebugLevel)
+            processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
 
           QueryTeleStats qts;
           qts.query_uuid = QueryTeleClient::genUUID();
@@ -560,6 +565,8 @@ class PackageHandler
 
           processor->fTxnid.id = fTxnid.id;
           processor->fTxnid.valid = true;
+          if (fDebugLevel)
+            processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
 
           QueryTeleStats qts;
           qts.query_uuid = QueryTeleClient::genUUID();
@@ -595,6 +602,8 @@ class PackageHandler
 
           processor->fTxnid.id = fTxnid.id;
           processor->fTxnid.valid = true;
+          if (fDebugLevel)
+            processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
 
           QueryTeleStats qts;
           qts.query_uuid = QueryTeleClient::genUUID();
@@ -628,6 +637,9 @@ class PackageHandler
           boost::scoped_ptr<MarkPartitionProcessor> processor(new MarkPartitionProcessor(fDbrm));
           (processor->fTxnid).id = fTxnid.id;
           (processor->fTxnid).valid = true;
+          if (fDebugLevel)
+            processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
+
           result = processor->processPackage(&markPartitionStmt);
           systemCatalogPtr->removeCalpontSystemCatalog(markPartitionStmt.fSessionID);
           systemCatalogPtr->removeCalpontSystemCatalog(markPartitionStmt.fSessionID | 0x80000000);
@@ -643,6 +655,9 @@ class PackageHandler
           boost::scoped_ptr<RestorePartitionProcessor> processor(new RestorePartitionProcessor(fDbrm));
           (processor->fTxnid).id = fTxnid.id;
           (processor->fTxnid).valid = true;
+          if (fDebugLevel)
+            processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
+
           result = processor->processPackage(&restorePartitionStmt);
           systemCatalogPtr->removeCalpontSystemCatalog(restorePartitionStmt.fSessionID);
           systemCatalogPtr->removeCalpontSystemCatalog(restorePartitionStmt.fSessionID | 0x80000000);
@@ -658,12 +673,24 @@ class PackageHandler
           boost::scoped_ptr<DropPartitionProcessor> processor(new DropPartitionProcessor(fDbrm));
           (processor->fTxnid).id = fTxnid.id;
           (processor->fTxnid).valid = true;
+          if (fDebugLevel)
+            processor->setDebugLevel(static_cast<DDLPackageProcessor::DebugLevel>(*fDebugLevel));
+
           result = processor->processPackage(&dropPartitionStmt);
           systemCatalogPtr->removeCalpontSystemCatalog(dropPartitionStmt.fSessionID);
           systemCatalogPtr->removeCalpontSystemCatalog(dropPartitionStmt.fSessionID | 0x80000000);
         }
         break;
 
+        case ddlpackage::DDL_DEBUG_STATEMENT:
+        {
+          DebugDDLStatement stmt;
+          stmt.unserialize(fByteStream);
+          if (fDebugLevel)
+            *fDebugLevel = stmt.fDebugLevel;
+        }
+        break;
+
         default: throw UNRECOGNIZED_PACKAGE_TYPE; break;
       }
 
@@ -728,6 +755,7 @@ class PackageHandler
   QueryTeleClient fQtc;
   oam::OamCache* fOamCache = nullptr;
   bool fConcurrentSupport;
+  uint32_t* fDebugLevel{nullptr};
 };
 
 }  // namespace
@@ -782,7 +810,7 @@ void DDLProcessor::process()
       try
       {
         ios = fMqServer.accept();
-        fDdlPackagepool.invoke(PackageHandler(fQtc, &dbrm, ios, concurrentSupport));
+        fDdlPackagepool.invoke(PackageHandler(fQtc, &dbrm, ios, concurrentSupport, &debugLevel));
       }
       catch (...)
       {

ddlproc/ddlprocessor.h

@@ -93,6 +93,7 @@ class DDLProcessor
   WriteEngine::WEClients* fWEClient;
   uint32_t fPMCount;
   querytele::QueryTeleClient fQtc;
+  uint32_t debugLevel{0};
 };
 
 }  // namespace ddlprocessor