�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fbig%2F'>big</a>/<a href='?path=%2Fbig%2FSRC%2F'>SRC</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2F'>mariadb-11.7.2</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2F'>storage</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2F'>columnstore</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2Fcolumnstore%2F'>columnstore</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2Fcolumnstore%2Fddlproc%2F'>ddlproc</a>/</p><hr><h3>✏️ Editing: ddlprocessor.cpp</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'>/* Copyright (C) 2014 InfiniDB, Inc.

   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. */

/***********************************************************************
 *   $Id: ddlprocessor.cpp 6 2006-06-23 17:58:51Z rcraighead $
 *
 *
 ***********************************************************************/
#include &lt;map&gt;
#include &lt;boost/scoped_ptr.hpp&gt;
using namespace std;

#include &quot;ddlpkg.h&quot;
#include &quot;ddlprocessor.h&quot;
#include &quot;createtableprocessor.h&quot;
#include &quot;altertableprocessor.h&quot;
#include &quot;droptableprocessor.h&quot;
#include &quot;calpontsystemcatalog.h&quot;
#include &quot;sqlparser.h&quot;
#include &quot;configcpp.h&quot;
#include &quot;markpartitionprocessor.h&quot;
#include &quot;restorepartitionprocessor.h&quot;
#include &quot;droppartitionprocessor.h&quot;
//#define 	SERIALIZE_DDL_DML_CPIMPORT 	1

#include &quot;cacheutils.h&quot;
#include &quot;vss.h&quot;
#include &quot;dbrm.h&quot;
#include &quot;idberrorinfo.h&quot;
#include &quot;errorids.h&quot;
#include &quot;we_messages.h&quot;
using namespace BRM;

using namespace config;
using namespace messageqcpp;
using namespace ddlpackageprocessor;
using namespace ddlpackage;
using namespace execplan;
using namespace logging;
using namespace WriteEngine;

#include &quot;querytele.h&quot;
using namespace querytele;

#include &quot;oamcache.h&quot;

namespace
{
typedef messageqcpp::ByteStream::quadbyte quadbyte;

const quadbyte UNRECOGNIZED_PACKAGE_TYPE = 100;

const std::string DDLProcName = &quot;DDLProc&quot;;

void cleanPMSysCache()
{
  vector&lt;BRM::OID_t&gt; oidList = getAllSysCatOIDs();
  cacheutils::flushOIDsFromCache(oidList);
}

class PackageHandler
{
 public:
  PackageHandler(QueryTeleClient qtc, DBRM* dbrm, messageqcpp::IOSocket&amp; ios, bool concurrentSupport,
                 uint32_t* debugLevel)
   : fIos(ios), fDbrm(dbrm), fQtc(qtc), fConcurrentSupport(concurrentSupport), fDebugLevel(debugLevel)
  {
  }

  void operator()()
  {
    try
    {
      fByteStream = fIos.read();
      if (fByteStream.empty())
      {
        logError(DDLPackageProcessor::NOT_ACCEPTING_PACKAGES, &quot;Empty package&quot;, true);
        return;
      }
      fByteStream &gt;&gt; fSessionID;
      fByteStream &gt;&gt; fPackageType;

      uint32_t stateFlags;
      if (fDbrm-&gt;getSystemState(stateFlags) &gt;
          0)  // &gt; 0 implies successful retrieval. It doesn&#039;t imply anything about the contents
      {
        messageqcpp::ByteStream results;
        const char* responseMsg = nullptr;
        messageqcpp::ByteStream::byte status;
        bool bReject = false;

        // Check to see if we&#039;re in write suspended or shutdown mode
        // If so, we can&#039;t process the request.
        if (stateFlags &amp; SessionManagerServer::SS_SUSPENDED ||
            stateFlags &amp; SessionManagerServer::SS_SUSPEND_PENDING ||
            stateFlags &amp; SessionManagerServer::SS_SHUTDOWN_PENDING)
        {
          if (stateFlags &amp; SessionManagerServer::SS_SUSPENDED ||
              stateFlags &amp; SessionManagerServer::SS_SUSPEND_PENDING)
          {
            responseMsg = &quot;Writing to the database is disabled.&quot;;
          }
          else
          {
            responseMsg = &quot;The database is being shut down.&quot;;
          }

          status = DDLPackageProcessor::NOT_ACCEPTING_PACKAGES;
          bReject = true;
        }

        if (bReject)
        {
          results &lt;&lt; status;
          //@bug 266
          MessageLog logger(LoggingID(27));
          logging::Message::Args args;
          logging::Message message(2);
          args.add(responseMsg);
          message.format(args);
          results &lt;&lt; message.msg();
          fIos.write(results);
          std::cout &lt;&lt; responseMsg &lt;&lt; endl;
          std::cout &lt;&lt; &quot;Command rejected. Status &quot; &lt;&lt; (int)status &lt;&lt; message.msg() &lt;&lt; endl;
          return;
        }
      }

      // check whether the system is ready to process statement.
      if (fDbrm-&gt;getSystemReady() &lt; 1)
      {
        logError(DDLPackageProcessor::NOT_ACCEPTING_PACKAGES, &quot;System is not ready yet. Please try again.&quot;,
                 true);
        return;
      }

      int rc = 0;

      if (!fConcurrentSupport)
      {
        // Check if any other active transaction
        bool bIsDbrmUp = true;
        bool anyOtherActiveTransaction = true;
        execplan::SessionManager sessionManager;
        BRM::SIDTIDEntry blockingsid;
        int i = 0;
        int waitPeriod = 10;
        //@Bug 2487 Check transaction map every 1/10 second

        int sleepTime = 100;  // sleep 100 milliseconds between checks
        int numTries = 10;    // try 10 times per second

        string waitPeriodStr = config::Config::makeConfig()-&gt;getConfig(&quot;SystemConfig&quot;, &quot;WaitPeriod&quot;);

        if (waitPeriodStr.length() != 0)
          waitPeriod = static_cast&lt;int&gt;(config::Config::fromText(waitPeriodStr));

        numTries = waitPeriod * 10;
        struct timespec rm_ts;

        rm_ts.tv_sec = sleepTime / 1000;
        rm_ts.tv_nsec = sleepTime % 1000 * 1000000;

        // cout &lt;&lt; &quot;starting i = &quot; &lt;&lt; i &lt;&lt; endl;
        // anyOtherActiveTransaction = sessionManager.checkActiveTransaction(fSessionID, bIsDbrmUp);
        while (anyOtherActiveTransaction)
        {
          anyOtherActiveTransaction =
              sessionManager.checkActiveTransaction(fSessionID, bIsDbrmUp, blockingsid);

          if (anyOtherActiveTransaction)
          {
            for (; i &lt; numTries; i++)
            {
              struct timespec abs_ts;

              // cout &lt;&lt; &quot;session &quot; &lt;&lt; fSessionID &lt;&lt; &quot; nanosleep on package type &quot; &lt;&lt; (int)packageType &lt;&lt;
              // endl;
              do
              {
                abs_ts.tv_sec = rm_ts.tv_sec;
                abs_ts.tv_nsec = rm_ts.tv_nsec;
              } while (nanosleep(&amp;abs_ts, &amp;rm_ts) &lt; 0);

              anyOtherActiveTransaction =
                  sessionManager.checkActiveTransaction(fSessionID, bIsDbrmUp, blockingsid);

              if (!anyOtherActiveTransaction)
              {
                // cout &lt;&lt; &quot;Ready to process type &quot; &lt;&lt; (int)packageType &lt;&lt; endl;
                fTxnid = sessionManager.getTxnID(fSessionID);

                if (!fTxnid.valid)
                {
                  fTxnid = sessionManager.newTxnID(fSessionID, true, true);

                  if (fTxnid.valid)
                  {
                    // cout &lt;&lt; &quot;Ready to process type &quot; &lt;&lt; (int)packageType &lt;&lt; &quot; with txnid &quot; &lt;&lt; txnid.id &lt;&lt;
                    // endl;
                    anyOtherActiveTransaction = false;
                    break;
                  }
                  else
                  {
                    anyOtherActiveTransaction = true;
                  }
                }
                else
                {
                  string errorMsg;
                  rc = commitTransaction(fTxnid.id, errorMsg);

                  if (rc != 0)
                    throw std::runtime_error(errorMsg);

                  // need unlock the table.
                  std::vector&lt;TableLockInfo&gt; tableLocks = fDbrm-&gt;getAllTableLocks();
                  bool lockReleased = true;

                  for (unsigned k = 0; k &lt; tableLocks.size(); k++)
                  {
                    if (tableLocks[k].ownerTxnID == fTxnid.id)
                    {
                      lockReleased = fDbrm-&gt;releaseTableLock(tableLocks[k].id);

                      if (!lockReleased)
                      {
                        ostringstream os;
                        os &lt;&lt; &quot;tablelock id &quot; &lt;&lt; tableLocks[k].id &lt;&lt; &quot; is not found&quot;;
                        throw std::runtime_error(os.str());
                      }
                    }
                  }

                  fDbrm-&gt;committed(fTxnid);
                  fTxnid = fDbrm-&gt;newTxnID(fSessionID, true, true);

                  if (fTxnid.valid)
                  {
                    // cout &lt;&lt; &quot;Ready to process type &quot; &lt;&lt; (int)packageType &lt;&lt; &quot; with txnid &quot; &lt;&lt; txnid.id &lt;&lt;
                    // endl;
                    anyOtherActiveTransaction = false;
                    break;
                  }
                  else
                  {
                    anyOtherActiveTransaction = true;
                  }
                }
              }
            }

            // cout &lt;&lt; &quot;ending i = &quot; &lt;&lt; i &lt;&lt; endl;
          }
          else
          {
            // cout &lt;&lt; &quot;Ready to process type &quot; &lt;&lt; (int)packageType &lt;&lt; endl;
            fTxnid = sessionManager.getTxnID(fSessionID);

            if (!fTxnid.valid)
            {
              fTxnid = sessionManager.newTxnID(fSessionID, true, true);

              if (!fTxnid.valid)
              {
                // cout &lt;&lt; &quot;cannot get txnid &quot; &lt;&lt; (int)packageType &lt;&lt; &quot; for session &quot; &lt;&lt; sessionID &lt;&lt;  endl;
                anyOtherActiveTransaction = true;
              }
              else
              {
                anyOtherActiveTransaction = false;
              }
            }
            else
            {
              string errorMsg;
              rc = commitTransaction(fTxnid.id, errorMsg);

              if (rc != 0)
                throw std::runtime_error(errorMsg);

              // need unlock the table.
              std::vector&lt;TableLockInfo&gt; tableLocks = fDbrm-&gt;getAllTableLocks();
              bool lockReleased = true;

              for (unsigned k = 0; k &lt; tableLocks.size(); k++)
              {
                if (tableLocks[k].ownerTxnID == fTxnid.id)
                {
                  lockReleased = fDbrm-&gt;releaseTableLock(tableLocks[k].id);

                  if (!lockReleased)
                  {
                    ostringstream os;
                    os &lt;&lt; &quot;tablelock id &quot; &lt;&lt; tableLocks[k].id &lt;&lt; &quot; is not found&quot;;
                    throw std::runtime_error(os.str());
                  }
                }
              }

              sessionManager.committed(fTxnid);
              fTxnid = sessionManager.newTxnID(fSessionID, true, true);

              if (!fTxnid.valid)
              {
                // cout &lt;&lt; &quot;cannot get txnid &quot; &lt;&lt; (int)packageType &lt;&lt; &quot; for session &quot; &lt;&lt; sessionID &lt;&lt;  endl;
                anyOtherActiveTransaction = true;
              }
              else
              {
                anyOtherActiveTransaction = false;
              }
            }
          }

          if ((anyOtherActiveTransaction) &amp;&amp; (i &gt;= numTries))
          {
            // cout &lt;&lt; &quot; Erroring out on package type &quot; &lt;&lt; (int)packageType &lt;&lt; endl;
            break;
          }
        }

        if ((anyOtherActiveTransaction) &amp;&amp; (i &gt;= numTries))
        {
          messageqcpp::ByteStream::byte status = DDLPackageProcessor::NOT_ACCEPTING_PACKAGES;

          Message::Args args;
          args.add(static_cast&lt;uint64_t&gt;(blockingsid.sessionid));

          //@Bug 3854 Log to debug.log
          LoggingID logid(15, 0, 0);
          logging::Message::Args args1;
          logging::Message msg(1);
          args1.add(IDBErrorInfo::instance()-&gt;errorMsg(ERR_ACTIVE_TRANSACTION, args));
          msg.format(args1);
          logging::Logger logger(logid.fSubsysID);
          logger.logMessage(LOG_TYPE_DEBUG, msg, logid);

          logError(status, IDBErrorInfo::instance()-&gt;errorMsg(ERR_ACTIVE_TRANSACTION, args));
        }
        else
        {
          processStatement();
        }
      }
      else
      {
        fTxnid = fDbrm-&gt;getTxnID(fSessionID);

        if (!fTxnid.valid)
        {
          fTxnid = fDbrm-&gt;newTxnID(fSessionID, true, true);

          if (!fTxnid.valid)
          {
            throw std::runtime_error(std::string(&quot;Unable to start a transaction. Check critical log.&quot;));
          }
        }
        else
        {
          string errorMsg;
          rc = commitTransaction(fTxnid.id, errorMsg);

          if (rc != 0)
            throw std::runtime_error(errorMsg);

          // need unlock the table.
          std::vector&lt;TableLockInfo&gt; tableLocks = fDbrm-&gt;getAllTableLocks();
          bool lockReleased = true;

          for (unsigned k = 0; k &lt; tableLocks.size(); k++)
          {
            if (tableLocks[k].ownerTxnID == fTxnid.id)
            {
              lockReleased = fDbrm-&gt;releaseTableLock(tableLocks[k].id);

              if (!lockReleased)
              {
                ostringstream os;
                os &lt;&lt; &quot;tablelock id &quot; &lt;&lt; tableLocks[k].id &lt;&lt; &quot; is not found&quot;;
                throw std::runtime_error(os.str());
              }
            }
          }

          fDbrm-&gt;committed(fTxnid);
          fTxnid = fDbrm-&gt;newTxnID(fSessionID, true, true);

          if (!fTxnid.valid)
          {
            throw std::runtime_error(std::string(&quot;Unable to start a transaction. Check critical log.&quot;));
          }
        }

        processStatement();
      }
    }
    catch (const std::exception&amp; ex)
    {
      logError(DDLPackageProcessor::NOT_ACCEPTING_PACKAGES, ex.what(), true);
      throw;
    }
  }

 private:
  int commitTransaction(uint32_t txnID, std::string&amp; errorMsg)
  {
    auto WEClient =
        std::unique_ptr&lt;WriteEngine::WEClients&gt;(new WriteEngine::WEClients(WriteEngine::WEClients::DDLPROC));
    auto PMCount = WEClient-&gt;getPmCount();
    ByteStream bytestream;
    uint64_t uniqueId = fDbrm-&gt;getUnique64();
    WEClient-&gt;addQueue(uniqueId);
    bytestream &lt;&lt; (ByteStream::byte)WE_SVR_COMMIT_VERSION;
    bytestream &lt;&lt; uniqueId;
    bytestream &lt;&lt; txnID;
    uint32_t msgRecived = 0;
    WEClient-&gt;write_to_all(bytestream);
    boost::shared_ptr&lt;messageqcpp::ByteStream&gt; bsIn;
    bsIn.reset(new ByteStream());
    int rc = 0;
    ByteStream::byte tmp8;

    while (true)
    {
      if (msgRecived == PMCount)
        break;

      WEClient-&gt;read(uniqueId, bsIn);

      if (bsIn-&gt;length() == 0)  // read error
      {
        rc = 1;
        errorMsg = &quot;DDL cannot communicate with WES&quot;;
        WEClient-&gt;removeQueue(uniqueId);
        break;
      }
      else
      {
        *bsIn &gt;&gt; tmp8;
        rc = tmp8;

        if (rc != 0)
        {
          *bsIn &gt;&gt; errorMsg;
          WEClient-&gt;removeQueue(uniqueId);
          break;
        }
        else
          msgRecived++;
      }
    }

    return rc;
  }

  void processStatement()
  {
    DDLPackageProcessor::DDLResult result;
    result.result = DDLPackageProcessor::NO_ERROR;
    // boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr;

    try
    {
      if (!fOamCache)
        fOamCache = oam::OamCache::makeOamCache();
      // cout &lt;&lt; &quot;DDLProc received package &quot; &lt;&lt; fPackageType &lt;&lt; endl;
      switch (fPackageType)
      {
        case ddlpackage::DDL_CREATE_TABLE_STATEMENT:
        {
          CreateTableStatement createTableStmt;
          createTableStmt.unserialize(fByteStream);
          boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr =
              CalpontSystemCatalog::makeCalpontSystemCatalog(createTableStmt.fSessionID);
          boost::scoped_ptr&lt;CreateTableProcessor&gt; processor(new CreateTableProcessor(fDbrm));
          processor-&gt;fTxnid.id = fTxnid.id;
          processor-&gt;fTxnid.valid = true;
          if (fDebugLevel)
            processor-&gt;setDebugLevel(static_cast&lt;DDLPackageProcessor::DebugLevel&gt;(*fDebugLevel));
          // cout &lt;&lt; &quot;create table using txnid &quot; &lt;&lt; fTxnid.id &lt;&lt; endl;

          QueryTeleStats qts;
          qts.query_uuid = QueryTeleClient::genUUID();
          qts.msg_type = QueryTeleStats::QT_START;
          qts.start_time = QueryTeleClient::timeNowms();
          qts.session_id = createTableStmt.fSessionID;
          qts.query_type = &quot;CREATE&quot;;
          qts.query = createTableStmt.fSql;
          qts.system_name = fOamCache-&gt;getSystemName();
          qts.module_name = fOamCache-&gt;getModuleName();
          qts.schema_name = createTableStmt.schemaName();
          fQtc.postQueryTele(qts);

          result = processor-&gt;processPackage(&amp;createTableStmt);

          systemCatalogPtr-&gt;removeCalpontSystemCatalog(createTableStmt.fSessionID);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(createTableStmt.fSessionID | 0x80000000);

          qts.msg_type = QueryTeleStats::QT_SUMMARY;
          qts.end_time = QueryTeleClient::timeNowms();
          fQtc.postQueryTele(qts);
        }
        break;

        case ddlpackage::DDL_ALTER_TABLE_STATEMENT:
        {
          AlterTableStatement alterTableStmt;
          alterTableStmt.unserialize(fByteStream);
          boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr =
              CalpontSystemCatalog::makeCalpontSystemCatalog(alterTableStmt.fSessionID);
          boost::scoped_ptr&lt;AlterTableProcessor&gt; processor(new AlterTableProcessor(fDbrm));
          processor-&gt;fTxnid.id = fTxnid.id;
          processor-&gt;fTxnid.valid = true;
          if (fDebugLevel)
            processor-&gt;setDebugLevel(static_cast&lt;DDLPackageProcessor::DebugLevel&gt;(*fDebugLevel));

          QueryTeleStats qts;
          qts.query_uuid = QueryTeleClient::genUUID();
          qts.msg_type = QueryTeleStats::QT_START;
          qts.start_time = QueryTeleClient::timeNowms();
          qts.session_id = alterTableStmt.fSessionID;
          qts.query_type = &quot;ALTER&quot;;
          qts.query = alterTableStmt.fSql;
          qts.system_name = fOamCache-&gt;getSystemName();
          qts.module_name = fOamCache-&gt;getModuleName();
          qts.schema_name = alterTableStmt.schemaName();
          fQtc.postQueryTele(qts);

          processor-&gt;fTimeZone = alterTableStmt.getTimeZone();

          result = processor-&gt;processPackage(&amp;alterTableStmt);

          systemCatalogPtr-&gt;removeCalpontSystemCatalog(alterTableStmt.fSessionID);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(alterTableStmt.fSessionID | 0x80000000);

          qts.msg_type = QueryTeleStats::QT_SUMMARY;
          qts.end_time = QueryTeleClient::timeNowms();
          fQtc.postQueryTele(qts);
        }
        break;

        case ddlpackage::DDL_DROP_TABLE_STATEMENT:
        {
          DropTableStatement dropTableStmt;
          dropTableStmt.unserialize(fByteStream);
          boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr =
              CalpontSystemCatalog::makeCalpontSystemCatalog(dropTableStmt.fSessionID);
          boost::scoped_ptr&lt;DropTableProcessor&gt; processor(new DropTableProcessor(fDbrm));

          processor-&gt;fTxnid.id = fTxnid.id;
          processor-&gt;fTxnid.valid = true;
          if (fDebugLevel)
            processor-&gt;setDebugLevel(static_cast&lt;DDLPackageProcessor::DebugLevel&gt;(*fDebugLevel));

          QueryTeleStats qts;
          qts.query_uuid = QueryTeleClient::genUUID();
          qts.msg_type = QueryTeleStats::QT_START;
          qts.start_time = QueryTeleClient::timeNowms();
          qts.session_id = dropTableStmt.fSessionID;
          qts.query_type = &quot;DROP&quot;;
          qts.query = dropTableStmt.fSql;
          qts.system_name = fOamCache-&gt;getSystemName();
          qts.module_name = fOamCache-&gt;getModuleName();
          qts.schema_name = dropTableStmt.schemaName();
          fQtc.postQueryTele(qts);

          // cout &lt;&lt; &quot;Drop table using txnid &quot; &lt;&lt; fTxnid.id &lt;&lt; endl;
          result = processor-&gt;processPackage(&amp;dropTableStmt);

          systemCatalogPtr-&gt;removeCalpontSystemCatalog(dropTableStmt.fSessionID);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(dropTableStmt.fSessionID | 0x80000000);

          qts.msg_type = QueryTeleStats::QT_SUMMARY;
          qts.end_time = QueryTeleClient::timeNowms();
          fQtc.postQueryTele(qts);
        }
        break;

        case ddlpackage::DDL_TRUNC_TABLE_STATEMENT:
        {
          TruncTableStatement truncTableStmt;
          truncTableStmt.unserialize(fByteStream);
          boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr =
              CalpontSystemCatalog::makeCalpontSystemCatalog(truncTableStmt.fSessionID);
          boost::scoped_ptr&lt;TruncTableProcessor&gt; processor(new TruncTableProcessor(fDbrm));

          processor-&gt;fTxnid.id = fTxnid.id;
          processor-&gt;fTxnid.valid = true;
          if (fDebugLevel)
            processor-&gt;setDebugLevel(static_cast&lt;DDLPackageProcessor::DebugLevel&gt;(*fDebugLevel));

          QueryTeleStats qts;
          qts.query_uuid = QueryTeleClient::genUUID();
          qts.msg_type = QueryTeleStats::QT_START;
          qts.start_time = QueryTeleClient::timeNowms();
          qts.session_id = truncTableStmt.fSessionID;
          qts.query_type = &quot;TRUNCATE&quot;;
          qts.query = truncTableStmt.fSql;
          qts.system_name = fOamCache-&gt;getSystemName();
          qts.module_name = fOamCache-&gt;getModuleName();
          qts.schema_name = truncTableStmt.schemaName();
          fQtc.postQueryTele(qts);

          result = processor-&gt;processPackage(&amp;truncTableStmt);

          systemCatalogPtr-&gt;removeCalpontSystemCatalog(truncTableStmt.fSessionID);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(truncTableStmt.fSessionID | 0x80000000);

          qts.msg_type = QueryTeleStats::QT_SUMMARY;
          qts.end_time = QueryTeleClient::timeNowms();
          fQtc.postQueryTele(qts);
        }
        break;

        case ddlpackage::DDL_MARK_PARTITION_STATEMENT:
        {
          MarkPartitionStatement markPartitionStmt;
          markPartitionStmt.unserialize(fByteStream);
          boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr =
              CalpontSystemCatalog::makeCalpontSystemCatalog(markPartitionStmt.fSessionID);
          boost::scoped_ptr&lt;MarkPartitionProcessor&gt; processor(new MarkPartitionProcessor(fDbrm));
          (processor-&gt;fTxnid).id = fTxnid.id;
          (processor-&gt;fTxnid).valid = true;
          if (fDebugLevel)
            processor-&gt;setDebugLevel(static_cast&lt;DDLPackageProcessor::DebugLevel&gt;(*fDebugLevel));

          result = processor-&gt;processPackage(&amp;markPartitionStmt);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(markPartitionStmt.fSessionID);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(markPartitionStmt.fSessionID | 0x80000000);
        }
        break;

        case ddlpackage::DDL_RESTORE_PARTITION_STATEMENT:
        {
          RestorePartitionStatement restorePartitionStmt;
          restorePartitionStmt.unserialize(fByteStream);
          boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr =
              CalpontSystemCatalog::makeCalpontSystemCatalog(restorePartitionStmt.fSessionID);
          boost::scoped_ptr&lt;RestorePartitionProcessor&gt; processor(new RestorePartitionProcessor(fDbrm));
          (processor-&gt;fTxnid).id = fTxnid.id;
          (processor-&gt;fTxnid).valid = true;
          if (fDebugLevel)
            processor-&gt;setDebugLevel(static_cast&lt;DDLPackageProcessor::DebugLevel&gt;(*fDebugLevel));

          result = processor-&gt;processPackage(&amp;restorePartitionStmt);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(restorePartitionStmt.fSessionID);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(restorePartitionStmt.fSessionID | 0x80000000);
        }
        break;

        case ddlpackage::DDL_DROP_PARTITION_STATEMENT:
        {
          DropPartitionStatement dropPartitionStmt;
          dropPartitionStmt.unserialize(fByteStream);
          boost::shared_ptr&lt;CalpontSystemCatalog&gt; systemCatalogPtr =
              CalpontSystemCatalog::makeCalpontSystemCatalog(dropPartitionStmt.fSessionID);
          boost::scoped_ptr&lt;DropPartitionProcessor&gt; processor(new DropPartitionProcessor(fDbrm));
          (processor-&gt;fTxnid).id = fTxnid.id;
          (processor-&gt;fTxnid).valid = true;
          if (fDebugLevel)
            processor-&gt;setDebugLevel(static_cast&lt;DDLPackageProcessor::DebugLevel&gt;(*fDebugLevel));

          result = processor-&gt;processPackage(&amp;dropPartitionStmt);
          systemCatalogPtr-&gt;removeCalpontSystemCatalog(dropPartitionStmt.fSessionID);
          systemCatalogPtr-&gt;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;
      }

      //@Bug 3427. No need to log user rror, just return the message to user.
      // Log errors
      if ((result.result != DDLPackageProcessor::NO_ERROR) &amp;&amp;
          (result.result != DDLPackageProcessor::USER_ERROR))
      {
        logging::LoggingID lid(23);
        logging::MessageLog ml(lid);

        ml.logErrorMessage(result.message);
      }

      string hdfstest = config::Config::makeConfig()-&gt;getConfig(&quot;Installation&quot;, &quot;DBRootStorageType&quot;);

      if (hdfstest == &quot;hdfs&quot; || hdfstest == &quot;HDFS&quot;)
        cleanPMSysCache();

      messageqcpp::ByteStream results;
      messageqcpp::ByteStream::byte status = result.result;
      results &lt;&lt; status;
      results &lt;&lt; result.message.msg();

      fIos.write(results);

      fIos.close();
    }
    catch (quadbyte&amp; /*foo*/)
    {
      logError(DDLPackageProcessor::NOT_ACCEPTING_PACKAGES, &quot;Unrecognized package type&quot;, true);
    }
    catch (logging::IDBExcept&amp; idbEx)
    {
      cleanPMSysCache();
      logError(DDLPackageProcessor::CREATE_ERROR, idbEx.what(), true);
    }
    catch (...)
    {
      logError(DDLPackageProcessor::NOT_ACCEPTING_PACKAGES, &quot;Unknown error&quot;, true);
    }
  }

  void logError(messageqcpp::ByteStream::byte status, const std::string&amp; msg, bool closeSocket = false)
  {
    messageqcpp::ByteStream res;
    res &lt;&lt; status;
    res &lt;&lt; msg;
    fIos.write(res);
    cerr &lt;&lt; &quot;DDLProc error: &quot; &lt;&lt; msg &lt;&lt; endl;
    if (closeSocket)
      fIos.close();
  }

 private:
  messageqcpp::IOSocket fIos;
  messageqcpp::ByteStream fByteStream;
  messageqcpp::ByteStream::quadbyte fPackageType;
  uint32_t fSessionID;
  BRM::TxnID fTxnid;
  BRM::DBRM* fDbrm;
  QueryTeleClient fQtc;
  oam::OamCache* fOamCache = nullptr;
  bool fConcurrentSupport;
  uint32_t* fDebugLevel{nullptr};
};

}  // namespace

namespace ddlprocessor
{
DDLProcessor::DDLProcessor(int packageMaxThreads, int packageWorkQueueSize)
 : fPackageMaxThreads(packageMaxThreads), fPackageWorkQueueSize(packageWorkQueueSize), fMqServer(DDLProcName)
{
  fDdlPackagepool.setMaxThreads(fPackageMaxThreads);
  fDdlPackagepool.setQueueSize(fPackageWorkQueueSize);
  fDdlPackagepool.setName(&quot;DdlPackagepool&quot;);
  csc = CalpontSystemCatalog::makeCalpontSystemCatalog();
  csc-&gt;identity(CalpontSystemCatalog::EC);
  string teleServerHost(config::Config::makeConfig()-&gt;getConfig(&quot;QueryTele&quot;, &quot;Host&quot;));

  if (!teleServerHost.empty())
  {
    int teleServerPort =
        config::Config::fromText(config::Config::makeConfig()-&gt;getConfig(&quot;QueryTele&quot;, &quot;Port&quot;));

    if (teleServerPort &gt; 0)
    {
      fQtc.serverParms(QueryTeleServerParms(teleServerHost, teleServerPort));
    }
  }
}

DDLProcessor::~DDLProcessor()
{
}
void DDLProcessor::process()
{
  DBRM dbrm;
  messageqcpp::IOSocket ios;
  bool concurrentSupport = true;
  string concurrentTranStr =
      config::Config::makeConfig()-&gt;getConfig(&quot;SystemConfig&quot;, &quot;ConcurrentTransactions&quot;);

  if (concurrentTranStr.length() != 0)
  {
    if ((concurrentTranStr.compare(&quot;N&quot;) == 0) || (concurrentTranStr.compare(&quot;n&quot;) == 0))
      concurrentSupport = false;
  }

  cout &lt;&lt; &quot;DDLProc is ready...&quot; &lt;&lt; endl;

  try
  {
    for (;;)
    {
      try
      {
        ios = fMqServer.accept();
        fDdlPackagepool.invoke(PackageHandler(fQtc, &amp;dbrm, ios, concurrentSupport, &amp;debugLevel));
      }
      catch (...)
      {
        throw;
      }
    }
  }
  catch (exception&amp; ex)
  {
    cerr &lt;&lt; ex.what() &lt;&lt; endl;
    messageqcpp::ByteStream results;
    messageqcpp::ByteStream::byte status = DDLPackageProcessor::NOT_ACCEPTING_PACKAGES;

    results &lt;&lt; status;
    results &lt;&lt; ex.what();
    ios.write(results);

    ios.close();
  }
  catch (...)
  {
    cerr &lt;&lt; &quot;Caught unknown exception!&quot; &lt;&lt; endl;
    messageqcpp::ByteStream results;
    messageqcpp::ByteStream::byte status = DDLPackageProcessor::NOT_ACCEPTING_PACKAGES;

    results &lt;&lt; status;
    results &lt;&lt; &quot;Caught unknown exception!&quot;;
    ios.write(results);

    ios.close();
  }

  // wait for all the threads to exit
  fDdlPackagepool.wait();
}

int DDLProcessor::commitTransaction(uint32_t txnID, std::string&amp; errorMsg)
{
  fWEClient = new WriteEngine::WEClients(WriteEngine::WEClients::DDLPROC);
  fPMCount = fWEClient-&gt;getPmCount();
  ByteStream bytestream;
  DBRM dbrm;
  uint64_t uniqueId = dbrm.getUnique64();
  fWEClient-&gt;addQueue(uniqueId);
  bytestream &lt;&lt; (ByteStream::byte)WE_SVR_COMMIT_VERSION;
  bytestream &lt;&lt; uniqueId;
  bytestream &lt;&lt; txnID;
  uint32_t msgRecived = 0;
  fWEClient-&gt;write_to_all(bytestream);
  boost::shared_ptr&lt;messageqcpp::ByteStream&gt; bsIn;
  bsIn.reset(new ByteStream());
  int rc = 0;
  ByteStream::byte tmp8;

  while (1)
  {
    if (msgRecived == fPMCount)
      break;

    fWEClient-&gt;read(uniqueId, bsIn);

    if (bsIn-&gt;length() == 0)  // read error
    {
      rc = 1;
      errorMsg = &quot;DDL cannot communicate with WES&quot;;
      fWEClient-&gt;removeQueue(uniqueId);
      break;
    }
    else
    {
      *bsIn &gt;&gt; tmp8;
      rc = tmp8;

      if (rc != 0)
      {
        *bsIn &gt;&gt; errorMsg;
        fWEClient-&gt;removeQueue(uniqueId);
        break;
      }
      else
        msgRecived++;
    }
  }

  delete fWEClient;
  fWEClient = 0;
  return rc;
}
}  // namespace ddlprocessor
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Fcolumnstore%2Fcolumnstore%2Fddlproc'>🔙 Back</a></p>