Trix.cpp

/*
   Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.

   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 St, Fifth Floor, Boston, MA 02110-1301  USA
*/

#include "Trix.hpp"

#include <string.h>
#include <kernel_types.h>
#include <NdbOut.hpp>

#include <signaldata/ReadNodesConf.hpp>
#include <signaldata/NodeFailRep.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <signaldata/GetTabInfo.hpp>
#include <signaldata/DictTabInfo.hpp>
#include <signaldata/CopyData.hpp>
#include <signaldata/BuildIndxImpl.hpp>
#include <signaldata/SumaImpl.hpp>
#include <signaldata/UtilPrepare.hpp>
#include <signaldata/UtilExecute.hpp>
#include <signaldata/UtilRelease.hpp>
#include <SectionReader.hpp>
#include <AttributeHeader.hpp>
#include <signaldata/TcKeyReq.hpp>

#include <signaldata/DbinfoScan.hpp>
#include <signaldata/TransIdAI.hpp>
#include <signaldata/WaitGCP.hpp>

#define CONSTRAINT_VIOLATION 893

static
bool
check_timeout(Uint32 errCode)
{
  switch(errCode){
  case 266:
    return true;
  }
  return false;
}

#define DEBUG(x) { ndbout << "TRIX::" << x << endl; }

Trix::Trix(Block_context& ctx) :
  SimulatedBlock(TRIX, ctx),
  c_theNodes(c_theNodeRecPool),
  c_masterNodeId(0),
  c_masterTrixRef(0),
  c_noNodesFailed(0),
  c_noActiveNodes(0),
  c_theSubscriptions(c_theSubscriptionRecPool)
{
  BLOCK_CONSTRUCTOR(Trix);

  // Add received signals
  addRecSignal(GSN_READ_CONFIG_REQ,  &Trix::execREAD_CONFIG_REQ);
  addRecSignal(GSN_STTOR,  &Trix::execSTTOR);
  addRecSignal(GSN_NDB_STTOR,  &Trix::execNDB_STTOR); // Forwarded from DICT
  addRecSignal(GSN_READ_NODESCONF, &Trix::execREAD_NODESCONF);
  addRecSignal(GSN_READ_NODESREF, &Trix::execREAD_NODESREF);
  addRecSignal(GSN_NODE_FAILREP, &Trix::execNODE_FAILREP);
  addRecSignal(GSN_INCL_NODEREQ, &Trix::execINCL_NODEREQ);
  addRecSignal(GSN_DUMP_STATE_ORD, &Trix::execDUMP_STATE_ORD);
  addRecSignal(GSN_DBINFO_SCANREQ, &Trix::execDBINFO_SCANREQ);

  // Index build
  addRecSignal(GSN_BUILD_INDX_IMPL_REQ, &Trix::execBUILD_INDX_IMPL_REQ);
  // Dump testing
  addRecSignal(GSN_BUILD_INDX_IMPL_CONF, &Trix::execBUILD_INDX_IMPL_CONF);
  addRecSignal(GSN_BUILD_INDX_IMPL_REF, &Trix::execBUILD_INDX_IMPL_REF);

  addRecSignal(GSN_COPY_DATA_IMPL_REQ, &Trix::execCOPY_DATA_IMPL_REQ);

  addRecSignal(GSN_UTIL_PREPARE_CONF, &Trix::execUTIL_PREPARE_CONF);
  addRecSignal(GSN_UTIL_PREPARE_REF, &Trix::execUTIL_PREPARE_REF);
  addRecSignal(GSN_UTIL_EXECUTE_CONF, &Trix::execUTIL_EXECUTE_CONF);
  addRecSignal(GSN_UTIL_EXECUTE_REF, &Trix::execUTIL_EXECUTE_REF);
  addRecSignal(GSN_UTIL_RELEASE_CONF, &Trix::execUTIL_RELEASE_CONF);
  addRecSignal(GSN_UTIL_RELEASE_REF, &Trix::execUTIL_RELEASE_REF);


  // Suma signals
  addRecSignal(GSN_SUB_CREATE_CONF, &Trix::execSUB_CREATE_CONF);
  addRecSignal(GSN_SUB_CREATE_REF, &Trix::execSUB_CREATE_REF);
  addRecSignal(GSN_SUB_REMOVE_CONF, &Trix::execSUB_REMOVE_CONF);
  addRecSignal(GSN_SUB_REMOVE_REF, &Trix::execSUB_REMOVE_REF);
  addRecSignal(GSN_SUB_SYNC_CONF, &Trix::execSUB_SYNC_CONF);
  addRecSignal(GSN_SUB_SYNC_REF, &Trix::execSUB_SYNC_REF);
  addRecSignal(GSN_SUB_SYNC_CONTINUE_REQ, &Trix::execSUB_SYNC_CONTINUE_REQ);
  addRecSignal(GSN_SUB_TABLE_DATA, &Trix::execSUB_TABLE_DATA);

  addRecSignal(GSN_WAIT_GCP_REF, &Trix::execWAIT_GCP_REF);
  addRecSignal(GSN_WAIT_GCP_CONF, &Trix::execWAIT_GCP_CONF);

  // index stats
  addRecSignal(GSN_INDEX_STAT_IMPL_REQ, &Trix::execINDEX_STAT_IMPL_REQ);
  addRecSignal(GSN_GET_TABINFO_CONF, &Trix::execGET_TABINFO_CONF);
  addRecSignal(GSN_GET_TABINFOREF, &Trix::execGET_TABINFO_REF);

  // index stats sys tables
  c_statGetMetaDone = false;
}

Trix::~Trix()
{
}

void 
Trix::execREAD_CONFIG_REQ(Signal* signal)
{
  jamEntry();

  const ReadConfigReq * req = (ReadConfigReq*)signal->getDataPtr();

  Uint32 ref = req->senderRef;
  Uint32 senderData = req->senderData;

  const ndb_mgm_configuration_iterator * p = 
    m_ctx.m_config.getOwnConfigIterator();
  ndbrequire(p != 0);

  // Allocate pool sizes
  c_theAttrOrderBufferPool.setSize(100);
  c_theSubscriptionRecPool.setSize(100);
  c_statOpPool.setSize(5);

  DLList<SubscriptionRecord> subscriptions(c_theSubscriptionRecPool);
  SubscriptionRecPtr subptr;
  while(subscriptions.seize(subptr) == true) {
    new (subptr.p) SubscriptionRecord(c_theAttrOrderBufferPool);
  }
  subscriptions.release();

  ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
  conf->senderRef = reference();
  conf->senderData = senderData;
  sendSignal(ref, GSN_READ_CONFIG_CONF, signal, 
             ReadConfigConf::SignalLength, JBB);
}

void Trix::execSTTOR(Signal* signal) 
{
  jamEntry();                            

  //const Uint32 startphase   = signal->theData[1];
  const Uint32 theSignalKey = signal->theData[6];
  
  signal->theData[0] = theSignalKey;
  signal->theData[3] = 1;
  signal->theData[4] = 255; // No more start phases from missra
  sendSignal(NDBCNTR_REF, GSN_STTORRY, signal, 5, JBB);
  return;
}//Trix::execSTTOR()

void Trix::execNDB_STTOR(Signal* signal) 
{
  jamEntry();                            
  BlockReference ndbcntrRef = signal->theData[0];        
  Uint16 startphase = signal->theData[2];      /* RESTART PHASE           */
  Uint16 mynode = signal->theData[1];            
  //Uint16 restarttype = signal->theData[3];     
  //UintR configInfo1 = signal->theData[6];     /* CONFIGRATION INFO PART 1 */
  //UintR configInfo2 = signal->theData[7];     /* CONFIGRATION INFO PART 2 */
  switch (startphase) {
  case 3:
    jam();
    /* SYMBOLIC START PHASE 4             */
    /* ABSOLUTE PHASE 5                   */
    /* REQUEST NODE IDENTITIES FROM DBDIH */
    signal->theData[0] = calcTrixBlockRef(mynode);
    sendSignal(ndbcntrRef, GSN_READ_NODESREQ, signal, 1, JBB);
    return;
    break;
  case 6:
    break;
  default:
    break;
  }
}

void Trix::execREAD_NODESCONF(Signal* signal)
{
  jamEntry();

  ReadNodesConf * const  readNodes = (ReadNodesConf *)signal->getDataPtr();
  //Uint32 noOfNodes   = readNodes->noOfNodes;
  NodeRecPtr nodeRecPtr;

  c_masterNodeId = readNodes->masterNodeId;
  c_masterTrixRef = RNIL;
  c_noNodesFailed = 0;

  for(unsigned i = 0; i < MAX_NDB_NODES; i++) {
    jam();
    if(NdbNodeBitmask::get(readNodes->allNodes, i)) {
      // Node is defined
      jam();
      ndbrequire(c_theNodes.seizeId(nodeRecPtr, i));
      nodeRecPtr.p->trixRef = calcTrixBlockRef(i);
      if (i == c_masterNodeId) {
        c_masterTrixRef = nodeRecPtr.p->trixRef;
      }
      if(NdbNodeBitmask::get(readNodes->inactiveNodes, i)){
        // Node is not active
        jam();
        arrGuard(c_noNodesFailed, MAX_NDB_NODES);
        nodeRecPtr.p->alive = false;
        c_noNodesFailed++;
        c_blockState = Trix::NODE_FAILURE;
      }
      else {
        // Node is active
        jam();
        c_noActiveNodes++;
        nodeRecPtr.p->alive = true;
      }
    }
  }
  if (c_noNodesFailed == 0) {
    c_blockState = Trix::STARTED;
  }
}

void Trix::execREAD_NODESREF(Signal* signal)
{
  // NYI
}

void Trix::execNODE_FAILREP(Signal* signal)
{
  jamEntry();
  NodeFailRep * const  nodeFail = (NodeFailRep *) signal->getDataPtr();

  //Uint32 failureNr    = nodeFail->failNo;
  //Uint32 numberNodes  = nodeFail->noOfNodes;
  Uint32 masterNodeId = nodeFail->masterNodeId;

  NodeRecPtr nodeRecPtr;

  for(c_theNodes.first(nodeRecPtr); 
      nodeRecPtr.i != RNIL; 
      c_theNodes.next(nodeRecPtr)) {
    if(NdbNodeBitmask::get(nodeFail->theNodes, nodeRecPtr.i)) {
      nodeRecPtr.p->alive = false;
      c_noNodesFailed++;
      c_noActiveNodes--;      
    }
  }
  if (c_masterNodeId != masterNodeId) {
    c_masterNodeId = masterNodeId;
    NodeRecord* nodeRec = c_theNodes.getPtr(masterNodeId);
    c_masterTrixRef = nodeRec->trixRef;
  }
}

void Trix::execINCL_NODEREQ(Signal* signal)
{
  jamEntry();
  UintR node_id = signal->theData[1];
  NodeRecord* nodeRec = c_theNodes.getPtr(node_id);
  nodeRec->alive = true;
  c_noNodesFailed--;
  c_noActiveNodes++;      
  nodeRec->trixRef = calcTrixBlockRef(node_id);
  if (c_noNodesFailed == 0) {
    c_blockState = Trix::STARTED;
  }  
}

// Debugging
void
Trix::execDUMP_STATE_ORD(Signal* signal)
{
  jamEntry();

  DumpStateOrd * dumpStateOrd = (DumpStateOrd *)signal->getDataPtr();

  switch(dumpStateOrd->args[0]) {
  case(300): {// ok
    // index2 -T; index2 -I -n10000; index2 -c
    // all dump 300 0 0 0 0 0 4 2
    // select_count INDEX0000
    BuildIndxImplReq * buildIndxReq = (BuildIndxImplReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
                     signal->theData + 1, 
                     BuildIndxImplReq::SignalLength);
    buildIndxReq->senderRef = reference(); // return to me
    buildIndxReq->parallelism = 10;
    Uint32 indexColumns[1] = {1};
    Uint32 keyColumns[1] = {0};
    struct LinearSectionPtr ls_ptr[3];
    ls_ptr[0].p = indexColumns;
    ls_ptr[0].sz = 1;
    ls_ptr[1].p = keyColumns;
    ls_ptr[1].sz = 1;
    sendSignal(reference(),
               GSN_BUILD_INDX_IMPL_REQ,
               signal,
               BuildIndxImplReq::SignalLength,
               JBB, ls_ptr, 2);
    break;
  }
  case(301): { // ok
    // index2 -T; index2 -I -n10000; index2 -c -p
    // all dump 301 0 0 0 0 0 4 2
    // select_count INDEX0000
    BuildIndxImplReq * buildIndxReq = (BuildIndxImplReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
                     signal->theData + 1, 
                     BuildIndxImplReq::SignalLength);
    buildIndxReq->senderRef = reference(); // return to me
    buildIndxReq->parallelism = 10;
    Uint32 indexColumns[2] = {0, 1};
    Uint32 keyColumns[1] = {0};
    struct LinearSectionPtr ls_ptr[3];
    ls_ptr[0].p = indexColumns;
    ls_ptr[0].sz = 2;
    ls_ptr[1].p = keyColumns;
    ls_ptr[1].sz = 1;
    sendSignal(reference(),
               GSN_BUILD_INDX_IMPL_REQ,
               signal,
               BuildIndxImplReq::SignalLength,
               JBB, ls_ptr, 2);
    break;
  }
  case(302): { // ok
    // index -T; index -I -n1000; index -c -p
    // all dump 302 0 0 0 0 0 4 2
    // select_count PNUMINDEX0000
    BuildIndxImplReq * buildIndxReq = (BuildIndxImplReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
                     signal->theData + 1, 
                     BuildIndxImplReq::SignalLength);
    buildIndxReq->senderRef = reference(); // return to me
    buildIndxReq->parallelism = 10;
    Uint32 indexColumns[3] = {0, 3, 5};
    Uint32 keyColumns[1] = {0};
    struct LinearSectionPtr ls_ptr[3];
    ls_ptr[0].p = indexColumns;
    ls_ptr[0].sz = 3;
    ls_ptr[1].p = keyColumns;
    ls_ptr[1].sz = 1;
    sendSignal(reference(),
               GSN_BUILD_INDX_IMPL_REQ,
               signal,
               BuildIndxImplReq::SignalLength,
               JBB, ls_ptr, 2);
    break;
  }
  case(303): { // ok
    // index -T -2; index -I -2 -n1000; index -c -p
    // all dump 303 0 0 0 0 0 4 2
    // select_count PNUMINDEX0000
    BuildIndxImplReq * buildIndxReq = (BuildIndxImplReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
                     signal->theData + 1, 
                     BuildIndxImplReq::SignalLength);
    buildIndxReq->senderRef = reference(); // return to me
    buildIndxReq->parallelism = 10;
    Uint32 indexColumns[3] = {0, 3, 5};
    Uint32 keyColumns[2] = {0, 1};
    struct LinearSectionPtr ls_ptr[3];
    ls_ptr[0].p = indexColumns;
    ls_ptr[0].sz = 3;
    ls_ptr[1].p = keyColumns;
    ls_ptr[1].sz = 2;
    sendSignal(reference(),
               GSN_BUILD_INDX_IMPL_REQ,
               signal,
               BuildIndxImplReq::SignalLength,
               JBB, ls_ptr, 2);
    break;
  }
  case(304): { // ok
    // index -T -L; index -I -L -n1000; index -c -p
    // all dump 304 0 0 0 0 0 4 2
    // select_count PNUMINDEX0000
    BuildIndxImplReq * buildIndxReq = (BuildIndxImplReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
                     signal->theData + 1, 
                     BuildIndxImplReq::SignalLength);
    buildIndxReq->senderRef = reference(); // return to me
    buildIndxReq->parallelism = 10;
    Uint32 indexColumns[3] = {0, 3, 5};
    Uint32 keyColumns[1] = {0};
    struct LinearSectionPtr ls_ptr[3];
    ls_ptr[0].p = indexColumns;
    ls_ptr[0].sz = 3;
    ls_ptr[1].p = keyColumns;
    ls_ptr[1].sz = 1;
    sendSignal(reference(),
               GSN_BUILD_INDX_IMPL_REQ,
               signal,
               BuildIndxImplReq::SignalLength,
               JBB, ls_ptr, 2);
    break;
  }
  case(305): { // ok
    // index -T -2 -L; index -I -2 -L -n1000; index -c -p
    // all dump 305 0 0 0 0 0 4 2
    // select_count PNUMINDEX0000
    BuildIndxImplReq * buildIndxReq = (BuildIndxImplReq *)signal->getDataPtrSend();
    
    MEMCOPY_NO_WORDS(buildIndxReq, 
                     signal->theData + 1, 
                     BuildIndxImplReq::SignalLength);
    buildIndxReq->senderRef = reference(); // return to me
    buildIndxReq->parallelism = 10;
    Uint32 indexColumns[3] = {0, 3, 5};
    Uint32 keyColumns[2] = {0, 1};
    struct LinearSectionPtr ls_ptr[3];
    ls_ptr[0].p = indexColumns;
    ls_ptr[0].sz = 3;
    ls_ptr[1].p = keyColumns;
    ls_ptr[1].sz = 2;
    sendSignal(reference(),
               GSN_BUILD_INDX_IMPL_REQ,
               signal,
               BuildIndxImplReq::SignalLength,
               JBB, ls_ptr, 2);
    break;
  }
  default: {
    // Ignore
  }
  }

  if (signal->theData[0] == DumpStateOrd::SchemaResourceSnapshot)
  {
    RSS_AP_SNAPSHOT_SAVE(c_theSubscriptionRecPool);
    RSS_AP_SNAPSHOT_SAVE(c_statOpPool);
    return;
  }

  if (signal->theData[0] == DumpStateOrd::SchemaResourceCheckLeak)
  {
    RSS_AP_SNAPSHOT_CHECK(c_theSubscriptionRecPool);
    RSS_AP_SNAPSHOT_CHECK(c_statOpPool);
    return;
  }
  
  if (signal->theData[0] == 8004)
  {
    infoEvent("TRIX: c_theSubscriptionRecPool size: %u free: %u",
              c_theSubscriptionRecPool.getSize(),
              c_theSubscriptionRecPool.getNoOfFree());
    return;
  }

}

void Trix::execDBINFO_SCANREQ(Signal *signal)
{
  DbinfoScanReq req= *(DbinfoScanReq*)signal->theData;
  const Ndbinfo::ScanCursor* cursor =
    CAST_CONSTPTR(Ndbinfo::ScanCursor, DbinfoScan::getCursorPtr(&req));
  Ndbinfo::Ratelimit rl;

  jamEntry();

  switch(req.tableId){
  case Ndbinfo::POOLS_TABLEID:
  {
    Ndbinfo::pool_entry pools[] =
    {
      { "Attribute Order Buffer",
        c_theAttrOrderBufferPool.getUsed(),
        c_theAttrOrderBufferPool.getSize(),
        c_theAttrOrderBufferPool.getEntrySize(),
        c_theAttrOrderBufferPool.getUsedHi(),
        { 0,0,0,0 }},
      { "Subscription Record",
        c_theSubscriptionRecPool.getUsed(),
        c_theSubscriptionRecPool.getSize(),
        c_theSubscriptionRecPool.getEntrySize(),
        c_theSubscriptionRecPool.getUsedHi(),
        { 0,0,0,0 }},
      { NULL, 0,0,0,0,{0,0,0,0}}
    };

    const size_t num_config_params =
      sizeof(pools[0].config_params) / sizeof(pools[0].config_params[0]);
    Uint32 pool = cursor->data[0];
    BlockNumber bn = blockToMain(number());
    while(pools[pool].poolname)
    {
      jam();
      Ndbinfo::Row row(signal, req);
      row.write_uint32(getOwnNodeId());
      row.write_uint32(bn);           // block number
      row.write_uint32(instance());   // block instance
      row.write_string(pools[pool].poolname);
      row.write_uint64(pools[pool].used);
      row.write_uint64(pools[pool].total);
      row.write_uint64(pools[pool].used_hi);
      row.write_uint64(pools[pool].entry_size);
      for (size_t i = 0; i < num_config_params; i++)
        row.write_uint32(pools[pool].config_params[i]);
      ndbinfo_send_row(signal, req, row, rl);
      pool++;
      if (rl.need_break(req))
      {
        jam();
        ndbinfo_send_scan_break(signal, req, rl, pool);
        return;
      }
    }
    break;
  }
  default:
    break;
  }

  ndbinfo_send_scan_conf(signal, req, rl);
}

// Build index
void Trix:: execBUILD_INDX_IMPL_REQ(Signal* signal)
{
  jamEntry();
  DBUG_ENTER("Trix:: execBUILD_INDX_IMPL_REQ");

  const BuildIndxImplReq
    buildIndxReqData = *(const BuildIndxImplReq*)signal->getDataPtr(),
    *buildIndxReq = &buildIndxReqData;

  // Seize a subscription record
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;
  SectionHandle handle(this, signal);

  if (!c_theSubscriptions.seizeId(subRecPtr, buildIndxReq->buildId)) {
    jam();
    // Failed to allocate subscription record
    BuildIndxRef* buildIndxRef = (BuildIndxRef*)signal->getDataPtrSend();

    buildIndxRef->errorCode = BuildIndxRef::AllocationFailure;
    releaseSections(handle);
    sendSignal(buildIndxReq->senderRef, GSN_BUILD_INDX_IMPL_REF, signal,
               BuildIndxRef::SignalLength, JBB);
    DBUG_VOID_RETURN;
  }

  subRec = subRecPtr.p;
  subRec->errorCode = BuildIndxRef::NoError;
  subRec->userReference = buildIndxReq->senderRef;
  subRec->connectionPtr = buildIndxReq->senderData;
  subRec->schemaTransId = buildIndxReq->transId;
  subRec->subscriptionId = buildIndxReq->buildId;
  subRec->subscriptionKey = buildIndxReq->buildKey;
  subRec->indexType = buildIndxReq->indexType;
  subRec->sourceTableId = buildIndxReq->tableId;
  subRec->targetTableId = buildIndxReq->indexId;
  subRec->parallelism = buildIndxReq->parallelism;
  subRec->expectedConf = 0;
  subRec->subscriptionCreated = false;
  subRec->pendingSubSyncContinueConf = false;
  subRec->prepareId = RNIL;
  subRec->requestType = INDEX_BUILD;
  subRec->fragCount = 0;
  subRec->fragId = ZNIL;
  subRec->m_rows_processed = 0;
  subRec->m_flags = SubscriptionRecord::RF_WAIT_GCP; // Todo make configurable
  subRec->m_gci = 0;
  if (buildIndxReq->requestType & BuildIndxImplReq::RF_NO_DISK)
  {
    subRec->m_flags |= SubscriptionRecord::RF_NO_DISK;
  }

  // Get column order segments
  Uint32 noOfSections = handle.m_cnt;
  if (noOfSections > 0) {
    jam();
    SegmentedSectionPtr ptr;
    handle.getSection(ptr, BuildIndxImplReq::INDEX_COLUMNS);
    append(subRec->attributeOrder, ptr, getSectionSegmentPool());
    subRec->noOfIndexColumns = ptr.sz;
  }
  if (noOfSections > 1) {
    jam();
    SegmentedSectionPtr ptr;
    handle.getSection(ptr, BuildIndxImplReq::KEY_COLUMNS);
    append(subRec->attributeOrder, ptr, getSectionSegmentPool());
    subRec->noOfKeyColumns = ptr.sz;
  }

#if 0
  // Debugging
  printf("Trix:: execBUILD_INDX_IMPL_REQ: Attribute order:\n");
  subRec->attributeOrder.print(stdout);
#endif

  releaseSections(handle);
  prepareInsertTransactions(signal, subRecPtr);
  DBUG_VOID_RETURN;
}

void Trix:: execBUILD_INDX_IMPL_CONF(Signal* signal)
{
  printf("Trix:: execBUILD_INDX_IMPL_CONF\n");
}

void Trix:: execBUILD_INDX_IMPL_REF(Signal* signal)
{
  printf("Trix:: execBUILD_INDX_IMPL_REF\n");
}

void Trix::execUTIL_PREPARE_CONF(Signal* signal)
{
  jamEntry();
  UtilPrepareConf * utilPrepareConf = (UtilPrepareConf *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;

  subRecPtr.i = utilPrepareConf->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execUTIL_PREPARE_CONF: Failed to find subscription data %u\n", subRecPtr.i);
    return;
  }
  if (subRec->requestType == STAT_UTIL)
  {
    statUtilPrepareConf(signal, subRec->m_statPtrI);
    return;
  }
  subRecPtr.p = subRec;
  subRec->prepareId = utilPrepareConf->prepareId;
  setupSubscription(signal, subRecPtr);
}

void Trix::execUTIL_PREPARE_REF(Signal* signal)
{
  jamEntry();
  UtilPrepareRef * utilPrepareRef = (UtilPrepareRef *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;

  subRecPtr.i = utilPrepareRef->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execUTIL_PREPARE_REF: Failed to find subscription data %u\n", subRecPtr.i);
    return;
  }
  if (subRec->requestType == STAT_UTIL)
  {
    statUtilPrepareRef(signal, subRec->m_statPtrI);
    return;
  }
  subRecPtr.p = subRec;
  subRec->errorCode = (BuildIndxRef::ErrorCode)utilPrepareRef->errorCode;

  UtilReleaseConf* conf = (UtilReleaseConf*)signal->getDataPtrSend();
  conf->senderData = subRecPtr.i;
  execUTIL_RELEASE_CONF(signal);
}

void Trix::execUTIL_EXECUTE_CONF(Signal* signal)
{
  jamEntry();
  UtilExecuteConf * utilExecuteConf = (UtilExecuteConf *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;

  const Uint32 gci_hi = utilExecuteConf->gci_hi;
  const Uint32 gci_lo = utilExecuteConf->gci_lo;
  const Uint64 gci = gci_lo | (Uint64(gci_hi) << 32);

  subRecPtr.i = utilExecuteConf->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("rix::execUTIL_EXECUTE_CONF: Failed to find subscription data %u\n", subRecPtr.i);
    return;
  }
  if (subRec->requestType == STAT_UTIL)
  {
    statUtilExecuteConf(signal, subRec->m_statPtrI);
    return;
  }
  subRecPtr.p = subRec;
  subRec->expectedConf--;

  if (gci > subRecPtr.p->m_gci)
  {
    jam();
    subRecPtr.p->m_gci = gci;
  }

  checkParallelism(signal, subRec);
  if (subRec->expectedConf == 0)
  {
    if (subRec->m_flags & SubscriptionRecord::RF_WAIT_GCP)
    {
      jam();
      wait_gcp(signal, subRecPtr);
      return;
    }
    buildComplete(signal, subRecPtr);
  }
}

void Trix::execUTIL_EXECUTE_REF(Signal* signal)
{
  jamEntry();
  UtilExecuteRef * utilExecuteRef = (UtilExecuteRef *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;

  subRecPtr.i = utilExecuteRef->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execUTIL_EXECUTE_REF: Failed to find subscription data %u\n", subRecPtr.i);
    return;
  }
  if (subRec->requestType == STAT_UTIL)
  {
    statUtilExecuteRef(signal, subRec->m_statPtrI);
    return;
  }
  subRecPtr.p = subRec;
  ndbrequire(utilExecuteRef->errorCode == UtilExecuteRef::TCError);
  if(utilExecuteRef->TCErrorCode == CONSTRAINT_VIOLATION)
  {
    jam();
    buildFailed(signal, subRecPtr, BuildIndxRef::IndexNotUnique);
  }
  else if (check_timeout(utilExecuteRef->TCErrorCode))
  {
    jam();
    buildFailed(signal, subRecPtr, BuildIndxRef::DeadlockError);
  }
  else
  {
    jam();
    buildFailed(signal, subRecPtr,
                (BuildIndxRef::ErrorCode)utilExecuteRef->TCErrorCode);
  }
}

void Trix::execSUB_CREATE_CONF(Signal* signal)
{
  jamEntry();
  DBUG_ENTER("Trix::execSUB_CREATE_CONF");
  SubCreateConf * subCreateConf = (SubCreateConf *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;

  subRecPtr.i = subCreateConf->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execSUB_CREATE_CONF: Failed to find subscription data %u\n", subRecPtr.i);
    DBUG_VOID_RETURN;
  }
  subRec->subscriptionCreated = true;
  subRecPtr.p = subRec;

  DBUG_PRINT("info",("i: %u subscriptionId: %u, subscriptionKey: %u",
                     subRecPtr.i, subRecPtr.p->subscriptionId,
                     subRecPtr.p->subscriptionKey));

  startTableScan(signal, subRecPtr);
  DBUG_VOID_RETURN;
}

void Trix::execSUB_CREATE_REF(Signal* signal)
{
  jamEntry();
  DBUG_ENTER("Trix::execSUB_CREATE_REF");

  SubCreateRef * subCreateRef = (SubCreateRef *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;

  subRecPtr.i = subCreateRef->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL)
  {
    printf("Trix::execSUB_CREATE_REF: Failed to find subscription data %u\n", subRecPtr.i);
    return;
  }
  subRecPtr.p = subRec;
  subRecPtr.p->errorCode = (BuildIndxRef::ErrorCode)subCreateRef->errorCode;

  UtilReleaseReq * const req = (UtilReleaseReq*)signal->getDataPtrSend();
  req->prepareId = subRecPtr.p->prepareId;
  req->senderData = subRecPtr.i;

  sendSignal(DBUTIL_REF, GSN_UTIL_RELEASE_REQ, signal,
             UtilReleaseReq::SignalLength, JBB);

  DBUG_VOID_RETURN;
}

void Trix::execSUB_SYNC_CONF(Signal* signal)
{
  jamEntry();
  DBUG_ENTER("Trix::execSUB_SYNC_CONF");
  SubSyncConf * subSyncConf = (SubSyncConf *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;
  
  subRecPtr.i = subSyncConf->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execSUB_SYNC_CONF: Failed to find subscription data %u\n",
           subRecPtr.i);
    DBUG_VOID_RETURN;
  }

  subRecPtr.p = subRec;
  subRec->expectedConf--;
  checkParallelism(signal, subRec);
  if (subRec->expectedConf == 0)
  {
    if (subRec->m_flags & SubscriptionRecord::RF_WAIT_GCP)
    {
      jam();
      wait_gcp(signal, subRecPtr);
      DBUG_VOID_RETURN;
    }
    buildComplete(signal, subRecPtr);
  }
  DBUG_VOID_RETURN;
}

void Trix::execSUB_SYNC_REF(Signal* signal)
{
  jamEntry();
  DBUG_ENTER("Trix::execSUB_SYNC_REF");
  SubSyncRef * subSyncRef = (SubSyncRef *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;

  subRecPtr.i = subSyncRef->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execSUB_SYNC_REF: Failed to find subscription data %u\n", subRecPtr.i);
    DBUG_VOID_RETURN;
  }
  subRecPtr.p = subRec;
  buildFailed(signal, subRecPtr, BuildIndxRef::InternalError);
  DBUG_VOID_RETURN;
}

void Trix::execSUB_SYNC_CONTINUE_REQ(Signal* signal)
{
  SubSyncContinueReq  * subSyncContinueReq = 
    (SubSyncContinueReq *) signal->getDataPtr();
  
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;
  subRecPtr.i = subSyncContinueReq->subscriberData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execSUB_SYNC_CONTINUE_REQ: Failed to find subscription data %u\n", subRecPtr.i);
    return;
  }
  subRecPtr.p = subRec;
  subRec->pendingSubSyncContinueConf = true;
  subRec->syncPtr = subSyncContinueReq->senderData;
  checkParallelism(signal, subRec);
}

void Trix::execSUB_TABLE_DATA(Signal* signal)
{
  jamEntry();
  DBUG_ENTER("Trix::execSUB_TABLE_DATA");
  SubTableData * subTableData = (SubTableData *)signal->getDataPtr();
  SubscriptionRecPtr subRecPtr;
  SubscriptionRecord* subRec;
  subRecPtr.i = subTableData->senderData;
  if ((subRec = c_theSubscriptions.getPtr(subRecPtr.i)) == NULL) {
    printf("Trix::execSUB_TABLE_DATA: Failed to find subscription data %u\n", subRecPtr.i);
    DBUG_VOID_RETURN;
  }
  subRecPtr.p = subRec;
  switch(subRecPtr.p->requestType){
  case INDEX_BUILD:
    executeBuildInsertTransaction(signal, subRecPtr);
    break;
  case REORG_COPY:
  case REORG_DELETE:
    executeReorgTransaction(signal, subRecPtr, subTableData->takeOver);
    break;
  case STAT_UTIL:
    ndbrequire(false);
    break;
  case STAT_CLEAN:
    {
      StatOp& stat = statOpGetPtr(subRecPtr.p->m_statPtrI);
      statCleanExecute(signal, stat);
    }
    break;
  case STAT_SCAN:
    {
      StatOp& stat = statOpGetPtr(subRecPtr.p->m_statPtrI);
      statScanExecute(signal, stat);
    }
    break;
  }

  subRecPtr.p->m_rows_processed++;

  DBUG_VOID_RETURN;
}

void Trix::setupSubscription(Signal* signal, SubscriptionRecPtr subRecPtr)
{
  jam();
  DBUG_ENTER("Trix::setupSubscription");
  SubscriptionRecord* subRec = subRecPtr.p;
  SubCreateReq * subCreateReq = (SubCreateReq *)signal->getDataPtrSend();
//  Uint32 listLen = subRec->noOfIndexColumns + subRec->noOfKeyColumns;
  subCreateReq->senderRef = reference();
  subCreateReq->senderData = subRecPtr.i;
  subCreateReq->subscriptionId = subRec->subscriptionId;
  subCreateReq->subscriptionKey = subRec->subscriptionKey;
  subCreateReq->tableId = subRec->sourceTableId;
  subCreateReq->subscriptionType = SubCreateReq::SingleTableScan;
  subCreateReq->schemaTransId = subRec->schemaTransId;

  DBUG_PRINT("info",("i: %u subscriptionId: %u, subscriptionKey: %u",
                     subRecPtr.i, subCreateReq->subscriptionId,
                     subCreateReq->subscriptionKey));

  sendSignal(SUMA_REF, GSN_SUB_CREATE_REQ, 
             signal, SubCreateReq::SignalLength, JBB);

  DBUG_VOID_RETURN;
}

void Trix::startTableScan(Signal* signal, SubscriptionRecPtr subRecPtr)
{
  jam();

  Uint32 attributeList[MAX_ATTRIBUTES_IN_TABLE * 2];
  SubscriptionRecord* subRec = subRecPtr.p;
  AttrOrderBuffer::DataBufferIterator iter;
  Uint32 i = 0;

  bool moreAttributes = subRec->attributeOrder.first(iter);
  while (moreAttributes) {
    attributeList[i++] = *iter.data;
    moreAttributes = subRec->attributeOrder.next(iter);

  }
  // Merge index and key column segments
  struct LinearSectionPtr orderPtr[3];
  Uint32 noOfSections;
  orderPtr[0].p = attributeList;
  orderPtr[0].sz = subRec->attributeOrder.getSize();
  noOfSections = 1;

  SubSyncReq * subSyncReq = (SubSyncReq *)signal->getDataPtrSend();
  subSyncReq->senderRef = reference();
  subSyncReq->senderData = subRecPtr.i;
  subSyncReq->subscriptionId = subRec->subscriptionId;
  subSyncReq->subscriptionKey = subRec->subscriptionKey;
  subSyncReq->part = SubscriptionData::TableData;
  subSyncReq->requestInfo = 0;
  subSyncReq->fragCount = subRec->fragCount;
  subSyncReq->fragId = subRec->fragId;

  if (subRec->m_flags & SubscriptionRecord::RF_NO_DISK)
  {
    jam();
    subSyncReq->requestInfo |= SubSyncReq::NoDisk;
  }

  if (subRec->m_flags & SubscriptionRecord::RF_TUP_ORDER)
  {
    jam();
    subSyncReq->requestInfo |= SubSyncReq::TupOrder;
  }
  
  if (subRec->requestType == REORG_COPY)
  {
    jam();
    subSyncReq->requestInfo |= SubSyncReq::LM_Exclusive;
  }
  else if (subRec->requestType == REORG_DELETE)
  {
    jam();
    subSyncReq->requestInfo |= SubSyncReq::LM_Exclusive;
    subSyncReq->requestInfo |= SubSyncReq::Reorg;
  }
  else if (subRec->requestType == STAT_CLEAN)
  {
    jam();
    StatOp& stat = statOpGetPtr(subRecPtr.p->m_statPtrI);
    StatOp::Clean clean = stat.m_clean;
    orderPtr[1].p = clean.m_bound;
    orderPtr[1].sz = clean.m_boundSize;
    noOfSections = 2;
    subSyncReq->requestInfo |= SubSyncReq::LM_CommittedRead;
    subSyncReq->requestInfo |= SubSyncReq::RangeScan;
  }
  else if (subRec->requestType == STAT_SCAN)
  {
    jam();
    orderPtr[1].p = 0;
    orderPtr[1].sz = 0;
    noOfSections = 2;
    subSyncReq->requestInfo |= SubSyncReq::LM_CommittedRead;
    subSyncReq->requestInfo |= SubSyncReq::RangeScan;
    subSyncReq->requestInfo |= SubSyncReq::StatScan;
  }
  subRecPtr.p->expectedConf = 1;

  DBUG_PRINT("info",("i: %u subscriptionId: %u, subscriptionKey: %u",
                     subRecPtr.i, subSyncReq->subscriptionId,
                     subSyncReq->subscriptionKey));

  sendSignal(SUMA_REF, GSN_SUB_SYNC_REQ,
             signal, SubSyncReq::SignalLength, JBB, orderPtr, noOfSections);
}

void Trix::prepareInsertTransactions(Signal* signal, 
                                     SubscriptionRecPtr subRecPtr)
{
  SubscriptionRecord* subRec = subRecPtr.p;
  UtilPrepareReq * utilPrepareReq = 
    (UtilPrepareReq *)signal->getDataPtrSend();
  
  jam();
  utilPrepareReq->senderRef = reference();
  utilPrepareReq->senderData = subRecPtr.i;
  utilPrepareReq->schemaTransId = subRec->schemaTransId;

  const Uint32 pageSizeInWords = 128;
  Uint32 propPage[pageSizeInWords];
  LinearWriter w(&propPage[0],128);
  w.first();
  w.add(UtilPrepareReq::NoOfOperations, 1);
  w.add(UtilPrepareReq::OperationType, UtilPrepareReq::Write);
  w.add(UtilPrepareReq::TableId, subRec->targetTableId);
  // Add index attributes in increasing order and one PK attribute
  for(Uint32 i = 0; i < subRec->noOfIndexColumns + 1; i++)
    w.add(UtilPrepareReq::AttributeId, i);

#if 0
  // Debugging
  SimplePropertiesLinearReader reader(propPage, w.getWordsUsed());
  printf("Trix::prepareInsertTransactions: Sent SimpleProperties:\n");
  reader.printAll(ndbout);
#endif

  struct LinearSectionPtr sectionsPtr[UtilPrepareReq::NoOfSections];
  sectionsPtr[UtilPrepareReq::PROPERTIES_SECTION].p = propPage;
  sectionsPtr[UtilPrepareReq::PROPERTIES_SECTION].sz = w.getWordsUsed();
  sendSignal(DBUTIL_REF, GSN_UTIL_PREPARE_REQ, signal,
             UtilPrepareReq::SignalLength, JBB, 
             sectionsPtr, UtilPrepareReq::NoOfSections);
}

void Trix::executeBuildInsertTransaction(Signal* signal,
                                         SubscriptionRecPtr subRecPtr)
{
  jam();
  SubscriptionRecord* subRec = subRecPtr.p;
  UtilExecuteReq * utilExecuteReq = 
    (UtilExecuteReq *)signal->getDataPtrSend();

  utilExecuteReq->senderRef = reference();
  utilExecuteReq->senderData = subRecPtr.i;
  utilExecuteReq->prepareId = subRec->prepareId;
#if 0
  printf("Header size %u\n", headerPtr.sz);
  for(int i = 0; i < headerPtr.sz; i++)
    printf("H'%.8x ", headerBuffer[i]);
  printf("\n");
  
  printf("Data size %u\n", dataPtr.sz);
  for(int i = 0; i < dataPtr.sz; i++)
    printf("H'%.8x ", dataBuffer[i]);
  printf("\n");
#endif
  // Save scan result in linear buffers
  SectionHandle handle(this, signal);
  SegmentedSectionPtr headerPtr, dataPtr;

  handle.getSection(headerPtr, 0);
  handle.getSection(dataPtr, 1);

  Uint32* headerBuffer = signal->theData + 25;
  Uint32* dataBuffer = headerBuffer + headerPtr.sz;

  copy(headerBuffer, headerPtr);
  copy(dataBuffer, dataPtr);
  releaseSections(handle);

  // Calculate packed key size
  Uint32 noOfKeyData = 0;
  for(Uint32 i = 0; i < headerPtr.sz; i++) {
    AttributeHeader* keyAttrHead = (AttributeHeader *) headerBuffer + i;

    // Filter out NULL attributes
    if (keyAttrHead->isNULL())
      return;

    if (i < subRec->noOfIndexColumns)
      // Renumber index attributes in consequtive order
      keyAttrHead->setAttributeId(i);
    else
      // Calculate total size of PK attribute
      noOfKeyData += keyAttrHead->getDataSize();
  }
  // Increase expected CONF count
  subRec->expectedConf++;

  // Pack key attributes
  AttributeHeader::init(headerBuffer + subRec->noOfIndexColumns,
                        subRec->noOfIndexColumns,
                        noOfKeyData << 2);

  struct LinearSectionPtr sectionsPtr[UtilExecuteReq::NoOfSections];
  sectionsPtr[UtilExecuteReq::HEADER_SECTION].p = headerBuffer;
  sectionsPtr[UtilExecuteReq::HEADER_SECTION].sz = 
    subRec->noOfIndexColumns + 1;
  sectionsPtr[UtilExecuteReq::DATA_SECTION].p = dataBuffer;
  sectionsPtr[UtilExecuteReq::DATA_SECTION].sz = dataPtr.sz;
  sendSignal(DBUTIL_REF, GSN_UTIL_EXECUTE_REQ, signal,
             UtilExecuteReq::SignalLength, JBB,
             sectionsPtr, UtilExecuteReq::NoOfSections);
}

void Trix::executeReorgTransaction(Signal* signal,
                                   SubscriptionRecPtr subRecPtr,
                                   Uint32 takeOver)
{
  jam();
  SubscriptionRecord* subRec = subRecPtr.p;
  UtilExecuteReq * utilExecuteReq =
    (UtilExecuteReq *)signal->getDataPtrSend();

  const Uint32 tScanInfo = takeOver & 0x3FFFF;
  const Uint32 tTakeOverFragment = takeOver >> 20;
  {
    UintR scanInfo = 0;
    TcKeyReq::setTakeOverScanFlag(scanInfo, 1);
    TcKeyReq::setTakeOverScanFragment(scanInfo, tTakeOverFragment);
    TcKeyReq::setTakeOverScanInfo(scanInfo, tScanInfo);
    utilExecuteReq->scanTakeOver = scanInfo;
  }

  utilExecuteReq->senderRef = reference();
  utilExecuteReq->senderData = subRecPtr.i;
  utilExecuteReq->prepareId = subRec->prepareId;
#if 0
  printf("Header size %u\n", headerPtr.sz);
  for(int i = 0; i < headerPtr.sz; i++)
    printf("H'%.8x ", headerBuffer[i]);
  printf("\n");

  printf("Data size %u\n", dataPtr.sz);
  for(int i = 0; i < dataPtr.sz; i++)
    printf("H'%.8x ", dataBuffer[i]);
  printf("\n");
#endif
  // Increase expected CONF count
  subRec->expectedConf++;

  SectionHandle handle(this, signal);
  sendSignal(DBUTIL_REF, GSN_UTIL_EXECUTE_REQ, signal,
             UtilExecuteReq::SignalLength, JBB,
             &handle);
}

void
Trix::wait_gcp(Signal* signal, SubscriptionRecPtr subRecPtr, Uint32 delay)
{
  WaitGCPReq * req = (WaitGCPReq*)signal->getDataPtrSend();
  req->senderRef = reference();
  req->senderData = subRecPtr.i;
  req->requestType = WaitGCPReq::CurrentGCI;

  if (delay == 0)
  {
    jam();
    sendSignal(DBDIH_REF, GSN_WAIT_GCP_REQ, signal,
               WaitGCPReq::SignalLength, JBB);
  }
  else
  {
    jam();
    sendSignalWithDelay(DBDIH_REF, GSN_WAIT_GCP_REQ, signal,
                        delay, WaitGCPReq::SignalLength);
  }
}

void
Trix::execWAIT_GCP_REF(Signal* signal)
{
  WaitGCPRef ref = *(WaitGCPRef*)signal->getDataPtr();

  SubscriptionRecPtr subRecPtr;
  c_theSubscriptions.getPtr(subRecPtr, ref.senderData);
  wait_gcp(signal, subRecPtr, 100);
}

void
Trix::execWAIT_GCP_CONF(Signal* signal)
{
  WaitGCPConf * conf = (WaitGCPConf*)signal->getDataPtr();
  
  SubscriptionRecPtr subRecPtr;
  c_theSubscriptions.getPtr(subRecPtr, conf->senderData);

  const Uint32 gci_hi = conf->gci_hi;
  const Uint32 gci_lo = conf->gci_lo;
  const Uint64 gci = gci_lo | (Uint64(gci_hi) << 32);
  
  if (gci > subRecPtr.p->m_gci)
  {
    jam();
    buildComplete(signal, subRecPtr);
  }
  else
  {
    jam();
    wait_gcp(signal, subRecPtr, 100);
  }
}

void Trix::buildComplete(Signal* signal, SubscriptionRecPtr subRecPtr)
{
  SubRemoveReq * const req = (SubRemoveReq*)signal->getDataPtrSend();
  req->senderRef       = reference();
  req->senderData      = subRecPtr.i;
  req->subscriptionId  = subRecPtr.p->subscriptionId;
  req->subscriptionKey = subRecPtr.p->subscriptionKey;
  sendSignal(SUMA_REF, GSN_SUB_REMOVE_REQ, signal,
             SubRemoveReq::SignalLength, JBB);
}

void Trix::buildFailed(Signal* signal, 
                       SubscriptionRecPtr subRecPtr, 
                       BuildIndxRef::ErrorCode errorCode)
{
  SubscriptionRecord* subRec = subRecPtr.p;

  subRec->errorCode = errorCode;
  // Continue accumulating since we currently cannot stop SUMA
  subRec->expectedConf--;
  checkParallelism(signal, subRec);
  if (subRec->expectedConf == 0)
    buildComplete(signal, subRecPtr);
}

void
Trix::execSUB_REMOVE_REF(Signal* signal){
  jamEntry();
  //@todo
  ndbrequire(false);
}

void
Trix::execSUB_REMOVE_CONF(Signal* signal){
  jamEntry();

  SubRemoveConf * const conf = (SubRemoveConf*)signal->getDataPtrSend();

  SubscriptionRecPtr subRecPtr;
  c_theSubscriptions.getPtr(subRecPtr, conf->senderData);

  if(subRecPtr.p->prepareId != RNIL){
    jam();

    UtilReleaseReq * const req = (UtilReleaseReq*)signal->getDataPtrSend();
    req->prepareId = subRecPtr.p->prepareId;
    req->senderData = subRecPtr.i;

    sendSignal(DBUTIL_REF, GSN_UTIL_RELEASE_REQ, signal, 
               UtilReleaseReq::SignalLength , JBB);  
    return;
  }
  
  {
    UtilReleaseConf * const conf = (UtilReleaseConf*)signal->getDataPtrSend();
    conf->senderData = subRecPtr.i;
    execUTIL_RELEASE_CONF(signal);
  }
}

void
Trix::execUTIL_RELEASE_REF(Signal* signal){
  jamEntry();
  ndbrequire(false);
}

void
Trix::execUTIL_RELEASE_CONF(Signal* signal){
  
  UtilReleaseConf * const conf = (UtilReleaseConf*)signal->getDataPtrSend();
  
  SubscriptionRecPtr subRecPtr;
  c_theSubscriptions.getPtr(subRecPtr, conf->senderData);
  
  switch(subRecPtr.p->requestType){
  case REORG_COPY:
  case REORG_DELETE:
    if (subRecPtr.p->errorCode == BuildIndxRef::NoError)
    {
      jam();
      // Build is complete, reply to original sender
      CopyDataImplConf* conf = (CopyDataImplConf*)signal->getDataPtrSend();
      conf->senderRef = reference(); //wl3600_todo ok?
      conf->senderData = subRecPtr.p->connectionPtr;

      sendSignal(subRecPtr.p->userReference, GSN_COPY_DATA_IMPL_CONF, signal,
                 CopyDataImplConf::SignalLength , JBB);

      infoEvent("%s table %u processed %llu rows",
                subRecPtr.p->requestType == REORG_COPY ?
                "reorg-copy" : "reorg-delete",
                subRecPtr.p->sourceTableId,
                subRecPtr.p->m_rows_processed);
    } else {
      jam();
      // Build failed, reply to original sender
      CopyDataImplRef* ref = (CopyDataImplRef*)signal->getDataPtrSend();
      ref->senderRef = reference();
      ref->senderData = subRecPtr.p->connectionPtr;
      ref->errorCode = subRecPtr.p->errorCode;

      sendSignal(subRecPtr.p->userReference, GSN_COPY_DATA_IMPL_REF, signal,
                 CopyDataImplRef::SignalLength , JBB);
    }
    break;
  case INDEX_BUILD:
    if (subRecPtr.p->errorCode == BuildIndxRef::NoError) {
      jam();
      // Build is complete, reply to original sender
      BuildIndxImplConf* buildIndxConf =
        (BuildIndxImplConf*)signal->getDataPtrSend();
      buildIndxConf->senderRef = reference(); //wl3600_todo ok?
      buildIndxConf->senderData = subRecPtr.p->connectionPtr;

      sendSignal(subRecPtr.p->userReference, GSN_BUILD_INDX_IMPL_CONF, signal,
                 BuildIndxConf::SignalLength , JBB);

      infoEvent("index-build table %u index: %u processed %llu rows",
                subRecPtr.p->sourceTableId,
                subRecPtr.p->targetTableId,
                subRecPtr.p->m_rows_processed);
    } else {
      jam();
      // Build failed, reply to original sender
      BuildIndxImplRef* buildIndxRef =
        (BuildIndxImplRef*)signal->getDataPtrSend();
      buildIndxRef->senderRef = reference();
      buildIndxRef->senderData = subRecPtr.p->connectionPtr;
      buildIndxRef->errorCode = subRecPtr.p->errorCode;

      sendSignal(subRecPtr.p->userReference, GSN_BUILD_INDX_IMPL_REF, signal,
                 BuildIndxRef::SignalLength , JBB);
    }
    break;
  case STAT_UTIL:
    ndbrequire(subRecPtr.p->errorCode == BuildIndxRef::NoError);
    statUtilReleaseConf(signal, subRecPtr.p->m_statPtrI);
    return;
  case STAT_CLEAN:
    {
      subRecPtr.p->prepareId = RNIL;
      StatOp& stat = statOpGetPtr(subRecPtr.p->m_statPtrI);
      statCleanRelease(signal, stat);
    }
    return;
  case STAT_SCAN:
    {
      subRecPtr.p->prepareId = RNIL;
      StatOp& stat = statOpGetPtr(subRecPtr.p->m_statPtrI);
      statScanRelease(signal, stat);
    }
    return;
  }

  // Release subscription record
  subRecPtr.p->attributeOrder.release();
  c_theSubscriptions.release(subRecPtr.i);
}

void Trix::checkParallelism(Signal* signal, SubscriptionRecord* subRec)
{
  if ((subRec->pendingSubSyncContinueConf) &&
      (subRec->expectedConf == 1)) {
    jam();
    SubSyncContinueConf  * subSyncContinueConf = 
      (SubSyncContinueConf *) signal->getDataPtrSend();
    subSyncContinueConf->subscriptionId = subRec->subscriptionId;
    subSyncContinueConf->subscriptionKey = subRec->subscriptionKey;
    subSyncContinueConf->senderData = subRec->syncPtr;
    sendSignal(SUMA_REF, GSN_SUB_SYNC_CONTINUE_CONF, signal, 
               SubSyncContinueConf::SignalLength , JBB);  
    subRec->pendingSubSyncContinueConf = false;
    return;
  }
}

// CopyData
void
Trix::execCOPY_DATA_IMPL_REQ(Signal* signal)
{
  jamEntry();

  const CopyDataImplReq reqData = *(const CopyDataImplReq*)signal->getDataPtr();
  const CopyDataImplReq *req = &reqData;

  // Seize a subscription record
  SubscriptionRecPtr subRecPtr;
  SectionHandle handle(this, signal);

  if (!c_theSubscriptions.seize(subRecPtr))
  {
    jam();
    // Failed to allocate subscription record
    releaseSections(handle);

    CopyDataImplRef* ref = (CopyDataRef*)signal->getDataPtrSend();

    ref->errorCode = -1; // XXX CopyDataImplRef::AllocationFailure;
    ref->senderData = req->senderData;
    ref->transId = req->transId;
    sendSignal(req->senderRef, GSN_COPY_DATA_IMPL_REF, signal,
               CopyDataImplRef::SignalLength, JBB);
    return;
  }

  SubscriptionRecord* subRec = subRecPtr.p;
  subRec->errorCode = BuildIndxRef::NoError;
  subRec->userReference = req->senderRef;
  subRec->connectionPtr = req->senderData;
  subRec->schemaTransId = req->transId;
  subRec->subscriptionId = rand();
  subRec->subscriptionKey = rand();
  subRec->indexType = RNIL;
  subRec->sourceTableId = req->srcTableId;
  subRec->targetTableId = req->dstTableId;
  subRec->parallelism = 16;
  subRec->expectedConf = 0;
  subRec->subscriptionCreated = false;
  subRec->pendingSubSyncContinueConf = false;
  subRec->prepareId = req->transId;
  subRec->fragCount = req->srcFragments;
  subRec->fragId = ZNIL;
  subRec->m_rows_processed = 0;
  subRec->m_flags = SubscriptionRecord::RF_WAIT_GCP; // Todo make configurable
  subRec->m_gci = 0;
  switch(req->requestType){
  case CopyDataImplReq::ReorgCopy:
    jam();
    subRec->requestType = REORG_COPY;
    break;
  case CopyDataImplReq::ReorgDelete:
    subRec->requestType = REORG_DELETE;
    break;
  default:
    jamLine(req->requestType);
    ndbrequire(false);
  }

  if (req->requestInfo & CopyDataReq::TupOrder)
  {
    jam();
    subRec->m_flags |= SubscriptionRecord::RF_TUP_ORDER;
  }

  // Get column order segments
  Uint32 noOfSections = handle.m_cnt;
  if (noOfSections > 0) {
    jam();
    SegmentedSectionPtr ptr;
    handle.getSection(ptr, 0);
    append(subRec->attributeOrder, ptr, getSectionSegmentPool());
    subRec->noOfIndexColumns = ptr.sz;
  }

  if (noOfSections > 1) {
    jam();
    SegmentedSectionPtr ptr;
    handle.getSection(ptr, 1);
    append(subRec->attributeOrder, ptr, getSectionSegmentPool());
    subRec->noOfKeyColumns = ptr.sz;
  }

  releaseSections(handle);
  {
    UtilPrepareReq * utilPrepareReq =
      (UtilPrepareReq *)signal->getDataPtrSend();

    utilPrepareReq->senderRef = reference();
    utilPrepareReq->senderData = subRecPtr.i;
    utilPrepareReq->schemaTransId = subRec->schemaTransId;

    const Uint32 pageSizeInWords = 128;
    Uint32 propPage[pageSizeInWords];
    LinearWriter w(&propPage[0],128);
    w.first();
    w.add(UtilPrepareReq::NoOfOperations, 1);
    if (subRec->requestType == REORG_COPY)
    {
      w.add(UtilPrepareReq::OperationType, UtilPrepareReq::Write);
    }
    else
    {
      w.add(UtilPrepareReq::OperationType, UtilPrepareReq::Delete);
    }
    w.add(UtilPrepareReq::ScanTakeOverInd, 1);
    w.add(UtilPrepareReq::ReorgInd, 1);
    w.add(UtilPrepareReq::TableId, subRec->targetTableId);

    AttrOrderBuffer::DataBufferIterator iter;
    ndbrequire(subRec->attributeOrder.first(iter));
    
    for(Uint32 i = 0; i < subRec->noOfIndexColumns; i++)
    {
      w.add(UtilPrepareReq::AttributeId, * iter.data);
      subRec->attributeOrder.next(iter);
    }
    
    struct LinearSectionPtr sectionsPtr[UtilPrepareReq::NoOfSections];
    sectionsPtr[UtilPrepareReq::PROPERTIES_SECTION].p = propPage;
    sectionsPtr[UtilPrepareReq::PROPERTIES_SECTION].sz = w.getWordsUsed();
    sendSignal(DBUTIL_REF, GSN_UTIL_PREPARE_REQ, signal,
               UtilPrepareReq::SignalLength, JBB,
               sectionsPtr, UtilPrepareReq::NoOfSections);
  }
}

// index stats

Trix::StatOp&
Trix::statOpGetPtr(Uint32 statPtrI)
{
  ndbrequire(statPtrI != RNIL);
  return *c_statOpPool.getPtr(statPtrI);
}

bool
Trix::statOpSeize(Uint32& statPtrI)
{
  StatOpPtr statPtr;
  if (ERROR_INSERTED(18001) ||
      !c_statOpPool.seize(statPtr))
  {
    jam();
    D("statOpSeize: seize statOp failed");
    return false;
  }
#ifdef VM_TRACE
  memset(statPtr.p, 0xf3, sizeof(*statPtr.p));
#endif
  new (statPtr.p) StatOp;
  statPtrI = statPtr.i;
  StatOp& stat = statOpGetPtr(statPtrI);
  stat.m_ownPtrI = statPtrI;

  SubscriptionRecPtr subRecPtr;
  if (ERROR_INSERTED(18002) ||
      !c_theSubscriptions.seize(subRecPtr))
  {
    jam();
    c_statOpPool.release(statPtr);
    D("statOpSeize: seize subRec failed");
    return false;
  }
  SubscriptionRecord* subRec = subRecPtr.p;
  subRec->m_statPtrI = stat.m_ownPtrI;
  stat.m_subRecPtrI = subRecPtr.i;

  D("statOpSeize" << V(statPtrI) << V(subRecPtr.i));
  return true;
}

void
Trix::statOpRelease(StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  D("statOpRelease" << V(stat));

  if (stat.m_subRecPtrI != RNIL)
  {
    jam();
    SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
    ndbrequire(subRec->prepareId == RNIL);
    subRec->attributeOrder.release();
    c_theSubscriptions.release(stat.m_subRecPtrI);
    stat.m_subRecPtrI = RNIL;
  }
  ndbrequire(util.m_prepareId == RNIL);
  c_statOpPool.release(stat.m_ownPtrI);
}

void
Trix::execINDEX_STAT_IMPL_REQ(Signal* signal)
{
  jamEntry();
  const IndexStatImplReq* req = (const IndexStatImplReq*)signal->getDataPtr();

  Uint32 statPtrI = RNIL;
  if (!statOpSeize(statPtrI))
  {
    jam();
    statOpRef(signal, req, IndexStatRef::NoFreeStatOp, __LINE__);
    return;
  }
  StatOp& stat = statOpGetPtr(statPtrI);
  stat.m_req = *req;
  stat.m_requestType = req->requestType;

  // set request name for cluster log message
  switch (stat.m_requestType) {
  case IndexStatReq::RT_CLEAN_NEW:
    jam();
    stat.m_requestName = "clean new";
    break;
  case IndexStatReq::RT_CLEAN_OLD:
    jam();
    stat.m_requestName = "clean old";
    break;
  case IndexStatReq::RT_CLEAN_ALL:
    jam();
    stat.m_requestName = "clean all";
    break;
  case IndexStatReq::RT_SCAN_FRAG:
    jam();
    stat.m_requestName = "scan frag";
    break;
  case IndexStatReq::RT_DROP_HEAD:
    jam();
    stat.m_requestName = "drop head";
    break;
  default:
    ndbrequire(false);
    break;
  }

  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  subRec->prepareId = RNIL;
  subRec->errorCode = BuildIndxRef::NoError;

  // sys tables are not recreated so do this only once
  if (!c_statGetMetaDone)
  {
    jam();
    statMetaGetHead(signal, stat);
    return;
  }
  statGetMetaDone(signal, stat);
}

// sys tables metadata

const Trix::SysColumn
Trix::g_statMetaHead_column[] = {
  { 0, "index_id",
    true
  },
  { 1, "index_version",
    true
  },
  { 2, "table_id",
    false
  },
  { 3, "frag_count",
    false
  },
  { 4, "value_format",
    false
  },
  { 5, "sample_version",
    false
  },
  { 6, "load_time",
    false
  },
  { 7, "sample_count",
    false
  },
  { 8, "key_bytes",
    false
  }
};

const Trix::SysColumn
Trix::g_statMetaSample_column[] = {
  { 0, "index_id",
    true
  },
  { 1, "index_version",
    true
  },
  { 2, "sample_version",
    true
  },
  { 3, "stat_key",
    true
  },
  { 4, "stat_value",
    false
  }
};

const Trix::SysTable
Trix::g_statMetaHead = {
  NDB_INDEX_STAT_DB "/" NDB_INDEX_STAT_SCHEMA "/" NDB_INDEX_STAT_HEAD_TABLE,
  ~(Uint32)0,
  sizeof(g_statMetaHead_column)/sizeof(g_statMetaHead_column[0]),
  g_statMetaHead_column
};

const Trix::SysTable
Trix::g_statMetaSample = {
  NDB_INDEX_STAT_DB "/" NDB_INDEX_STAT_SCHEMA "/" NDB_INDEX_STAT_SAMPLE_TABLE,
  ~(Uint32)0,
  sizeof(g_statMetaSample_column)/sizeof(g_statMetaSample_column[0]),
  g_statMetaSample_column
};

const Trix::SysIndex
Trix::g_statMetaSampleX1 = {
  // indexes are always in "sys"
  "sys" "/" NDB_INDEX_STAT_SCHEMA "/%u/" NDB_INDEX_STAT_SAMPLE_INDEX1,
  ~(Uint32)0,
  ~(Uint32)0
};

void
Trix::statMetaGetHead(Signal* signal, StatOp& stat)
{
  D("statMetaGetHead" << V(stat));
  StatOp::Meta& meta = stat.m_meta;
  meta.m_cb.m_callbackFunction = safe_cast(&Trix::statMetaGetHeadCB);
  meta.m_cb.m_callbackData = stat.m_ownPtrI;
  const char* name = g_statMetaHead.name;
  sendGetTabInfoReq(signal, stat, name);
}

void
Trix::statMetaGetHeadCB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statMetaGetHeadCB" << V(stat) << V(ret));
  StatOp::Meta& meta = stat.m_meta;
  if (ret != 0)
  {
    jam();
    Uint32 supress[] = { GetTabInfoRef::TableNotDefined, 0 };
    statOpError(signal, stat, ret, __LINE__, supress);
    return;
  }
  g_statMetaHead.tableId = meta.m_conf.tableId;
  statMetaGetSample(signal, stat);
}

void
Trix::statMetaGetSample(Signal* signal, StatOp& stat)
{
  D("statMetaGetSample" << V(stat));
  StatOp::Meta& meta = stat.m_meta;
  meta.m_cb.m_callbackFunction = safe_cast(&Trix::statMetaGetSampleCB);
  meta.m_cb.m_callbackData = stat.m_ownPtrI;
  const char* name = g_statMetaSample.name;
  sendGetTabInfoReq(signal, stat, name);
}

void
Trix::statMetaGetSampleCB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statMetaGetSampleCB" << V(stat) << V(ret));
  StatOp::Meta& meta = stat.m_meta;
  if (ret != 0)
  {
    jam();
    statOpError(signal, stat, ret, __LINE__);
    return;
  }
  g_statMetaSample.tableId = meta.m_conf.tableId;
  statMetaGetSampleX1(signal, stat);
}

void
Trix::statMetaGetSampleX1(Signal* signal, StatOp& stat)
{
  D("statMetaGetSampleX1" << V(stat));
  StatOp::Meta& meta = stat.m_meta;
  meta.m_cb.m_callbackFunction = safe_cast(&Trix::statMetaGetSampleX1CB);
  meta.m_cb.m_callbackData = stat.m_ownPtrI;
  const char* name_fmt = g_statMetaSampleX1.name;
  char name[MAX_TAB_NAME_SIZE];
  BaseString::snprintf(name, sizeof(name), name_fmt, g_statMetaSample.tableId);
  sendGetTabInfoReq(signal, stat, name);
}

void
Trix::statMetaGetSampleX1CB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statMetaGetSampleX1CB" << V(stat) << V(ret));
  StatOp::Meta& meta = stat.m_meta;
  if (ret != 0)
  {
    jam();
    statOpError(signal, stat, ret, __LINE__);
    return;
  }
  g_statMetaSampleX1.tableId = g_statMetaSample.tableId;
  g_statMetaSampleX1.indexId = meta.m_conf.tableId;
  statGetMetaDone(signal, stat);
}

void
Trix::sendGetTabInfoReq(Signal* signal, StatOp& stat, const char* name)
{
  D("sendGetTabInfoReq" << V(stat) << V(name));
  GetTabInfoReq* req = (GetTabInfoReq*)signal->getDataPtrSend();

  Uint32 name_len = (Uint32)strlen(name) + 1;
  Uint32 name_len_words = (name_len + 3 ) / 4;
  Uint32 name_buf[32];
  ndbrequire(name_len_words <= 32);
  memset(name_buf, 0, sizeof(name_buf));
  memcpy(name_buf, name, name_len);

  req->senderData = stat.m_ownPtrI;
  req->senderRef = reference();
  req->requestType = GetTabInfoReq::RequestByName |
                     GetTabInfoReq::LongSignalConf;
  req->tableNameLen = name_len;
  req->schemaTransId = 0;
  LinearSectionPtr ptr[3];
  ptr[0].p = name_buf;
  ptr[0].sz = name_len_words;
  sendSignal(DBDICT_REF, GSN_GET_TABINFOREQ,
             signal, GetTabInfoReq::SignalLength, JBB, ptr, 1);
}

void
Trix::execGET_TABINFO_CONF(Signal* signal)
{
  jamEntry();
  if (!assembleFragments(signal)) {
    jam();
    return;
  }
  const GetTabInfoConf* conf = (const GetTabInfoConf*)signal->getDataPtr();
  StatOp& stat = statOpGetPtr(conf->senderData);
  D("execGET_TABINFO_CONF" << V(stat));
  StatOp::Meta& meta = stat.m_meta;
  meta.m_conf = *conf;

  // do not need DICTTABINFO
  SectionHandle handle(this, signal);
  releaseSections(handle);

  execute(signal, meta.m_cb, 0);
}

void
Trix::execGET_TABINFO_REF(Signal* signal)
{
  jamEntry();
  const GetTabInfoRef* ref = (const GetTabInfoRef*)signal->getDataPtr();
  StatOp& stat = statOpGetPtr(ref->senderData);
  D("execGET_TABINFO_REF" << V(stat));
  StatOp::Meta& meta = stat.m_meta;

  ndbrequire(ref->errorCode != 0);
  execute(signal, meta.m_cb, ref->errorCode);
}

// continue after metadata retrieval

void
Trix::statGetMetaDone(Signal* signal, StatOp& stat)
{
  const IndexStatImplReq* req = &stat.m_req;
  StatOp::Data& data = stat.m_data;
  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  D("statGetMetaDone" << V(stat));

  // c_statGetMetaDone = true;

  subRec->requestType = STAT_UTIL;
  // fill in constant part
  ndbrequire(req->fragCount != 0);
  data.m_indexId = req->indexId;
  data.m_indexVersion = req->indexVersion;
  data.m_fragCount = req->fragCount;
  statHeadRead(signal, stat);
}

// head table ops

void
Trix::statHeadRead(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statHeadRead" << V(stat));

  util.m_not_found = false;
  util.m_cb.m_callbackFunction = safe_cast(&Trix::statHeadReadCB);
  util.m_cb.m_callbackData = stat.m_ownPtrI;
  send.m_sysTable = &g_statMetaHead;
  send.m_operationType = UtilPrepareReq::Read;
  statUtilPrepare(signal, stat);
}

void
Trix::statHeadReadCB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  StatOp::Data& data = stat.m_data;
  StatOp::Util& util = stat.m_util;
  D("statHeadReadCB" << V(stat) << V(ret));

  ndbrequire(ret == 0);
  data.m_head_found = !util.m_not_found;
  statReadHeadDone(signal, stat);
}

void
Trix::statHeadInsert(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statHeadInsert" << V(stat));

  util.m_cb.m_callbackFunction = safe_cast(&Trix::statHeadInsertCB);
  util.m_cb.m_callbackData = stat.m_ownPtrI;
  send.m_sysTable = &g_statMetaHead;
  send.m_operationType = UtilPrepareReq::Insert;
  statUtilPrepare(signal, stat);
}

void
Trix::statHeadInsertCB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statHeadInsertCB" << V(stat) << V(ret));

  ndbrequire(ret == 0);
  statInsertHeadDone(signal, stat);
}

void
Trix::statHeadUpdate(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statHeadUpdate" << V(stat));

  util.m_cb.m_callbackFunction = safe_cast(&Trix::statHeadUpdateCB);
  util.m_cb.m_callbackData = stat.m_ownPtrI;
  send.m_sysTable = &g_statMetaHead;
  send.m_operationType = UtilPrepareReq::Update;
  statUtilPrepare(signal, stat);
}

void
Trix::statHeadUpdateCB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statHeadUpdateCB" << V(stat) << V(ret));

  ndbrequire(ret == 0);
  statUpdateHeadDone(signal, stat);
}

void
Trix::statHeadDelete(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statHeadDelete" << V(stat));

  util.m_cb.m_callbackFunction = safe_cast(&Trix::statHeadDeleteCB);
  util.m_cb.m_callbackData = stat.m_ownPtrI;
  send.m_sysTable = &g_statMetaHead;
  send.m_operationType = UtilPrepareReq::Delete;
  statUtilPrepare(signal, stat);
}

void
Trix::statHeadDeleteCB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statHeadDeleteCB" << V(stat) << V(ret));

  ndbrequire(ret == 0);
  statDeleteHeadDone(signal, stat);
}

// util (PK ops, only HEAD for now)

void
Trix::statUtilPrepare(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  D("statUtilPrepare" << V(stat));

  util.m_prepareId = RNIL;
  statSendPrepare(signal, stat);
}

void
Trix::statUtilPrepareConf(Signal* signal, Uint32 statPtrI)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statUtilPrepareConf" << V(stat));

  const UtilPrepareConf* utilConf =
    (const UtilPrepareConf*)signal->getDataPtr();
  util.m_prepareId = utilConf->prepareId;

  const Uint32 ot = send.m_operationType;
  if ((ERROR_INSERTED(18011) && ot == UtilPrepareReq::Read) ||
      (ERROR_INSERTED(18012) && ot != UtilPrepareReq::Read))
  {
    jam();
    CLEAR_ERROR_INSERT_VALUE;
    UtilExecuteRef* utilRef =
      (UtilExecuteRef*)signal->getDataPtrSend();
    utilRef->senderData = stat.m_ownPtrI;
    utilRef->errorCode = UtilExecuteRef::AllocationError;
    utilRef->TCErrorCode = 0;
    sendSignal(reference(), GSN_UTIL_EXECUTE_REF,
               signal, UtilExecuteRef::SignalLength, JBB);
    return;
  }

  statUtilExecute(signal, stat);
}

void
Trix::statUtilPrepareRef(Signal* signal, Uint32 statPtrI)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statUtilPrepareRef" << V(stat));

  const UtilPrepareRef* utilRef =
    (const UtilPrepareRef*)signal->getDataPtr();
  Uint32 errorCode = utilRef->errorCode;
  ndbrequire(errorCode != 0);

  switch (errorCode) {
  case UtilPrepareRef::PREPARE_SEIZE_ERROR:
  case UtilPrepareRef::PREPARE_PAGES_SEIZE_ERROR:
  case UtilPrepareRef::PREPARED_OPERATION_SEIZE_ERROR:
    errorCode = IndexStatRef::BusyUtilPrepare;
    break;
  case UtilPrepareRef::DICT_TAB_INFO_ERROR:
    errorCode = IndexStatRef::InvalidSysTable;
    break;
  case UtilPrepareRef::MISSING_PROPERTIES_SECTION:
  default:
    ndbrequire(false);
    break;
  }
  statOpError(signal, stat, errorCode, __LINE__);
}

void
Trix::statUtilExecute(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statUtilExecute" << V(stat));

  send.m_prepareId = util.m_prepareId;
  statSendExecute(signal, stat);
}

void
Trix::statUtilExecuteConf(Signal* signal, Uint32 statPtrI)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  StatOp::Attr& attr = stat.m_attr;
  StatOp::Send& send = stat.m_send;
  D("statUtilExecuteConf" << V(stat));

  if (send.m_operationType == UtilPrepareReq::Read)
  {
    jam();
    SectionHandle handle(this, signal);
    Uint32 rattr[20];
    Uint32 rdata[2048];
    attr.m_attr = rattr;
    attr.m_attrMax = 20;
    attr.m_attrSize = 0;
    attr.m_data = rdata;
    attr.m_dataMax = 2048;
    attr.m_dataSize = 0;
    {
      SegmentedSectionPtr ssPtr;
      handle.getSection(ssPtr, 0);
      ::copy(rattr, ssPtr);
    }
    {
      SegmentedSectionPtr ssPtr;
      handle.getSection(ssPtr, 1);
      ::copy(rdata, ssPtr);
    }
    releaseSections(handle);

    const SysTable& sysTable = *send.m_sysTable;
    for (Uint32 i = 0; i < sysTable.columnCount; i++)
    {
      jam();
      statDataIn(stat, i);
    }
  }

  statUtilRelease(signal, stat);
}

void
Trix::statUtilExecuteRef(Signal* signal, Uint32 statPtrI)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statUtilExecuteRef" << V(stat));

  const UtilExecuteRef* utilRef =
    (const UtilExecuteRef*)signal->getDataPtr();
  Uint32 errorCode = utilRef->errorCode;
  ndbrequire(errorCode != 0);

  switch (errorCode) {
  case UtilExecuteRef::TCError:
    errorCode = utilRef->TCErrorCode;
    ndbrequire(errorCode != 0);
    if (send.m_operationType == UtilPrepareReq::Read &&
        errorCode == ZNOT_FOUND)
    {
      jam();
      util.m_not_found = true;
      errorCode = 0;
    }
    break;
  case UtilExecuteRef::AllocationError:
    errorCode = IndexStatRef::BusyUtilExecute;
    break;
  default:
    ndbrequire(false);
    break;
  }

  if (errorCode != 0)
  {
    jam();
    statOpError(signal, stat, errorCode, __LINE__);
    return;
  }
  statUtilRelease(signal, stat);
}

void
Trix::statUtilRelease(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  StatOp::Send& send = stat.m_send;
  D("statUtilRelease" << V(stat));

  send.m_prepareId = util.m_prepareId;
  statSendRelease(signal, stat);
}

void
Trix::statUtilReleaseConf(Signal* signal, Uint32 statPtrI)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  StatOp::Util& util = stat.m_util;
  D("statUtilReleaseConf" << V(stat));

  util.m_prepareId = RNIL;
  execute(signal, util.m_cb, 0);
}

// continue after head table ops

void
Trix::statReadHeadDone(Signal* signal, StatOp& stat)
{
  //UNUSED StatOp::Data& data = stat.m_data;
  D("statReadHeadDone" << V(stat));

  switch (stat.m_requestType) {
  case IndexStatReq::RT_CLEAN_NEW:
    jam();
  case IndexStatReq::RT_CLEAN_OLD:
    jam();
  case IndexStatReq::RT_CLEAN_ALL:
    jam();
    statCleanBegin(signal, stat);
    break;

  case IndexStatReq::RT_SCAN_FRAG:
    jam();
    statScanBegin(signal, stat);
    break;

  case IndexStatReq::RT_DROP_HEAD:
    jam();
    statDropBegin(signal, stat);
    break;

  default:
    ndbrequire(false);
    break;
  }
}

void
Trix::statInsertHeadDone(Signal* signal, StatOp& stat)
{
  D("statInsertHeadDone" << V(stat));

  switch (stat.m_requestType) {
  case IndexStatReq::RT_SCAN_FRAG:
    jam();
    statScanEnd(signal, stat);
    break;
  default:
    ndbrequire(false);
    break;
  }
}

void
Trix::statUpdateHeadDone(Signal* signal, StatOp& stat)
{
  D("statUpdateHeadDone" << V(stat));

  switch (stat.m_requestType) {
  case IndexStatReq::RT_SCAN_FRAG:
    jam();
    statScanEnd(signal, stat);
    break;
  default:
    ndbrequire(false);
    break;
  }
}

void
Trix::statDeleteHeadDone(Signal* signal, StatOp& stat)
{
  D("statDeleteHeadDone" << V(stat));

  switch (stat.m_requestType) {
  case IndexStatReq::RT_DROP_HEAD:
    jam();
    statDropEnd(signal, stat);
    break;
  default:
    ndbrequire(false);
    break;
  }
}

// clean

void
Trix::statCleanBegin(Signal* signal, StatOp& stat)
{
  const IndexStatImplReq* req = &stat.m_req;
  StatOp::Data& data = stat.m_data;
  D("statCleanBegin" << V(stat));

  if (data.m_head_found == true)
  {
    jam();
    if (data.m_tableId != req->tableId &&
        stat.m_requestType != IndexStatReq::RT_CLEAN_ALL)
    {
      jam();
      // must run ndb_index_stat --drop
      statOpError(signal, stat, IndexStatRef::InvalidSysTableData, __LINE__);
      return;
    }
  }
  else
  {
    if (stat.m_requestType != IndexStatReq::RT_CLEAN_ALL)
    {
      jam();
      // happens normally on first stats scan
      stat.m_requestType = IndexStatReq::RT_CLEAN_ALL;
    }
  }
  statCleanPrepare(signal, stat);
}

void
Trix::statCleanPrepare(Signal* signal, StatOp& stat)
{
  const IndexStatImplReq* req = &stat.m_req;
  StatOp::Data& data = stat.m_data;
  StatOp::Clean& clean = stat.m_clean;
  StatOp::Send& send = stat.m_send;
  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  D("statCleanPrepare" << V(stat));

  // count of deleted samples is just for info
  clean.m_cleanCount = 0;

  const Uint32 ao_list[] = {
    0,  // INDEX_ID
    1,  // INDEX_VERSION
    2,  // SAMPLE_VERSION
    3   // STAT_KEY
  };
  const Uint32 ao_size = sizeof(ao_list)/sizeof(ao_list[0]);

  ndbrequire(req->fragId == ZNIL);
  subRec->m_flags = 0;
  subRec->requestType = STAT_CLEAN;
  subRec->schemaTransId = req->transId;
  subRec->userReference = 0; // not used
  subRec->connectionPtr = RNIL;
  subRec->subscriptionId = rand();
  subRec->subscriptionKey = rand();
  subRec->prepareId = RNIL;
  subRec->indexType = 0; // not used
  subRec->sourceTableId = g_statMetaSampleX1.indexId;
  subRec->targetTableId = RNIL;
  subRec->noOfIndexColumns = ao_size;
  subRec->noOfKeyColumns = 0;
  subRec->parallelism = 16;
  subRec->fragCount = 0;
  subRec->fragId = ZNIL;
  subRec->syncPtr = RNIL;
  subRec->errorCode = BuildIndxRef::NoError;
  subRec->subscriptionCreated = false;
  subRec->pendingSubSyncContinueConf = false;
  subRec->expectedConf = 0;
  subRec->m_rows_processed = 0;
  subRec->m_gci = 0;

  AttrOrderBuffer& ao_buf = subRec->attributeOrder;
  ndbrequire(ao_buf.isEmpty());
  ao_buf.append(ao_list, ao_size);

  // create TUX bounds
  clean.m_bound[0] = TuxBoundInfo::BoundEQ;
  clean.m_bound[1] = AttributeHeader(0, 4).m_value;
  clean.m_bound[2] = data.m_indexId;
  clean.m_bound[3] = TuxBoundInfo::BoundEQ;
  clean.m_bound[4] = AttributeHeader(1, 4).m_value;
  clean.m_bound[5] = data.m_indexVersion;
  switch (stat.m_requestType) {
  case IndexStatReq::RT_CLEAN_NEW:
    D("statCleanPrepare delete sample versions > " << data.m_sampleVersion);
    clean.m_bound[6] = TuxBoundInfo::BoundLT;
    clean.m_bound[7] = AttributeHeader(2, 4).m_value;
    clean.m_bound[8] = data.m_sampleVersion;
    clean.m_boundCount = 3;
    break;
  case IndexStatReq::RT_CLEAN_OLD:
    D("statCleanPrepare delete sample versions < " << data.m_sampleVersion);
    clean.m_bound[6] = TuxBoundInfo::BoundGT;
    clean.m_bound[7] = AttributeHeader(2, 4).m_value;
    clean.m_bound[8] = data.m_sampleVersion;
    clean.m_boundCount = 3;
    break;
  case IndexStatReq::RT_CLEAN_ALL:
    D("statCleanPrepare delete all sample versions");
    clean.m_boundCount = 2;
    break;
  default:
    ndbrequire(false);
    break;
  }
  clean.m_boundSize = 3 * clean.m_boundCount;

  // TRIX traps the CONF
  send.m_sysTable = &g_statMetaSample;
  send.m_operationType = UtilPrepareReq::Delete;
  statSendPrepare(signal, stat);
}

void
Trix::statCleanExecute(Signal* signal, StatOp& stat)
{
  StatOp::Data& data = stat.m_data;
  StatOp::Send& send = stat.m_send;
  StatOp::Clean& clean = stat.m_clean;
  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  D("statCleanExecute" << V(stat));

  SectionHandle handle(this, signal);
  ndbrequire(handle.m_cnt == 2);

  // ATTR_INFO
  AttributeHeader ah[4];
  SegmentedSectionPtr ptr0;
  handle.getSection(ptr0, SubTableData::ATTR_INFO);
  ndbrequire(ptr0.sz == 4);
  ::copy((Uint32*)ah, ptr0);
  ndbrequire(ah[0].getAttributeId() == 0 && ah[0].getDataSize() == 1);
  ndbrequire(ah[1].getAttributeId() == 1 && ah[1].getDataSize() == 1);
  ndbrequire(ah[2].getAttributeId() == 2 && ah[2].getDataSize() == 1);
  // read via TUP rounds bytes to words
  const Uint32 kz = ah[3].getDataSize();
  ndbrequire(ah[3].getAttributeId() == 3 && kz != 0);

  // AFTER_VALUES
  const Uint32 avmax = 3 + MAX_INDEX_STAT_KEY_SIZE;
  Uint32 av[avmax];
  SegmentedSectionPtr ptr1;
  handle.getSection(ptr1, SubTableData::AFTER_VALUES);
  ndbrequire(ptr1.sz <= avmax);
  ::copy(av, ptr1);
  ndbrequire(data.m_indexId == av[0]);
  ndbrequire(data.m_indexVersion == av[1]);
  data.m_sampleVersion = av[2];
  data.m_statKey = &av[3];
  const char* kp = (const char*)data.m_statKey;
  const Uint32 kb = kp[0] + (kp[1] << 8);
  // key is not empty
  ndbrequire(kb != 0);
  ndbrequire(kz == ((2 + kb) + 3) / 4);

  clean.m_cleanCount++;
  releaseSections(handle);

  const Uint32 rt = stat.m_requestType;
  if ((ERROR_INSERTED(18021) && rt == IndexStatReq::RT_CLEAN_NEW) ||
      (ERROR_INSERTED(18022) && rt == IndexStatReq::RT_CLEAN_OLD) ||
      (ERROR_INSERTED(18023) && rt == IndexStatReq::RT_CLEAN_ALL))
  {
    jam();
    CLEAR_ERROR_INSERT_VALUE;
    UtilExecuteRef* utilRef =
      (UtilExecuteRef*)signal->getDataPtrSend();
    utilRef->senderData = stat.m_ownPtrI;
    utilRef->errorCode = UtilExecuteRef::TCError;
    utilRef->TCErrorCode = 626;
    sendSignal(reference(), GSN_UTIL_EXECUTE_REF,
               signal, UtilExecuteRef::SignalLength, JBB);
    subRec->expectedConf++;
    return;
  }

  // TRIX traps the CONF
  send.m_sysTable = &g_statMetaSample;
  send.m_operationType = UtilPrepareReq::Delete;
  send.m_prepareId = subRec->prepareId;
  subRec->expectedConf++;
  statSendExecute(signal, stat);
}

void
Trix::statCleanRelease(Signal* signal, StatOp& stat)
{
  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  D("statCleanRelease" << V(stat) << V(subRec->errorCode));

  if (subRec->errorCode != 0)
  {
    jam();
    statOpError(signal, stat, subRec->errorCode, __LINE__);
    return;
  }
  statCleanEnd(signal, stat);
}

void
Trix::statCleanEnd(Signal* signal, StatOp& stat)
{
  D("statCleanEnd" << V(stat));
  statOpSuccess(signal, stat);
}

// scan

void
Trix::statScanBegin(Signal* signal, StatOp& stat)
{
  const IndexStatImplReq* req = &stat.m_req;
  StatOp::Data& data = stat.m_data;
  D("statScanBegin" << V(stat));

  if (data.m_head_found == true &&
      data.m_tableId != req->tableId)
  {
    jam();
    statOpError(signal, stat, IndexStatRef::InvalidSysTableData, __LINE__);
    return;
  }
  data.m_tableId = req->tableId;
  statScanPrepare(signal, stat);
}

void
Trix::statScanPrepare(Signal* signal, StatOp& stat)
{
  const IndexStatImplReq* req = &stat.m_req;
  StatOp::Data& data = stat.m_data;
  StatOp::Scan& scan = stat.m_scan;
  StatOp::Send& send = stat.m_send;
  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  D("statScanPrepare" << V(stat));

  // update sample version prior to scan
  if (data.m_head_found == false)
    data.m_sampleVersion = 0;
  data.m_sampleVersion += 1;

  // zero totals
  scan.m_sampleCount = 0;
  scan.m_keyBytes = 0;

  const Uint32 ao_list[] = {
    AttributeHeader::INDEX_STAT_KEY,
    AttributeHeader::INDEX_STAT_VALUE
  };
  const Uint32 ao_size = sizeof(ao_list)/sizeof(ao_list[0]);

  ndbrequire(req->fragId != ZNIL);
  subRec->m_flags = 0;
  subRec->requestType = STAT_SCAN;
  subRec->schemaTransId = req->transId;
  subRec->userReference = 0; // not used
  subRec->connectionPtr = RNIL;
  subRec->subscriptionId = rand();
  subRec->subscriptionKey = rand();
  subRec->prepareId = RNIL;
  subRec->indexType = 0; // not used
  subRec->sourceTableId = data.m_indexId;
  subRec->targetTableId = RNIL;
  subRec->noOfIndexColumns = ao_size;
  subRec->noOfKeyColumns = 0;
  subRec->parallelism = 16;
  subRec->fragCount = 0; // XXX Suma currently checks all frags
  subRec->fragId = req->fragId;
  subRec->syncPtr = RNIL;
  subRec->errorCode = BuildIndxRef::NoError;
  subRec->subscriptionCreated = false;
  subRec->pendingSubSyncContinueConf = false;
  subRec->expectedConf = 0;
  subRec->m_rows_processed = 0;
  subRec->m_gci = 0;

  AttrOrderBuffer& ao_buf = subRec->attributeOrder;
  ndbrequire(ao_buf.isEmpty());
  ao_buf.append(ao_list, ao_size);

  // TRIX traps the CONF
  send.m_sysTable = &g_statMetaSample;
  send.m_operationType = UtilPrepareReq::Insert;
  statSendPrepare(signal, stat);
}

void
Trix::statScanExecute(Signal* signal, StatOp& stat)
{
  StatOp::Data& data = stat.m_data;
  StatOp::Scan& scan = stat.m_scan;
  StatOp::Send& send = stat.m_send;
  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  D("statScanExecute" << V(stat));

  SectionHandle handle(this, signal);
  ndbrequire(handle.m_cnt == 2);

  // ATTR_INFO
  AttributeHeader ah[2];
  SegmentedSectionPtr ptr0;
  handle.getSection(ptr0, SubTableData::ATTR_INFO);
  ndbrequire(ptr0.sz == 2);
  ::copy((Uint32*)ah, ptr0);
  ndbrequire(ah[0].getAttributeId() == AttributeHeader::INDEX_STAT_KEY);
  ndbrequire(ah[1].getAttributeId() == AttributeHeader::INDEX_STAT_VALUE);
  // read via TUP rounds bytes to words
  const Uint32 kz = ah[0].getDataSize();
  const Uint32 vz = ah[1].getDataSize();
  ndbrequire(kz != 0 && vz != 0);

  // AFTER_VALUES
  const Uint32 avmax = MAX_INDEX_STAT_KEY_SIZE + MAX_INDEX_STAT_VALUE_SIZE;
  Uint32 av[avmax];
  SegmentedSectionPtr ptr1;
  handle.getSection(ptr1, SubTableData::AFTER_VALUES);
  ndbrequire(ptr1.sz <= avmax);
  ::copy(av, ptr1);
  data.m_statKey = &av[0];
  data.m_statValue = &av[kz];
  const char* kp = (const char*)data.m_statKey;
  const char* vp = (const char*)data.m_statValue;
  const Uint32 kb = kp[0] + (kp[1] << 8);
  const Uint32 vb = vp[0] + (vp[1] << 8);
  // key and value are not empty
  ndbrequire(kb != 0 && vb != 0);
  ndbrequire(kz == ((2 + kb) + 3) / 4);
  ndbrequire(vz == ((2 + vb) + 3) / 4);

  scan.m_sampleCount++;
  scan.m_keyBytes += kb;
  releaseSections(handle);

  if (ERROR_INSERTED(18024))
  {
    jam();
    CLEAR_ERROR_INSERT_VALUE;
    UtilExecuteRef* utilRef =
      (UtilExecuteRef*)signal->getDataPtrSend();
    utilRef->senderData = stat.m_ownPtrI;
    utilRef->errorCode = UtilExecuteRef::TCError;
    utilRef->TCErrorCode = 630;
    sendSignal(reference(), GSN_UTIL_EXECUTE_REF,
               signal, UtilExecuteRef::SignalLength, JBB);
    subRec->expectedConf++;
    return;
  }

  // TRIX traps the CONF
  send.m_sysTable = &g_statMetaSample;
  send.m_operationType = UtilPrepareReq::Insert;
  send.m_prepareId = subRec->prepareId;
  subRec->expectedConf++;
  statSendExecute(signal, stat);
}

void
Trix::statScanRelease(Signal* signal, StatOp& stat)
{
  StatOp::Data& data = stat.m_data;
  StatOp::Scan& scan = stat.m_scan;
  SubscriptionRecord* subRec = c_theSubscriptions.getPtr(stat.m_subRecPtrI);
  D("statScanRelease" << V(stat) << V(subRec->errorCode));

  if (subRec->errorCode != 0)
  {
    jam();
    statOpError(signal, stat, subRec->errorCode, __LINE__);
    return;
  }
  subRec->requestType = STAT_UTIL;

  const Uint32 now = (Uint32)time(0);
  data.m_loadTime = now;
  data.m_sampleCount = scan.m_sampleCount;
  data.m_keyBytes = scan.m_keyBytes;
  data.m_valueFormat = MAX_INDEX_STAT_VALUE_FORMAT;

  if (data.m_head_found == false)
  {
    jam();
    statHeadInsert(signal, stat);
  }
  else
  {
    jam();
    statHeadUpdate(signal, stat);
  }
}

void
Trix::statScanEnd(Signal* signal, StatOp& stat)
{
  StatOp::Data& data = stat.m_data;
  const IndexStatImplReq* req = &stat.m_req;
  D("statScanEnd" << V(stat));

  /*
   * TRIX reports stats load time to TUX for proper stats monitoring.
   * Passing this via DBDICT RT_START_MON is not feasible.  For MT-LQH
   * we prefer DbtuxProxy to avoid introducing MT-LQH into TRIX.
   */

#if trix_index_stat_rep_to_tux_instance
  Uint32 instanceKey = getInstanceKey(req->indexId, req->fragId);
  BlockReference tuxRef = numberToRef(DBTUX, instanceKey, getOwnNodeId());
#else
  BlockReference tuxRef = DBTUX_REF;
#endif

  IndexStatRep* rep = (IndexStatRep*)signal->getDataPtrSend();
  rep->senderRef = reference();
  rep->senderData = 0;
  rep->requestType = IndexStatRep::RT_UPDATE_CONF;
  rep->requestFlag = 0;
  rep->indexId = req->indexId;
  rep->indexVersion = req->indexVersion;
  rep->tableId = req->tableId;
  rep->fragId = req->fragId;
  rep->loadTime = data.m_loadTime;
  sendSignal(tuxRef, GSN_INDEX_STAT_REP,
             signal, IndexStatRep::SignalLength, JBB);

  statOpSuccess(signal, stat);
}

// drop

void
Trix::statDropBegin(Signal* signal, StatOp& stat)
{
  StatOp::Data& data = stat.m_data;
  D("statDropBegin" << V(stat));

  if (data.m_head_found == true)
  {
    jam();
    statHeadDelete(signal, stat);
    return;
  }
  statDropEnd(signal, stat);
}

void
Trix::statDropEnd(Signal* signal, StatOp& stat)
{
  D("statDropEnd");
  statOpSuccess(signal, stat);
}

// send

void
Trix::statSendPrepare(Signal* signal, StatOp& stat)
{
  StatOp::Send& send = stat.m_send;
  const IndexStatImplReq* req = &stat.m_req;
  const SysTable& sysTable = *send.m_sysTable;
  D("statSendPrepare" << V(stat));

  UtilPrepareReq* utilReq =
    (UtilPrepareReq*)signal->getDataPtrSend();
  utilReq->senderData = stat.m_ownPtrI;
  utilReq->senderRef = reference();
  utilReq->schemaTransId = req->transId;

  Uint32 wbuf[256];
  LinearWriter w(&wbuf[0], sizeof(wbuf) >> 2);

  w.first();
  w.add(UtilPrepareReq::NoOfOperations, 1);
  w.add(UtilPrepareReq::OperationType, send.m_operationType);
  w.add(UtilPrepareReq::TableId, sysTable.tableId);

  Uint32 i;
  for (i = 0; i < sysTable.columnCount; i++) {
    const SysColumn& c = sysTable.columnList[i];
    switch (send.m_operationType) {
    case UtilPrepareReq::Read:
    case UtilPrepareReq::Insert:
    case UtilPrepareReq::Update:
      jam();
      w.add(UtilPrepareReq::AttributeId, i);
      break;
    case UtilPrepareReq::Delete:
      jam();
      if (c.keyFlag)
        w.add(UtilPrepareReq::AttributeId, i);
      break;
    default:
      ndbrequire(false);
      break;
    }
  }

  LinearSectionPtr ptr[3];
  ptr[0].p = &wbuf[0];
  ptr[0].sz = w.getWordsUsed();
  sendSignal(DBUTIL_REF, GSN_UTIL_PREPARE_REQ,
             signal, UtilPrepareReq::SignalLength, JBB, ptr, 1);
}

void
Trix::statSendExecute(Signal* signal, StatOp& stat)
{
  D("statSendExecute" << V(stat));
  StatOp::Send& send = stat.m_send;
  StatOp::Attr& attr = stat.m_attr;
  const SysTable& sysTable = *send.m_sysTable;

  UtilExecuteReq* utilReq =
    (UtilExecuteReq*)signal->getDataPtrSend();
  utilReq->senderData = stat.m_ownPtrI;
  utilReq->senderRef = reference();
  utilReq->prepareId = send.m_prepareId;
  utilReq->scanTakeOver = 0;

  Uint32 wattr[20];
  Uint32 wdata[2048];
  attr.m_attr = wattr;
  attr.m_attrMax = 20;
  attr.m_attrSize = 0;
  attr.m_data = wdata;
  attr.m_dataMax = 2048;
  attr.m_dataSize = 0;

  for (Uint32 i = 0; i < sysTable.columnCount; i++) {
    const SysColumn& c = sysTable.columnList[i];
    switch (send.m_operationType) {
    case UtilPrepareReq::Read:
    case UtilPrepareReq::Insert:
    case UtilPrepareReq::Update:
      jam();
      statDataOut(stat, i);
      break;
    case UtilPrepareReq::Delete:
      jam();
      if (c.keyFlag)
        statDataOut(stat, i);
      break;
    default:
      ndbrequire(false);
      break;
    }
  }

  LinearSectionPtr ptr[3];
  ptr[0].p = attr.m_attr;
  ptr[0].sz = attr.m_attrSize;
  ptr[1].p = attr.m_data;
  ptr[1].sz = attr.m_dataSize;
  sendSignal(DBUTIL_REF, GSN_UTIL_EXECUTE_REQ,
             signal, UtilExecuteReq::SignalLength, JBB, ptr, 2);
}

void
Trix::statSendRelease(Signal* signal, StatOp& stat)
{
  D("statSendRelease" << V(stat));
  StatOp::Send& send = stat.m_send;
  ndbrequire(send.m_prepareId != RNIL);

  UtilReleaseReq* utilReq =
    (UtilReleaseReq*)signal->getDataPtrSend();
  utilReq->senderData = stat.m_ownPtrI;
  utilReq->prepareId = send.m_prepareId;
  sendSignal(DBUTIL_REF, GSN_UTIL_RELEASE_REQ,
             signal, UtilReleaseReq::SignalLength, JBB);
}

// data

void
Trix::statDataPtr(StatOp& stat, Uint32 i, Uint32*& dptr, Uint32& bytes)
{
  StatOp::Data& data = stat.m_data;
  StatOp::Send& send = stat.m_send;

  const SysTable& sysTable = *send.m_sysTable;
  ndbrequire(i < sysTable.columnCount);
  //UNUSED const SysColumn& c = sysTable.columnList[i];

  if (&sysTable == &g_statMetaHead)
  {
    switch (i) {
    case 0:
      dptr = &data.m_indexId;
      bytes = 4;
      break;
    case 1:
      dptr = &data.m_indexVersion;
      bytes = 4;
      break;
    case 2:
      dptr = &data.m_tableId;
      bytes = 4;
      break;
    case 3:
      dptr = &data.m_fragCount;
      bytes = 4;
      break;
    case 4:
      dptr = &data.m_valueFormat;
      bytes = 4;
      break;
    case 5:
      dptr = &data.m_sampleVersion;
      bytes = 4;
      break;
    case 6:
      dptr = &data.m_loadTime;
      bytes = 4;
      break;
    case 7:
      dptr = &data.m_sampleCount;
      bytes = 4;
      break;
    case 8:
      dptr = &data.m_keyBytes;
      bytes = 4;
      break;
    default:
      ndbrequire(false);
      break;
    }
    return;
  }

  if (&sysTable == &g_statMetaSample)
  {
    switch (i) {
    case 0:
      dptr = &data.m_indexId;
      bytes = 4;
      break;
    case 1:
      dptr = &data.m_indexVersion;
      bytes = 4;
      break;
    case 2:
      dptr = &data.m_sampleVersion;
      bytes = 4;
      break;
    case 3:
      {
        dptr = data.m_statKey;
        const uchar* p = (uchar*)dptr;
        ndbrequire(p != 0);
        bytes = 2 + p[0] + (p[1] << 8);
      }
      break;
    case 4:
      {
        dptr = data.m_statValue;
        const uchar* p = (uchar*)dptr;
        ndbrequire(p != 0);
        bytes = 2 + p[0] + (p[1] << 8);
      }
      break;
    default:
      ndbrequire(false);
      break;
    }
    return;
  }

  ndbrequire(false);
}

void
Trix::statDataOut(StatOp& stat, Uint32 i)
{
  StatOp::Attr& attr = stat.m_attr;
  Uint32* dptr = 0;
  Uint32 bytes = 0;
  statDataPtr(stat, i, dptr, bytes);

  ndbrequire(attr.m_attrSize + 1 <= attr.m_attrMax);
  AttributeHeader::init(&attr.m_attr[attr.m_attrSize], i, bytes);
  attr.m_attrSize++;

  Uint32 words = (bytes + 3) / 4;
  ndbrequire(attr.m_dataSize + words <= attr.m_dataMax);
  Uint8* dst = (Uint8*)&attr.m_data[attr.m_dataSize];
  memcpy(dst, dptr, bytes);
  while (bytes < words * 4)
    dst[bytes++] = 0;
  attr.m_dataSize += words;
  D("statDataOut" << V(i) << V(bytes) << hex << V(dptr[0]));
}

void
Trix::statDataIn(StatOp& stat, Uint32 i)
{
  StatOp::Attr& attr = stat.m_attr;
  Uint32* dptr = 0;
  Uint32 bytes = 0;
  statDataPtr(stat, i, dptr, bytes);

  ndbrequire(attr.m_attrSize + 1 <= attr.m_attrMax);
  const AttributeHeader& ah = attr.m_attr[attr.m_attrSize];
  attr.m_attrSize++;

  ndbrequire(ah.getByteSize() == bytes);
  Uint32 words = (bytes + 3) / 4;
  ndbrequire(attr.m_dataSize + words <= attr.m_dataMax);
  const char* src = (const char*)&attr.m_data[attr.m_dataSize];
  memcpy(dptr, src, bytes);
  attr.m_dataSize += words;
  D("statDataIn" << V(i) << V(bytes) << hex << V(dptr[0]));
}

// abort ongoing

void
Trix::statAbortUtil(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  D("statAbortUtil" << V(stat));

  ndbrequire(util.m_prepareId != RNIL);
  util.m_cb.m_callbackFunction = safe_cast(&Trix::statAbortUtilCB);
  util.m_cb.m_callbackData = stat.m_ownPtrI;
  statUtilRelease(signal, stat);
}

void
Trix::statAbortUtilCB(Signal* signal, Uint32 statPtrI, Uint32 ret)
{
  StatOp& stat = statOpGetPtr(statPtrI);
  D("statAbortUtilCB" << V(stat) << V(ret));

  ndbrequire(ret == 0);
  statOpAbort(signal, stat);
}

// conf and ref

void
Trix::statOpSuccess(Signal* signal, StatOp& stat)
{
  StatOp::Data& data = stat.m_data;
  D("statOpSuccess" << V(stat));

  if (stat.m_requestType == IndexStatReq::RT_SCAN_FRAG)
    statOpEvent(stat, "I", "created %u samples", data.m_sampleCount);

  statOpConf(signal, stat);
  statOpRelease(stat);
}

void
Trix::statOpConf(Signal* signal, StatOp& stat)
{
  const IndexStatImplReq* req = &stat.m_req;
  D("statOpConf" << V(stat));

  IndexStatImplConf* conf = (IndexStatImplConf*)signal->getDataPtrSend();
  conf->senderRef = reference();
  conf->senderData = req->senderData;
  sendSignal(req->senderRef, GSN_INDEX_STAT_IMPL_CONF,
             signal, IndexStatImplConf::SignalLength, JBB);
}

void
Trix::statOpError(Signal* signal, StatOp& stat,
                  Uint32 errorCode, Uint32 errorLine,
                  const Uint32 * supress)
{
  D("statOpError" << V(stat) << V(errorCode) << V(errorLine));

  if (supress)
  {
    for (Uint32 i = 0; supress[i] != 0; i++)
    {
      if (errorCode == supress[i])
      {
        goto do_supress;
      }
    }
  }
  statOpEvent(stat, "W", "error %u line %u", errorCode, errorLine);

do_supress:
  ndbrequire(stat.m_errorCode == 0);
  stat.m_errorCode = errorCode;
  stat.m_errorLine = errorLine;
  statOpAbort(signal, stat);
}

void
Trix::statOpAbort(Signal* signal, StatOp& stat)
{
  StatOp::Util& util = stat.m_util;
  D("statOpAbort" << V(stat));
  
  if (util.m_prepareId != RNIL)
  {
    jam();
    // returns here when done
    statAbortUtil(signal, stat);
    return;
  }
  statOpRef(signal, stat);
  statOpRelease(stat);
}

void
Trix::statOpRef(Signal* signal, StatOp& stat)
{
  const IndexStatImplReq* req = &stat.m_req;
  D("statOpRef" << V(stat));

  statOpRef(signal, req, stat.m_errorCode, stat.m_errorLine);
}

void
Trix::statOpRef(Signal* signal, const IndexStatImplReq* req,
                Uint32 errorCode, Uint32 errorLine)
{
  D("statOpRef" << V(errorCode) << V(errorLine));

  IndexStatImplRef* ref = (IndexStatImplRef*)signal->getDataPtrSend();
  ref->senderRef = reference();
  ref->senderData = req->senderData;
  ref->errorCode = errorCode;
  ref->errorLine = errorLine;
  sendSignal(req->senderRef, GSN_INDEX_STAT_IMPL_REF,
             signal, IndexStatImplRef::SignalLength, JBB);
}

void
Trix::statOpEvent(StatOp& stat, const char* level, const char* msg, ...)
{
  //UNUSED const IndexStatImplReq* req = &stat.m_req;
  StatOp::Data& data = stat.m_data;

  char tmp1[100];
  va_list ap;
  va_start(ap, msg);
  BaseString::vsnprintf(tmp1, sizeof(tmp1), msg, ap);
  va_end(ap);

  char tmp2[100];
  BaseString::snprintf(tmp2, sizeof(tmp2),
                       "index %u stats version %u: %s: %s",
                       data.m_indexId, data.m_sampleVersion,
                       stat.m_requestName, tmp1);

  D("statOpEvent" << V(level) << V(tmp2));

  if (level[0] == 'I')
    infoEvent("%s", tmp2);
  if (level[0] == 'W')
    warningEvent("%s", tmp2);
}

// debug

class NdbOut&
operator<<(NdbOut& out, const Trix::StatOp& stat)
{
  out << "[";
  out << " i:" << stat.m_ownPtrI;
  out << " head_found:" << stat.m_data.m_head_found;
  out << " ]";
  return out;
}


BLOCK_FUNCTIONS(Trix)

template void append(DataBuffer<15>&,SegmentedSectionPtr,SectionSegmentPool&);