void TableFactory::configureStats(voltdb::CatalogId databaseId,
        ExecutorContext *ctx,
        std::string name,
        Table *table) {
    std::string hostname = "";
    if(ctx != NULL)
    	hostname = ctx->m_hostname;

    // initialize stats for the table
    table->getTableStats()->configure(name + " stats",
            ctx->m_hostId,
            hostname,
            ctx->m_siteId,
            ctx->m_partitionId,
            databaseId);

    // initialize stats for all the indexes for the table
    std::vector<TableIndex*> tindexes = table->allIndexes();
    for (size_t i = 0; i < tindexes.size(); i++) {
        TableIndex *index = tindexes[i];
        index->getIndexStats()->configure(index->getName() + " stats",
                table->name(),
                ctx->m_hostId,
                hostname,
                ctx->m_siteId,
                ctx->m_partitionId,
                databaseId);
    }
}

void TableFactory::initConstraints(PersistentTable* table) {

    // count the unique indexes
    table->m_uniqueIndexCount = 0;
    for (int i = 0; i < table->m_indexCount; ++i) {
        TableIndex *index = table->m_indexes[i];
        if (index->isUniqueIndex()) {
            table->m_uniqueIndexCount++;
        }
    }

    if (table->m_uniqueIndexes)
        delete[] table->m_uniqueIndexes;
    table->m_uniqueIndexes = new TableIndex*[table->m_uniqueIndexCount];
    int curIndex = 0;
    if (table->m_pkeyIndex != NULL) {
        table->m_uniqueIndexes[curIndex++] = table->m_pkeyIndex;
    }

    for (int i = 0; i < table->m_indexCount; ++i) {
        TableIndex *index = table->m_indexes[i];
        if ((index->isUniqueIndex()) && (index != table->m_pkeyIndex)) {
            table->m_uniqueIndexes[curIndex++] = index;
        }
    }
}

 double TableColumn::eval( const TableIndex& index) const {
     size_t index1 = index.getIndex1();
     double weight1 = index.getWeight1( );
     double value = m_values[index1] * weight1;
     if (weight1 < 1.0) {
         double weight2 = index.getWeight2( );
         value += weight2 * m_values[index1 + 1];
     }
     return value;
 }

/*
 * Delete and rebuild id based table collections. Does not affect
 * any currently stored tuples.
 */
bool VoltDBEngine::rebuildTableCollections() {
    // 1. See header comments explaining m_snapshottingTables.
    // 2. Don't clear m_exportTables. They are still exporting, even if deleted.
    // 3. Clear everything else.
    m_tables.clear();
    m_tablesByName.clear();

    // need to re-map all the table ids / indexes
    getStatsManager().unregisterStatsSource(STATISTICS_SELECTOR_TYPE_TABLE);
    getStatsManager().unregisterStatsSource(STATISTICS_SELECTOR_TYPE_INDEX);

    map<string, CatalogDelegate*>::iterator cdIt = m_catalogDelegates.begin();

    // walk the table delegates and update local table collections
    while (cdIt != m_catalogDelegates.end()) {
        TableCatalogDelegate *tcd = dynamic_cast<TableCatalogDelegate*>(cdIt->second);
        if (tcd) {
            catalog::Table *catTable = m_database->tables().get(tcd->getTable()->name());
            m_tables[catTable->relativeIndex()] = tcd->getTable();
            m_tablesByName[tcd->getTable()->name()] = tcd->getTable();

            getStatsManager().registerStatsSource(STATISTICS_SELECTOR_TYPE_TABLE,
                                                  catTable->relativeIndex(),
                                                  tcd->getTable()->getTableStats());

            // add all of the indexes to the stats source
            std::vector<TableIndex*> tindexes = tcd->getTable()->allIndexes();
            for (int i = 0; i < tindexes.size(); i++) {
                TableIndex *index = tindexes[i];
                getStatsManager().registerStatsSource(STATISTICS_SELECTOR_TYPE_INDEX,
                                                      catTable->relativeIndex(),
                                                      index->getIndexStats());
            }


            /*map<string, catalog::Index*>::const_iterator index_iterator;
            for (index_iterator = catTable->indexes().begin();
                 index_iterator != catTable->indexes().end(); index_iterator++) {

                const catalog::Index *catalogIndex = index_iterator->second;
                TableIndex *index = tcd->getTable()->index(catalogIndex->name());
                printf("Looking for index named: %s\n", catalogIndex->name().c_str());
                assert(index);

                getStatsManager().registerStatsSource(STATISTICS_SELECTOR_TYPE_INDEX,
                                                      catTable->relativeIndex(),
                                                      index->getIndexStats());
            }*/
        }
        cdIt++;
    }

    return true;
}

    double PvtxTable::evaluate(const std::string& column, double outerArg, double innerArg) const
    {
        TableIndex outerIndex = m_outerColumn.lookup( outerArg );
        const auto& underSaturatedTable1 = getUnderSaturatedTable( outerIndex.getIndex1( ) );
        double weight1 = outerIndex.getWeight1( );
        double value = weight1 * underSaturatedTable1.evaluate( column , innerArg );

        if (weight1 < 1) {
            const auto& underSaturatedTable2 = getUnderSaturatedTable( outerIndex.getIndex2( ) );
            double weight2 = outerIndex.getWeight2( );

            value += weight2 * underSaturatedTable2.evaluate( column , innerArg );
        }

        return value;
    }

TEST_F(CompactingHashIndexTest, ENG1193) {
    TableIndex *index = NULL;
    vector<int> columnIndices;
    vector<ValueType> columnTypes;
    vector<int32_t> columnLengths;
    vector<bool> columnAllowNull;

    columnIndices.push_back(0);
    columnTypes.push_back(VALUE_TYPE_BIGINT);
    columnLengths.push_back(NValue::getTupleStorageSize(VALUE_TYPE_BIGINT));
    columnAllowNull.push_back(false);

    TupleSchema *schema = TupleSchema::createTupleSchemaForTest(columnTypes,
                                                         columnLengths,
                                                         columnAllowNull);

    TableIndexScheme scheme("test_index", HASH_TABLE_INDEX,
                            columnIndices, TableIndex::simplyIndexColumns(),
                            false, false, schema);
    index = TableIndexFactory::getInstance(scheme);

    TableTuple *tuple1 = newTuple(schema, 0, 10);
    index->addEntry(tuple1);
    TableTuple *tuple2 = newTuple(schema, 0, 11);
    index->addEntry(tuple2);
    TableTuple *tuple3 = newTuple(schema, 0, 12);
    index->addEntry(tuple3);

    TableTuple *tuple4 = newTuple(schema, 0, 10);
    EXPECT_TRUE(index->replaceEntryNoKeyChange(*tuple4, *tuple1));

    EXPECT_FALSE(index->exists(tuple1));
    EXPECT_TRUE(index->exists(tuple2));
    EXPECT_TRUE(index->exists(tuple3));
    EXPECT_TRUE(index->exists(tuple4));

    delete index;
    TupleSchema::freeTupleSchema(schema);
    delete[] tuple1->address();
    delete tuple1;
    delete[] tuple2->address();
    delete tuple2;
    delete[] tuple3->address();
    delete tuple3;
    delete[] tuple4->address();
    delete tuple4;
}

bool IndexScanExecutor::p_init(AbstractPlanNode *abstractNode,
        TempTableLimits* limits)
{
    VOLT_TRACE("init IndexScan Executor");

    m_projectionNode = NULL;

    m_node = dynamic_cast<IndexScanPlanNode*>(abstractNode);
    assert(m_node);
    assert(m_node->getTargetTable());

    // Create output table based on output schema from the plan
    setTempOutputTable(limits, m_node->getTargetTable()->name());

    //
    // INLINE PROJECTION
    //
    if (m_node->getInlinePlanNode(PLAN_NODE_TYPE_PROJECTION) != NULL) {
        m_projectionNode = static_cast<ProjectionPlanNode*>
            (m_node->getInlinePlanNode(PLAN_NODE_TYPE_PROJECTION));

        m_projector = OptimizedProjector(m_projectionNode->getOutputColumnExpressions());
        m_projector.optimize(m_projectionNode->getOutputTable()->schema(),
                             m_node->getTargetTable()->schema());
    }

    // Inline aggregation can be serial, partial or hash
    m_aggExec = voltdb::getInlineAggregateExecutor(m_abstractNode);

    //
    // Make sure that we have search keys and that they're not null
    //
    m_numOfSearchkeys = (int)m_node->getSearchKeyExpressions().size();
    m_searchKeyArrayPtr =
            boost::shared_array<AbstractExpression*>
    (new AbstractExpression*[m_numOfSearchkeys]);
    m_searchKeyArray = m_searchKeyArrayPtr.get();

    for (int ctr = 0; ctr < m_numOfSearchkeys; ctr++)
    {
        if (m_node->getSearchKeyExpressions()[ctr] == NULL)
        {
            VOLT_ERROR("The search key expression at position '%d' is NULL for"
                    " PlanNode '%s'", ctr, m_node->debug().c_str());
            return false;
        }
        m_searchKeyArrayPtr[ctr] =
                m_node->getSearchKeyExpressions()[ctr];
    }

    //output table should be temptable
    m_outputTable = static_cast<TempTable*>(m_node->getOutputTable());

    Table* targetTable = m_node->getTargetTable();
    //target table should be persistent table
    assert(dynamic_cast<PersistentTable*>(targetTable));

    TableIndex *tableIndex = targetTable->index(m_node->getTargetIndexName());
    m_searchKeyBackingStore = new char[tableIndex->getKeySchema()->tupleLength()];

    // Grab the Index from our inner table
    // We'll throw an error if the index is missing
    VOLT_TRACE("Index key schema: '%s'", tableIndex->getKeySchema()->debug().c_str());
    //
    // Miscellanous Information
    //
    m_lookupType = m_node->getLookupType();
    m_sortDirection = m_node->getSortDirection();

    VOLT_DEBUG("IndexScan: %s.%s\n", targetTable->name().c_str(), tableIndex->getName().c_str());

    return true;
}

bool IndexScanExecutor::p_execute(const NValueArray &params)
{
    assert(m_node);
    assert(m_node == dynamic_cast<IndexScanPlanNode*>(m_abstractNode));

    // update local target table with its most recent reference
    Table* targetTable = m_node->getTargetTable();
    TableIndex *tableIndex = targetTable->index(m_node->getTargetIndexName());
    IndexCursor indexCursor(tableIndex->getTupleSchema());

    TableTuple searchKey(tableIndex->getKeySchema());
    searchKey.moveNoHeader(m_searchKeyBackingStore);

    assert(m_lookupType != INDEX_LOOKUP_TYPE_EQ ||
            searchKey.getSchema()->columnCount() == m_numOfSearchkeys);

    int activeNumOfSearchKeys = m_numOfSearchkeys;
    IndexLookupType localLookupType = m_lookupType;
    SortDirectionType localSortDirection = m_sortDirection;

    //
    // INLINE LIMIT
    //
    LimitPlanNode* limit_node = dynamic_cast<LimitPlanNode*>(m_abstractNode->getInlinePlanNode(PLAN_NODE_TYPE_LIMIT));

    TableTuple temp_tuple;
    ProgressMonitorProxy pmp(m_engine, this);
    if (m_aggExec != NULL) {
        const TupleSchema * inputSchema = tableIndex->getTupleSchema();
        if (m_projectionNode != NULL) {
            inputSchema = m_projectionNode->getOutputTable()->schema();
        }
        temp_tuple = m_aggExec->p_execute_init(params, &pmp, inputSchema, m_outputTable);
    } else {
        temp_tuple = m_outputTable->tempTuple();
    }

    // Short-circuit an empty scan
    if (m_node->isEmptyScan()) {
        VOLT_DEBUG ("Empty Index Scan :\n %s", m_outputTable->debug().c_str());
        if (m_aggExec != NULL) {
            m_aggExec->p_execute_finish();
        }
        return true;
    }

    //
    // SEARCH KEY
    //
    bool earlyReturnForSearchKeyOutOfRange = false;

    searchKey.setAllNulls();
    VOLT_TRACE("Initial (all null) search key: '%s'", searchKey.debugNoHeader().c_str());

    for (int ctr = 0; ctr < activeNumOfSearchKeys; ctr++) {
        NValue candidateValue = m_searchKeyArray[ctr]->eval(NULL, NULL);
        if (candidateValue.isNull()) {
            // when any part of the search key is NULL, the result is false when it compares to anything.
            // do early return optimization, our index comparator may not handle null comparison correctly.
            earlyReturnForSearchKeyOutOfRange = true;
            break;
        }

        try {
            searchKey.setNValue(ctr, candidateValue);
        }
        catch (const SQLException &e) {
            // This next bit of logic handles underflow, overflow and search key length
            // exceeding variable length column size (variable lenght mismatch) when
            // setting up the search keys.
            // e.g. TINYINT > 200 or INT <= 6000000000
            // VarChar(3 bytes) < "abcd" or VarChar(3) > "abbd"

            // re-throw if not an overflow, underflow or variable length mismatch
            // currently, it's expected to always be an overflow or underflow
            if ((e.getInternalFlags() & (SQLException::TYPE_OVERFLOW | SQLException::TYPE_UNDERFLOW | SQLException::TYPE_VAR_LENGTH_MISMATCH)) == 0) {
                throw e;
            }

            // handle the case where this is a comparison, rather than equality match
            // comparison is the only place where the executor might return matching tuples
            // e.g. TINYINT < 1000 should return all values
            if ((localLookupType != INDEX_LOOKUP_TYPE_EQ) &&
                    (ctr == (activeNumOfSearchKeys - 1))) {

                if (e.getInternalFlags() & SQLException::TYPE_OVERFLOW) {
                    if ((localLookupType == INDEX_LOOKUP_TYPE_GT) ||
                            (localLookupType == INDEX_LOOKUP_TYPE_GTE)) {

                        // gt or gte when key overflows returns nothing except inline agg
                        earlyReturnForSearchKeyOutOfRange = true;
                        break;
                    }
                    else {
                        // for overflow on reverse scan, we need to
                        // do a forward scan to find the correct start
                        // point, which is exactly what LTE would do.
                        // so, set the lookupType to LTE and the missing
                        // searchkey will be handled by extra post filters
                        localLookupType = INDEX_LOOKUP_TYPE_LTE;
//.........这里部分代码省略.........

bool UpdateExecutor::p_execute(const NValueArray &params, ReadWriteTracker *tracker) {
    assert(m_inputTable);
    assert(m_targetTable);

    VOLT_TRACE("INPUT TABLE: %s\n", m_inputTable->debug().c_str());
    VOLT_TRACE("TARGET TABLE - BEFORE: %s\n", m_targetTable->debug().c_str());

    assert(m_inputTuple.sizeInValues() == m_inputTable->columnCount());
    assert(m_targetTuple.sizeInValues() == m_targetTable->columnCount());
    TableIterator input_iterator(m_inputTable);
    while (input_iterator.next(m_inputTuple)) {
        //
        // OPTIMIZATION: Single-Sited Query Plans
        // If our beloved UpdatePlanNode is apart of a single-site query plan,
        // then the first column in the input table will be the address of a
        // tuple on the target table that we will want to update. This saves us
        // the trouble of having to do an index lookup
        //
        void *target_address = m_inputTuple.getNValue(0).castAsAddress();
        m_targetTuple.move(target_address);
        
        // Read/Write Set Tracking
        if (tracker != NULL) {
            tracker->markTupleWritten(m_targetTable, &m_targetTuple);
        }

        // Loop through INPUT_COL_IDX->TARGET_COL_IDX mapping and only update
        // the values that we need to. The key thing to note here is that we
        // grab a temp tuple that is a copy of the target tuple (i.e., the tuple
        // we want to update). This insures that if the input tuple is somehow
        // bringing garbage with it, we're only going to copy what we really
        // need to into the target tuple.
        //
        TableTuple &tempTuple = m_targetTable->getTempTupleInlined(m_targetTuple);
        for (int map_ctr = 0; map_ctr < m_inputTargetMapSize; map_ctr++) {
            tempTuple.setNValue(m_inputTargetMap[map_ctr].second,
                                m_inputTuple.getNValue(m_inputTargetMap[map_ctr].first));
        }

        // if there is a partition column for the target table
        if (m_partitionColumn != -1) {
            // check for partition problems
            // get the value for the partition column
            NValue value = tempTuple.getNValue(m_partitionColumn);
            bool isLocal = m_engine->isLocalSite(value);

            // if it doesn't map to this site
            if (!isLocal) {
                VOLT_ERROR("Mispartitioned tuple in single-partition plan for"
                           " table '%s'", m_targetTable->name().c_str());
                return false;
            }
        }

        #ifdef ARIES
        if(m_engine->isARIESEnabled()){

            // add persistency check:
            PersistentTable* table = dynamic_cast<PersistentTable*>(m_targetTable);

            // only log if we are writing to a persistent table.
            if (table != NULL) {
                // before image -- target is old val with no updates
                // XXX: what about uninlined fields?
                // should we not be doing
                // m_targetTable->getTempTupleInlined(m_targetTuple); instead?
                TableTuple *beforeImage = &m_targetTuple;

                // after image -- temp is NEW, created using target and input
                TableTuple *afterImage = &tempTuple;

                TableTuple *keyTuple = NULL;
                char *keydata = NULL;
                std::vector<int32_t> modifiedCols;

                int32_t numCols = -1;

                // See if we can do better by using an index instead
                TableIndex *index = table->primaryKeyIndex();

                if (index != NULL) {
                    // First construct tuple for primary key
                    keydata = new char[index->getKeySchema()->tupleLength()];
                    keyTuple = new TableTuple(keydata, index->getKeySchema());

                    for (int i = 0; i < index->getKeySchema()->columnCount(); i++) {
                        keyTuple->setNValue(i, beforeImage->getNValue(index->getColumnIndices()[i]));
                    }

                    // no before image need be recorded, just the primary key
                    beforeImage = NULL;
                }

                // Set the modified column list
                numCols = m_inputTargetMapSize;

                modifiedCols.resize(m_inputTargetMapSize, -1);

                for (int map_ctr = 0; map_ctr < m_inputTargetMapSize; map_ctr++) {
                    // can't use column-id directly, otherwise we would go over vector bounds
//.........这里部分代码省略.........

bool IndexScanExecutor::p_execute(const NValueArray &params)
{
    assert(m_node);
    assert(m_node == dynamic_cast<IndexScanPlanNode*>(m_abstractNode));
    assert(m_outputTable);
    assert(m_outputTable == static_cast<TempTable*>(m_node->getOutputTable()));

    // update local target table with its most recent reference
    Table* targetTable = m_node->getTargetTable();
    TableIndex *tableIndex = targetTable->index(m_node->getTargetIndexName());
    TableTuple searchKey(tableIndex->getKeySchema());
    searchKey.moveNoHeader(m_searchKeyBackingStore);

    assert(m_lookupType != INDEX_LOOKUP_TYPE_EQ ||
            searchKey.getSchema()->columnCount() == m_numOfSearchkeys);

    int activeNumOfSearchKeys = m_numOfSearchkeys;
    IndexLookupType localLookupType = m_lookupType;
    SortDirectionType localSortDirection = m_sortDirection;

    // INLINE PROJECTION
    // Set params to expression tree via substitute()
    assert(m_numOfColumns == m_outputTable->columnCount());
    if (m_projectionNode != NULL && m_projectionAllTupleArray == NULL)
    {
        for (int ctr = 0; ctr < m_numOfColumns; ctr++)
        {
            assert(m_projectionNode->getOutputColumnExpressions()[ctr]);
            m_projectionExpressions[ctr]->substitute(params);
            assert(m_projectionExpressions[ctr]);
        }
    }

    //
    // INLINE LIMIT
    //
    LimitPlanNode* limit_node = dynamic_cast<LimitPlanNode*>(m_abstractNode->getInlinePlanNode(PLAN_NODE_TYPE_LIMIT));

    //
    // SEARCH KEY
    //
    searchKey.setAllNulls();
    VOLT_TRACE("Initial (all null) search key: '%s'", searchKey.debugNoHeader().c_str());
    for (int ctr = 0; ctr < activeNumOfSearchKeys; ctr++) {
        m_searchKeyArray[ctr]->substitute(params);
        NValue candidateValue = m_searchKeyArray[ctr]->eval(NULL, NULL);
        try {
            searchKey.setNValue(ctr, candidateValue);
        }
        catch (const SQLException &e) {
            // This next bit of logic handles underflow and overflow while
            // setting up the search keys.
            // e.g. TINYINT > 200 or INT <= 6000000000

            // re-throw if not an overflow or underflow
            // currently, it's expected to always be an overflow or underflow
            if ((e.getInternalFlags() & (SQLException::TYPE_OVERFLOW | SQLException::TYPE_UNDERFLOW)) == 0) {
                throw e;
            }

            // handle the case where this is a comparison, rather than equality match
            // comparison is the only place where the executor might return matching tuples
            // e.g. TINYINT < 1000 should return all values
            if ((localLookupType != INDEX_LOOKUP_TYPE_EQ) &&
                (ctr == (activeNumOfSearchKeys - 1))) {

                if (e.getInternalFlags() & SQLException::TYPE_OVERFLOW) {
                    if ((localLookupType == INDEX_LOOKUP_TYPE_GT) ||
                        (localLookupType == INDEX_LOOKUP_TYPE_GTE)) {

                        // gt or gte when key overflows returns nothing
                        return true;
                    }
                    else {
                        // for overflow on reverse scan, we need to
                        // do a forward scan to find the correct start
                        // point, which is exactly what LTE would do.
                        // so, set the lookupType to LTE and the missing
                        // searchkey will be handled by extra post filters
                        localLookupType = INDEX_LOOKUP_TYPE_LTE;
                    }
                }
                if (e.getInternalFlags() & SQLException::TYPE_UNDERFLOW) {
                    if ((localLookupType == INDEX_LOOKUP_TYPE_LT) ||
                        (localLookupType == INDEX_LOOKUP_TYPE_LTE)) {

                        // lt or lte when key underflows returns nothing
                        return true;
                    }
                    else {
                        // don't allow GTE because it breaks null handling
                        localLookupType = INDEX_LOOKUP_TYPE_GT;
                    }
                }

                // if here, means all tuples with the previous searchkey
                // columns need to be scaned. Note, if only one column,
                // then all tuples will be scanned
                activeNumOfSearchKeys--;
                if (localSortDirection == SORT_DIRECTION_TYPE_INVALID) {
//.........这里部分代码省略.........

bool DeleteExecutor::p_execute(const NValueArray &params, ReadWriteTracker *tracker) {
    assert(m_targetTable);

    if (m_truncate) {
        VOLT_TRACE("truncating table %s...", m_targetTable->name().c_str());
        // count the truncated tuples as deleted
        m_engine->m_tuplesModified += m_inputTable->activeTupleCount();

        #ifdef ARIES
        if(m_engine->isARIESEnabled()){
            // no need of persistency check, m_targetTable is
            // always persistent for deletes

            LogRecord *logrecord = new LogRecord(computeTimeStamp(),
                    LogRecord::T_TRUNCATE,// this is a truncate record
                    LogRecord::T_FORWARD,// the system is running normally
                    -1,// XXX: prevLSN must be fetched from table!
                    m_engine->getExecutorContext()->currentTxnId() ,// txn id
                    m_engine->getSiteId(),// which execution site
                    m_targetTable->name(),// the table affected
                    NULL,// primary key irrelevant
                    -1,// irrelevant numCols
                    NULL,// list of modified cols irrelevant
                    NULL,// before image irrelevant
                    NULL// after image irrelevant
            );

            size_t logrecordLength = logrecord->getEstimatedLength();
            char *logrecordBuffer = new char[logrecordLength];

            FallbackSerializeOutput output;
            output.initializeWithPosition(logrecordBuffer, logrecordLength, 0);

            logrecord->serializeTo(output);

            LogManager* m_logManager = this->m_engine->getLogManager();
            Logger m_ariesLogger = m_logManager->getAriesLogger();
            //VOLT_WARN("m_logManager : %p AriesLogger : %p",&m_logManager, &m_ariesLogger);
            const Logger *logger = m_logManager->getThreadLogger(LOGGERID_MM_ARIES);

            logger->log(LOGLEVEL_INFO, output.data(), output.position());

            delete[] logrecordBuffer;
            logrecordBuffer = NULL;

            delete logrecord;
            logrecord = NULL;

        }
        #endif

        //m_engine->context().incrementTuples(m_targetTable->activeTupleCount());
        // actually delete all the tuples
        m_targetTable->deleteAllTuples(true);
        return true;
    }
    // XXX : ARIES : Not sure if else is needed ?
    assert(m_inputTable);

    assert(m_inputTuple.sizeInValues() == m_inputTable->columnCount());
    assert(m_targetTuple.sizeInValues() == m_targetTable->columnCount());
    TableIterator inputIterator(m_inputTable);
    while (inputIterator.next(m_inputTuple)) {
        //
        // OPTIMIZATION: Single-Sited Query Plans
        // If our beloved DeletePlanNode is apart of a single-site query plan,
        // then the first column in the input table will be the address of a
        // tuple on the target table that we will want to blow away. This saves
        // us the trouble of having to do an index lookup
        //
        void *targetAddress = m_inputTuple.getNValue(0).castAsAddress();
        m_targetTuple.move(targetAddress);
        
        // Read/Write Set Tracking
        if (tracker != NULL) {
            tracker->markTupleWritten(m_targetTable, &m_targetTuple);
        }

        #ifdef ARIES
        if(m_engine->isARIESEnabled()){
            // no need of persistency check, m_targetTable is
            // always persistent for deletes

            // before image -- target is tuple to be deleted.
            TableTuple *beforeImage = &m_targetTuple;

            TableTuple *keyTuple = NULL;
            char *keydata = NULL;


            // See if we use an index instead
            TableIndex *index = m_targetTable->primaryKeyIndex();

            if (index != NULL) {
                // First construct tuple for primary key
                keydata = new char[index->getKeySchema()->tupleLength()];
                keyTuple = new TableTuple(keydata, index->getKeySchema());

                for (int i = 0; i < index->getKeySchema()->columnCount(); i++) {
                    keyTuple->setNValue(i, beforeImage->getNValue(index->getColumnIndices()[i]));
//.........这里部分代码省略.........

bool NestLoopIndexExecutor::p_init(AbstractPlanNode* abstractNode,
                                   TempTableLimits* limits)
{
    VOLT_TRACE("init NLIJ Executor");
    assert(limits);

    // Init parent first
    if (!AbstractJoinExecutor::p_init(abstractNode, limits)) {
        return false;
    }

    NestLoopIndexPlanNode* node = dynamic_cast<NestLoopIndexPlanNode*>(m_abstractNode);
    assert(node);
    m_indexNode =
        dynamic_cast<IndexScanPlanNode*>(m_abstractNode->getInlinePlanNode(PLAN_NODE_TYPE_INDEXSCAN));
    assert(m_indexNode);
    VOLT_TRACE("<NestLoopIndexPlanNode> %s, <IndexScanPlanNode> %s",
               m_abstractNode->debug().c_str(), m_indexNode->debug().c_str());

    m_lookupType = m_indexNode->getLookupType();
    m_sortDirection = m_indexNode->getSortDirection();

    //
    // We need exactly one input table and a target table
    //
    assert(node->getInputTableCount() == 1);

    node->getOutputColumnExpressions(m_outputExpressions);

    //
    // Make sure that we actually have search keys
    //
    int num_of_searchkeys = static_cast <int> (m_indexNode->getSearchKeyExpressions().size());
    //nshi commented this out in revision 4495 of the old repo in index scan executor
    VOLT_TRACE ("<Nested Loop Index exec, INIT...> Number of searchKeys: %d \n", num_of_searchkeys);

    for (int ctr = 0; ctr < num_of_searchkeys; ctr++) {
        if (m_indexNode->getSearchKeyExpressions()[ctr] == NULL) {
            VOLT_ERROR("The search key expression at position '%d' is NULL for"
                       " internal PlanNode '%s' of PlanNode '%s'",
                       ctr, m_indexNode->debug().c_str(), node->debug().c_str());
            return false;
        }
    }

    assert(node->getInputTable());

    PersistentTable* inner_table = dynamic_cast<PersistentTable*>(m_indexNode->getTargetTable());
    assert(inner_table);

    // Grab the Index from our inner table
    // We'll throw an error if the index is missing
    TableIndex* index = inner_table->index(m_indexNode->getTargetIndexName());
    if (index == NULL) {
        VOLT_ERROR("Failed to retreive index '%s' from inner table '%s' for"
                   " internal PlanNode '%s'",
                   m_indexNode->getTargetIndexName().c_str(),
                   inner_table->name().c_str(), m_indexNode->debug().c_str());
        return false;
    }

    // NULL tuples for left and full joins
    p_init_null_tuples(node->getInputTable(), m_indexNode->getTargetTable());

    m_indexValues.init(index->getKeySchema());
    return true;
}

bool NestLoopIndexExecutor::p_execute(const NValueArray &params)
{
    assert(dynamic_cast<NestLoopIndexPlanNode*>(m_abstractNode));
    NestLoopIndexPlanNode* node = static_cast<NestLoopIndexPlanNode*>(m_abstractNode);

    // output table must be a temp table
    assert(m_tmpOutputTable);
    // target table is a persistent table
    assert(dynamic_cast<PersistentTable*>(m_indexNode->getTargetTable()));
    PersistentTable* inner_table = static_cast<PersistentTable*>(m_indexNode->getTargetTable());

    TableIndex* index = inner_table->index(m_indexNode->getTargetIndexName());
    assert(index);
    IndexCursor indexCursor(index->getTupleSchema());

    //outer_table is the input table that have tuples to be iterated
    assert(node->getInputTableCount() == 1);
    Table* outer_table = node->getInputTable();
    assert(outer_table);
    VOLT_TRACE("executing NestLoopIndex with outer table: %s, inner table: %s",
               outer_table->debug().c_str(), inner_table->debug().c_str());

    //
    // Substitute parameter to SEARCH KEY Note that the expressions
    // will include TupleValueExpression even after this substitution
    //
    int num_of_searchkeys = static_cast <int> (m_indexNode->getSearchKeyExpressions().size());
    for (int ctr = 0; ctr < num_of_searchkeys; ctr++) {
        VOLT_TRACE("Search Key[%d]:\n%s",
                   ctr, m_indexNode->getSearchKeyExpressions()[ctr]->debug(true).c_str());
    }

    // end expression
    AbstractExpression* end_expression = m_indexNode->getEndExpression();
    if (end_expression) {
        VOLT_TRACE("End Expression:\n%s", end_expression->debug(true).c_str());
    }

    // post expression
    AbstractExpression* post_expression = m_indexNode->getPredicate();
    if (post_expression != NULL) {
        VOLT_TRACE("Post Expression:\n%s", post_expression->debug(true).c_str());
    }

    // initial expression
    AbstractExpression* initial_expression = m_indexNode->getInitialExpression();
    if (initial_expression != NULL) {
        VOLT_TRACE("Initial Expression:\n%s", initial_expression->debug(true).c_str());
    }

    // SKIP NULL EXPRESSION
    AbstractExpression* skipNullExpr = m_indexNode->getSkipNullPredicate();
    // For reverse scan edge case NULL values and forward scan underflow case.
    if (skipNullExpr != NULL) {
        VOLT_DEBUG("Skip NULL Expression:\n%s", skipNullExpr->debug(true).c_str());
    }

    // pre join expression
    AbstractExpression* prejoin_expression = node->getPreJoinPredicate();
    if (prejoin_expression != NULL) {
        VOLT_TRACE("Prejoin Expression:\n%s", prejoin_expression->debug(true).c_str());
    }

    // where expression
    AbstractExpression* where_expression = node->getWherePredicate();
    if (where_expression != NULL) {
        VOLT_TRACE("Where Expression:\n%s", where_expression->debug(true).c_str());
    }

    LimitPlanNode* limit_node = dynamic_cast<LimitPlanNode*>(node->getInlinePlanNode(PLAN_NODE_TYPE_LIMIT));
    int limit = CountingPostfilter::NO_LIMIT;
    int offset = CountingPostfilter::NO_OFFSET;
    if (limit_node) {
        limit_node->getLimitAndOffsetByReference(params, limit, offset);
    }
    // Init the postfilter
    CountingPostfilter postfilter(m_tmpOutputTable, where_expression, limit, offset);

    //
    // OUTER TABLE ITERATION
    //
    TableTuple outer_tuple(outer_table->schema());
    TableTuple inner_tuple(inner_table->schema());
    TableIterator outer_iterator = outer_table->iteratorDeletingAsWeGo();
    int num_of_outer_cols = outer_table->columnCount();
    assert (outer_tuple.sizeInValues() == outer_table->columnCount());
    assert (inner_tuple.sizeInValues() == inner_table->columnCount());
    const TableTuple &null_inner_tuple = m_null_inner_tuple.tuple();
    ProgressMonitorProxy pmp(m_engine->getExecutorContext(), this);

    // The table filter to keep track of inner tuples that don't match any of outer tuples for FULL joins
    TableTupleFilter innerTableFilter;
    if (m_joinType == JOIN_TYPE_FULL) {
        // Prepopulate the set with all inner tuples
        innerTableFilter.init(inner_table);
    }

    TableTuple join_tuple;
    // It's not immediately obvious here, so there's some subtlety to
    // note with respect to the schema of the join_tuple.
//.........这里部分代码省略.........

bool IndexCountExecutor::p_init(AbstractPlanNode *abstractNode,
                                TempTableLimits* limits)
{
    VOLT_DEBUG("init IndexCount Executor");

    m_node = dynamic_cast<IndexCountPlanNode*>(abstractNode);
    assert(m_node);
    assert(m_node->getTargetTable());
    assert(m_node->getPredicate() == NULL);

    // Create output table based on output schema from the plan
    setTempOutputTable(limits);

    //
    // Make sure that we have search keys and that they're not null
    //
    m_numOfSearchkeys = (int)m_node->getSearchKeyExpressions().size();
    if (m_numOfSearchkeys != 0) {
        m_searchKeyArrayPtr =
            boost::shared_array<AbstractExpression*>
            (new AbstractExpression*[m_numOfSearchkeys]);
        m_searchKeyArray = m_searchKeyArrayPtr.get();

        for (int ctr = 0; ctr < m_numOfSearchkeys; ctr++) {
            if (m_node->getSearchKeyExpressions()[ctr] == NULL) {
                VOLT_ERROR("The search key expression at position '%d' is NULL for"
                    " PlanNode '%s'", ctr, m_node->debug().c_str());
                return false;
            }
            m_searchKeyArrayPtr[ctr] = m_node->getSearchKeyExpressions()[ctr];
        }
    }

    m_numOfEndkeys = (int)m_node->getEndKeyExpressions().size();
    if (m_numOfEndkeys != 0) {
        m_endKeyArrayPtr =
            boost::shared_array<AbstractExpression*> (new AbstractExpression*[m_numOfEndkeys]);
        m_endKeyArray = m_endKeyArrayPtr.get();
        for (int ctr = 0; ctr < m_numOfEndkeys; ctr++)
        {
            if (m_node->getEndKeyExpressions()[ctr] == NULL) {
                VOLT_ERROR("The end key expression at position '%d' is NULL for"
                    " PlanNode '%s'", ctr, m_node->debug().c_str());
                return false;
            }
            m_endKeyArrayPtr[ctr] = m_node->getEndKeyExpressions()[ctr];
        }
    }
    //output table should be temptable
    m_outputTable = static_cast<TempTable*>(m_node->getOutputTable());
    m_numOfColumns = static_cast<int>(m_outputTable->columnCount());
    assert(m_numOfColumns == 1);

    // Miscellanous Information
    m_lookupType = INDEX_LOOKUP_TYPE_INVALID;
    if (m_numOfSearchkeys != 0) {
        m_lookupType = m_node->getLookupType();
    }

    if (m_numOfEndkeys != 0) {
        m_endType = m_node->getEndType();
    }

    //
    // Grab the Index from our inner table
    // We'll throw an error if the index is missing
    //
    Table* targetTable = m_node->getTargetTable();
    //target table should be persistenttable
    assert(dynamic_cast<PersistentTable*>(targetTable));

    TableIndex *tableIndex = targetTable->index(m_node->getTargetIndexName());
    assert (tableIndex != NULL);

    // This index should have a true countable flag
    assert(tableIndex->isCountableIndex());

    if (m_numOfSearchkeys != 0) {
        m_searchKeyBackingStore = new char[tableIndex->getKeySchema()->tupleLength()];
    }

    if (m_numOfEndkeys != 0) {
        m_endKeyBackingStore = new char[tableIndex->getKeySchema()->tupleLength()];
    }

    VOLT_DEBUG("IndexCount: %s.%s\n", targetTable->name().c_str(),
            tableIndex->getName().c_str());

    return true;
}

bool IndexCountExecutor::p_execute(const NValueArray &params)
{
    // update local target table with its most recent reference
    Table* targetTable = m_node->getTargetTable();
    TableIndex * tableIndex = targetTable->index(m_node->getTargetIndexName());

    TableTuple searchKey, endKey;
    if (m_numOfSearchkeys != 0) {
        searchKey = TableTuple(tableIndex->getKeySchema());
        searchKey.moveNoHeader(m_searchKeyBackingStore);
    }
    if (m_numOfEndkeys != 0) {
        endKey = TableTuple(tableIndex->getKeySchema());
        endKey.moveNoHeader(m_endKeyBackingStore);
    }

    // Need to move GTE to find (x,_) when doing a partial covering search.
    // The planner sometimes used to lie in this case: index_lookup_type_eq is incorrect.
    // Index_lookup_type_gte is necessary.
    assert(m_lookupType != INDEX_LOOKUP_TYPE_EQ ||
            searchKey.getSchema()->columnCount() == m_numOfSearchkeys ||
            searchKey.getSchema()->columnCount() == m_numOfEndkeys);

    int activeNumOfSearchKeys = m_numOfSearchkeys;
    IndexLookupType localLookupType = m_lookupType;
    bool searchKeyUnderflow = false, endKeyOverflow = false;
    // Overflow cases that can return early without accessing the index need this
    // default 0 count as their result.
    TableTuple& tmptup = m_outputTable->tempTuple();
    tmptup.setNValue(0, ValueFactory::getBigIntValue( 0 ));

    //
    // SEARCH KEY
    //
    if (m_numOfSearchkeys != 0) {
        searchKey.setAllNulls();
        VOLT_DEBUG("<Index Count>Initial (all null) search key: '%s'", searchKey.debugNoHeader().c_str());
        for (int ctr = 0; ctr < activeNumOfSearchKeys; ctr++) {
            NValue candidateValue = m_searchKeyArray[ctr]->eval(NULL, NULL);
            try {
                searchKey.setNValue(ctr, candidateValue);
            }
            catch (const SQLException &e) {
                // This next bit of logic handles underflow and overflow while
                // setting up the search keys.
                // e.g. TINYINT > 200 or INT <= 6000000000

                // re-throw if not an overflow or underflow
                // currently, it's expected to always be an overflow or underflow
                if ((e.getInternalFlags() & (SQLException::TYPE_OVERFLOW | SQLException::TYPE_UNDERFLOW)) == 0) {
                    throw e;
                }

                // handle the case where this is a comparison, rather than equality match
                // comparison is the only place where the executor might return matching tuples
                // e.g. TINYINT < 1000 should return all values

                if ((localLookupType != INDEX_LOOKUP_TYPE_EQ) &&
                    (ctr == (activeNumOfSearchKeys - 1))) {
                    assert (localLookupType == INDEX_LOOKUP_TYPE_GT || localLookupType == INDEX_LOOKUP_TYPE_GTE);

                    if (e.getInternalFlags() & SQLException::TYPE_OVERFLOW) {
                        m_outputTable->insertTuple(tmptup);
                        return true;
                    } else if (e.getInternalFlags() & SQLException::TYPE_UNDERFLOW) {
                        searchKeyUnderflow = true;
                        break;
                    } else {
                        throw e;
                    }
                }
                // if a EQ comparision is out of range, then return no tuples
                else {
                    m_outputTable->insertTuple(tmptup);
                    return true;
                }
                break;
            }
        }
        VOLT_TRACE("Search key after substitutions: '%s'", searchKey.debugNoHeader().c_str());
    }

    if (m_numOfEndkeys != 0) {
        //
        // END KEY
        //
        endKey.setAllNulls();
        VOLT_DEBUG("Initial (all null) end key: '%s'", endKey.debugNoHeader().c_str());
        for (int ctr = 0; ctr < m_numOfEndkeys; ctr++) {
            NValue endKeyValue = m_endKeyArray[ctr]->eval(NULL, NULL);
            try {
                endKey.setNValue(ctr, endKeyValue);
            }
            catch (const SQLException &e) {