feat: add vacuum_partition functionality with initialization and execution logic

Author: amrdb

dbcon/mysql/ha_mcs_client_udfs.cpp

@@ -1230,6 +1230,67 @@ extern "C"
   {
   }
 
+  my_bool vacuum_partition_init(UDF_INIT* initid, UDF_ARGS* args, char* message, const char* funcname)
+  {
+    if (args->arg_count != 3 || 
+        args->arg_type[0] != STRING_RESULT || 
+        args->arg_type[1] != STRING_RESULT || 
+        args->arg_type[2] != STRING_RESULT)
+    {
+      sprintf(message, "%s() requires three string arguments", funcname);
+      return 1;
+    }
+
+    initid->maybe_null = 0;
+    initid->max_length = 255;
+
+    return 0;
+  }
+
+  my_bool mcs_vacuum_partition_init(UDF_INIT* initid, UDF_ARGS* args, char* message)
+  {
+    return vacuum_partition_init(initid, args, message, "MCSVACUUMPARTITION");
+  }
+
+  const char* mcs_vacuum_partition(UDF_INIT* /*initid*/, UDF_ARGS* args, char* result, 
+                                   unsigned long* length, char* /*is_null*/, char* /*error*/)
+  {
+    THD* thd = current_thd;
+
+    if (get_fe_conn_info_ptr() == NULL)
+    {
+      set_fe_conn_info_ptr((void*)new cal_connection_info());
+      thd_set_ha_data(thd, mcs_hton, get_fe_conn_info_ptr());
+    }
+
+    cal_connection_info* ci = reinterpret_cast<cal_connection_info*>(get_fe_conn_info_ptr());
+    execplan::CalpontSystemCatalog::TableName tableName;
+
+    tableName.schema = args->args[0];
+    tableName.table = args->args[1];
+    std::string partition = args->args[2];
+
+    if (lower_case_table_names) {
+        boost::algorithm::to_lower(tableName.schema);
+        boost::algorithm::to_lower(tableName.table);
+    }
+
+    if (!ci->dmlProc)
+    {
+      ci->dmlProc = new MessageQueueClient("DMLProc");
+    }
+
+    std::string vacuumResult = ha_mcs_impl_vacuum_partition(*ci, tableName, partition);
+
+    memcpy(result, vacuumResult.c_str(), vacuumResult.length());
+    *length = vacuumResult.length();
+    return result;
+  }
+
+  void mcs_vacuum_partition_deinit(UDF_INIT* /*initid*/)
+  {
+  }
+
   my_bool analyze_table_bloat_init(UDF_INIT* initid, UDF_ARGS* args, char* message, const char* funcname)
   {
     if (args->arg_count != 2 || 

dbcon/mysql/ha_mcs_dml.cpp

@@ -28,6 +28,7 @@
 #include <unordered.h>
 #include <fstream>
 #include <sstream>
+#include <vector>
 #include <cerrno>
 #include <cstring>
 
@@ -1077,6 +1078,158 @@ std::string ha_mcs_impl_analyze_partition_bloat(cal_impl_if::cal_connection_info
   return analysisResult;
 }
 
+std::string ha_mcs_impl_vacuum_partition(cal_impl_if::cal_connection_info& ci,
+                                         execplan::CalpontSystemCatalog::TableName& tablename,
+                                         const std::string& partition)
+{
+  THD* thd = current_thd;
+  ulong sessionID = tid2sid(thd->thread_id);
+  std::string result;
+
+  try
+  {
+    // Parse partition triplet string: "dbroot.segment.partition"
+    uint16_t dbRoot;
+    uint16_t segmentNum;
+    uint32_t partitionNum;
+    
+    std::istringstream ss(partition);
+    std::string item;
+    std::vector<std::string> tokens;
+    
+    while (std::getline(ss, item, '.')) {
+      tokens.push_back(item);
+    }
+    
+    if (tokens.size() != 3) {
+      return "Error: Invalid partition triplet format. Expected: dbroot.segment.partition";
+    }
+    
+    try
+    {
+      dbRoot = static_cast<uint16_t>(std::stoul(tokens[0]));
+      segmentNum = static_cast<uint16_t>(std::stoul(tokens[1]));
+      partitionNum = std::stoul(tokens[2]);
+    }
+    catch (const std::exception&)
+    {
+      return "Error: Invalid numeric values in partition triplet";
+    }
+
+    // Get system catalog
+    boost::shared_ptr<execplan::CalpontSystemCatalog> systemCatalogPtr =
+        execplan::CalpontSystemCatalog::makeCalpontSystemCatalog(sessionID);
+    systemCatalogPtr->identity(execplan::CalpontSystemCatalog::EC);
+
+    // Get table information
+    execplan::CalpontSystemCatalog::TableName sysCatalogTableName;
+    sysCatalogTableName.schema = tablename.schema;
+    sysCatalogTableName.table = tablename.table;
+
+    // Check if table exists
+    execplan::CalpontSystemCatalog::RIDList ridList;
+    try
+    {
+      ridList = systemCatalogPtr->columnRIDs(sysCatalogTableName);
+    }
+    catch (const std::exception& ex)
+    {
+      return std::string("Error: Table not found - ") + ex.what();
+    }
+
+    if (ridList.empty())
+    {
+      return "Error: Table has no columns";
+    }
+
+    // Get AUX column OID
+    execplan::CalpontSystemCatalog::OID auxColumnOid = systemCatalogPtr->tableAUXColumnOID(sysCatalogTableName);
+
+    // Build column list for createHiddenStripeColumnExtents
+    std::vector<BRM::CreateStripeColumnExtentsArgIn> cols;
+
+    // Add regular columns
+    for (const auto& roPair : ridList)
+    {
+      BRM::CreateStripeColumnExtentsArgIn colArg;
+      colArg.oid = roPair.objnum;
+      
+      execplan::CalpontSystemCatalog::ColType colType = systemCatalogPtr->colType(roPair.objnum);
+      colArg.colDataType = colType.colDataType;
+      
+      // Set width based on whether it's a dictionary column
+      if (colType.ddn.dictOID > 3000)
+      {
+        colArg.width = 8; // Dictionary columns use 8-byte tokens
+      }
+      else
+      {
+        colArg.width = colType.colWidth;
+      }
+      
+      cols.push_back(colArg);
+    }
+
+    // Add AUX column if it exists
+    if (auxColumnOid > 3000)
+    {
+      BRM::CreateStripeColumnExtentsArgIn auxColArg;
+      auxColArg.oid = auxColumnOid;
+      execplan::CalpontSystemCatalog::ColType auxColType = systemCatalogPtr->colType(auxColumnOid);
+      auxColArg.colDataType = auxColType.colDataType;
+      auxColArg.width = auxColType.colWidth;
+      cols.push_back(auxColArg);
+    }
+
+    std::vector<BRM::CreateStripeColumnExtentsArgOut> extents;
+
+    // Store original values for comparison
+    uint16_t originalSegmentNum = segmentNum;
+    uint32_t originalPartitionNum = partitionNum;
+    
+    // Call createHiddenStripeColumnExtents
+    BRM::DBRM dbrm;
+    int rc = dbrm.createHiddenStripeColumnExtents(cols, dbRoot, partitionNum, segmentNum, extents);
+
+    if (rc != 0)
+    {
+      std::ostringstream oss;
+      oss << "Error: Failed to create hidden stripe column extents, error code: " << rc;
+      return oss.str();
+    }
+
+    // Build success message
+    std::ostringstream successMsg;
+    successMsg << "Successfully created hidden partition on DBRoot " << dbRoot;
+    
+    // Check if the function used our requested values or assigned different ones
+    if (partitionNum != originalPartitionNum || segmentNum != originalSegmentNum)
+    {
+      successMsg << " (requested: " << originalPartitionNum << "." << originalSegmentNum 
+                 << ", assigned: " << partitionNum << "." << segmentNum << ")";
+    }
+    else
+    {
+      successMsg << " at partition " << partitionNum << ", segment " << segmentNum;
+    }
+    
+    successMsg << " with " << extents.size() << " extents";
+
+    result = successMsg.str();
+  }
+  catch (const std::exception& ex)
+  {
+    std::ostringstream errorMsg;
+    errorMsg << "Error: " << ex.what();
+    result = errorMsg.str();
+  }
+  catch (...)
+  {
+    result = "Error: Unknown exception occurred during vacuum partition operation";
+  }
+
+  return result;
+}
 
 std::string ha_mcs_impl_analyze_table_bloat(cal_impl_if::cal_connection_info& ci,
                                                execplan::CalpontSystemCatalog::TableName& tablename)

dbcon/mysql/ha_mcs_impl.h

@@ -80,4 +80,7 @@ extern std::string ha_mcs_impl_analyze_partition_bloat(cal_impl_if::cal_connecti
                                                         const std::string& partition);
 extern std::string ha_mcs_impl_analyze_table_bloat(cal_impl_if::cal_connection_info& ci,
                                                     execplan::CalpontSystemCatalog::TableName& tablename);
+extern std::string ha_mcs_impl_vacuum_partition(cal_impl_if::cal_connection_info& ci,
+                                                 execplan::CalpontSystemCatalog::TableName& tablename,
+                                                 const std::string& partition);
 #endif

dbcon/mysql/install_mcs_mysql.sh.in

@@ -58,6 +58,7 @@ CREATE OR REPLACE FUNCTION mcs_emindex_size RETURNS INTEGER SONAME 'ha_columnsto
 CREATE OR REPLACE FUNCTION mcs_emindex_free RETURNS INTEGER SONAME 'ha_columnstore.so';
 CREATE OR REPLACE FUNCTION mcs_analyze_partition_bloat RETURNS STRING SONAME 'ha_columnstore.so';
 CREATE OR REPLACE FUNCTION mcs_analyze_table_bloat RETURNS STRING SONAME 'ha_columnstore.so';
+CREATE OR REPLACE FUNCTION mcs_vacuum_partition RETURNS STRING SONAME 'ha_columnstore.so';
 CREATE OR REPLACE FUNCTION columnstore_dataload RETURNS STRING SONAME 'ha_columnstore.so';
 CREATE OR REPLACE AGGREGATE FUNCTION regr_avgx RETURNS REAL SONAME 'libregr_mysql.so';
 CREATE OR REPLACE AGGREGATE FUNCTION regr_avgy RETURNS REAL SONAME 'libregr_mysql.so';

utils/libmarias3/libmarias3

@@ -481,6 +481,8 @@ const uint8_t MARK_ALL_PARTITION_FOR_DELETION = 41;
 const uint8_t CREATE_COLUMN_EXTENT_EXACT_FILE = 42;
 const uint8_t DELETE_DBROOT = 43;
 const uint8_t CREATE_STRIPE_COLUMN_EXTENTS = 44;
+const uint8_t CREATE_HIDDEN_STRIPE_COLUMN_EXTENTS = 107;
+const uint8_t MAKE_PARTITION_VISIBLE = 108;
 
 /* SessionManager interface */
 const uint8_t VER_ID = 45;

versioning/BRM/brmtypes.h

@@ -991,6 +991,114 @@ int DBRM::createStripeColumnExtents(const std::vector<CreateStripeColumnExtentsA
   return 0;
 }
 
+//------------------------------------------------------------------------------
+// Send a request to create hidden stripe column extents
+//------------------------------------------------------------------------------
+int DBRM::createHiddenStripeColumnExtents(const std::vector<CreateStripeColumnExtentsArgIn>& cols, uint16_t dbRoot,
+                                          uint32_t& partitionNum, uint16_t& segmentNum,
+                                          std::vector<CreateStripeColumnExtentsArgOut>& extents) DBRM_THROW
+{
+#ifdef BRM_INFO
+
+  if (fDebug)
+  {
+    TRACER_WRITELATER("createHiddenStripeColumnExtents");
+    TRACER_WRITE;
+  }
+
+#endif
+
+  ByteStream command, response;
+  uint8_t err;
+  uint16_t tmp16;
+  uint32_t tmp32;
+
+  command << CREATE_HIDDEN_STRIPE_COLUMN_EXTENTS;
+  serializeInlineVector(command, cols);
+  command << dbRoot << partitionNum;
+
+  err = send_recv(command, response);
+
+  if (err != ERR_OK)
+    return err;
+
+  if (response.length() == 0)
+    return ERR_NETWORK;
+
+  try
+  {
+    response >> err;
+
+    if (err != 0)
+      return (int)err;
+
+    response >> tmp32;
+    partitionNum = tmp32;
+    response >> tmp16;
+    segmentNum = tmp16;
+    deserializeInlineVector(response, extents);
+  }
+  catch (exception& e)
+  {
+    cerr << e.what() << endl;
+    return ERR_FAILURE;
+  }
+
+  CHECK_EMPTY(response);
+  return 0;
+}
+
+//------------------------------------------------------------------------------
+// Send a request to make a hidden partition visible
+//------------------------------------------------------------------------------
+int DBRM::makePartitionVisible(const std::set<OID_t>& oids, uint16_t dbRoot, uint32_t partitionNum) DBRM_THROW
+{
+#ifdef BRM_INFO
+
+  if (fDebug)
+  {
+    TRACER_WRITELATER("makePartitionVisible");
+    TRACER_WRITE;
+  }
+
+#endif
+
+  ByteStream command, response;
+  uint8_t err;
+
+  command << MAKE_PARTITION_VISIBLE;
+  command << static_cast<uint32_t>(oids.size());
+  for (const auto& oid : oids)
+  {
+    command << oid;
+  }
+  command << dbRoot << partitionNum;
+
+  err = send_recv(command, response);
+
+  if (err != ERR_OK)
+    return err;
+
+  if (response.length() == 0)
+    return ERR_NETWORK;
+
+  try
+  {
+    response >> err;
+
+    if (err != 0)
+      return (int)err;
+  }
+  catch (exception& e)
+  {
+    cerr << e.what() << endl;
+    return ERR_FAILURE;
+  }
+
+  CHECK_EMPTY(response);
+  return 0;
+}
+
 //------------------------------------------------------------------------------
 // Send a request to create a column extent for the specified OID and DBRoot.
 //------------------------------------------------------------------------------