/*
 * Serialize more tuples to one or more output streams.
 * Returns a long for the remaining tuple count, -1 for an error.
 * Streams an int position array through the reused result buffer.
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeTableStreamSerializeMore
 * Signature: (JII[B)J;
 */
SHAREDLIB_JNIEXPORT jlong JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeTableStreamSerializeMore
(JNIEnv *env,
 jobject obj,
 jlong engine_ptr,
 jint tableId,
 jint streamType,
 jbyteArray serialized_buffers) {
    VOLT_DEBUG("nativeTableStreamSerializeMore in C++ called");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    jsize length = env->GetArrayLength(serialized_buffers);
    VOLT_DEBUG("nativeTableStreamSerializeMore: deserializing %d buffer bytes ...", (int) length);
    jbyte *bytes = env->GetByteArrayElements(serialized_buffers, NULL);
    ReferenceSerializeInput serialize_in(bytes, length);
    try {
        try {
            voltdb::TableStreamType tst = static_cast<voltdb::TableStreamType>(streamType);
            jlong tuplesRemaining = engine->tableStreamSerializeMore(tableId, tst, serialize_in);
            return tuplesRemaining;
        } catch (const SQLException &e) {
            throwFatalException("%s", e.message().c_str());
        }
    } catch (const FatalException &e) {
        topend->crashVoltDB(e);
    }
    // Won't get here.
    return TABLE_STREAM_SERIALIZATION_ERROR;
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeGetUSOForExportTable
 * Signature: (JLjava/lang/String;)[J
 */
SHAREDLIB_JNIEXPORT jlongArray JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeGetUSOForExportTable
  (JNIEnv *env, jobject obj, jlong engine_ptr, jbyteArray tableSignature) {

    VOLT_DEBUG("nativeGetUSOForExportTable in C++ called");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    jbyte *signatureChars = env->GetByteArrayElements(tableSignature, NULL);
    std::string signature(reinterpret_cast<char *>(signatureChars), env->GetArrayLength(tableSignature));
    env->ReleaseByteArrayElements(tableSignature, signatureChars, JNI_ABORT);
    try {
        jlong data[2];
        size_t ackOffset;
        int64_t seqNo;
        engine->getUSOForExportTable(ackOffset, seqNo, signature);
        data[0] = ackOffset;
        data[1] = seqNo;
        jlongArray retval = env->NewLongArray(2);
        env->SetLongArrayRegion(retval, 0, 2, data);
        return retval;
    }
    catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return NULL;
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeActivateTableStream
 * Signature: (JIIIJ[B)Z
 */
SHAREDLIB_JNIEXPORT jboolean JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeActivateTableStream(
        JNIEnv *env, jobject obj, jlong engine_ptr, jint tableId, jint streamType, jlong undoToken,
        jbyteArray serialized_predicates)
{
    VOLT_DEBUG("nativeActivateTableStream in C++ called");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);

    // deserialize predicates.
    jsize length = env->GetArrayLength(serialized_predicates);
    VOLT_DEBUG("deserializing %d predicate bytes ...", (int) length);
    jbyte *bytes = env->GetByteArrayElements(serialized_predicates, NULL);
    ReferenceSerializeInput serialize_in(bytes, length);
    try {
        try {
            voltdb::TableStreamType tableStreamType = static_cast<voltdb::TableStreamType>(streamType);
            bool success = engine->activateTableStream(tableId, tableStreamType, undoToken, serialize_in);
            env->ReleaseByteArrayElements(serialized_predicates, bytes, JNI_ABORT);
            VOLT_DEBUG("deserialized predicates (success=%d)", (int)success);
            return success;
        } catch (SerializableEEException &e) {
            engine->resetReusedResultOutputBuffer();
            e.serialize(engine->getExceptionOutputSerializer());
        }
    } catch (const FatalException& e) {
        topend->crashVoltDB(e);
    }

    return false;
}

/**
 * Serialize the result temporary table.
 * @param engine_ptr the VoltDBEngine pointer
 * @param table_id Id of the table to be serialized
 * @return serialized temporary table
*/
SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeSerializeTable(
        JNIEnv *env,
        jobject obj,
        jlong engine_ptr,
        jint table_id,
        jobject output_buffer,
        jint output_capacity) {
    //VOLT_DEBUG("nativeSerializeTable() start");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    if (engine == NULL) {
        VOLT_ERROR("The VoltDBEngine pointer is null!");
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    }
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    try {
        updateJNILogProxy(engine); //JNIEnv pointer can change between calls, must be updated
        void* data = env->GetDirectBufferAddress(output_buffer);
        ReferenceSerializeOutput out(data, output_capacity);

        bool success = engine->serializeTable(table_id, &out);

        if (!success) return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
        else return org_voltdb_jni_ExecutionEngine_ERRORCODE_SUCCESS;
    } catch (const FatalException &e) {
        topend->crashVoltDB(e);
    }
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeHashinate
 * Signature: (JI)I
 */
SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeHashinate(JNIEnv *env, jobject obj, jlong engine_ptr)
{
    VOLT_DEBUG("nativeHashinate in C++ called");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    assert(engine);
    try {
        updateJNILogProxy(engine); //JNIEnv pointer can change between calls, must be updated
        NValueArray& params = engine->getParameterContainer();
        Pool *stringPool = engine->getStringPool();
        deserializeParameterSet(engine->getParameterBuffer(), engine->getParameterBufferCapacity(), params, engine->getStringPool());
        HashinatorType hashinatorType = static_cast<HashinatorType>(voltdb::ValuePeeker::peekAsInteger(params[1]));
        boost::scoped_ptr<TheHashinator> hashinator;
        const char *configValue = static_cast<const char*>(voltdb::ValuePeeker::peekObjectValue(params[2]));
        switch (hashinatorType) {
        case HASHINATOR_LEGACY:
            hashinator.reset(LegacyHashinator::newInstance(configValue));
            break;
        case HASHINATOR_ELASTIC:
            hashinator.reset(ElasticHashinator::newInstance(configValue));
            break;
        default:
            return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
        }
        int retval =
            hashinator->hashinate(params[0]);
        stringPool->purge();
        return retval;
    } catch (const FatalException &e) {
        std::cout << "HASHINATE ERROR: " << e.m_reason << std::endl;
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    }
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

TEST_F(PersistentTableTest, TruncateTableTest) {
    VoltDBEngine* engine = getEngine();
    engine->loadCatalog(0, catalogPayload());
    PersistentTable *table = dynamic_cast<PersistentTable*>(
        engine->getTable("T"));
    ASSERT_NE(NULL, table);

    const int tuplesToInsert = 10;
    (void) tuplesToInsert;  // to make compiler happy
    ASSERT_EQ(1, table->allocatedBlockCount());
    bool addTuples = tableutil::addRandomTuples(table, tuplesToInsert);
    if(!addTuples) {
        assert(!"Failed adding random tuples");
    }
    size_t blockCount = table->allocatedBlockCount();

    table = dynamic_cast<PersistentTable*>(engine->getTable("T"));
    ASSERT_NE(NULL, table);
    ASSERT_EQ(blockCount, table->allocatedBlockCount());
    addTuples = tableutil::addRandomTuples(table, tuplesToInsert);
    if(!addTuples) {
        assert(!"Failed adding random tuples");
    }
    table->truncateTable(engine);

    // refresh table pointer by fetching the table from catalog as in truncate old table
    // gets replaced with new cloned empty table
    table = dynamic_cast<PersistentTable*>(engine->getTable("T"));
    ASSERT_NE(NULL, table);
    ASSERT_EQ(1, table->allocatedBlockCount());
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeLoadPlanFragment
 * Signature: (JJ[B)I
 */
SHAREDLIB_JNIEXPORT jint JNICALL
Java_org_voltdb_jni_ExecutionEngine_nativeLoadPlanFragment (
    JNIEnv *env,
    jobject obj,
    jlong engine_ptr,
    jbyteArray plan) {

    VOLT_DEBUG("nativeUnloadPlanFragment() start");

    // setup
    VoltDBEngine *engine = castToEngine(engine_ptr);
    assert(engine);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);

    //JNIEnv pointer can change between calls, must be updated
    updateJNILogProxy(engine);

    // convert java plan string to stdc++ string plan
    jbyte *str = env->GetByteArrayElements(plan, NULL);
    assert(str);

    // get the buffer to write results to
    engine->resetReusedResultOutputBuffer();
    ReferenceSerializeOutput* out = engine->getResultOutputSerializer();

    // output from the engine's loadFragment method
    int64_t fragId = 0;
    bool wasHit = 0;
    int64_t cacheSize = 0;

    // load
    int result = 1;
    try {
        result = engine->loadFragment(reinterpret_cast<char *>(str),
                                      env->GetArrayLength(plan),
                                      fragId, wasHit, cacheSize);
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    assert((result == 1) || (fragId != 0));

    // release plan memory
    env->ReleaseByteArrayElements(plan, str, JNI_ABORT);

    // write results back to java
    out->writeLong(fragId);
    out->writeBool(wasHit);
    out->writeLong(cacheSize);

    if (result == 1)
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    else
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_SUCCESS;
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeTick
 * Signature: (JJJ)V
 *
 * Called roughly every 1 second by the Java Runtime to allow the EE to do
 * periodic non-transactional work.
 *
 * @param env Pointer to the JNIEnv for this thread
 * @param obj Pointer to the object on which this method was called
 * @param engine_ptr Pointer to a VoltDBEngine instance
 * @param timeInMillis The current java timestamp (System.currentTimeMillis());
 * @param lastCommittedSpHandle The id of the last committed transaction.
 */
SHAREDLIB_JNIEXPORT void JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeTick
  (JNIEnv *env, jobject obj, jlong engine_ptr, jlong timeInMillis, jlong lastCommittedSpHandle) {
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    try {
        updateJNILogProxy(engine); //JNIEnv pointer can change between calls, must be updated
        engine->tick(timeInMillis, lastCommittedSpHandle);
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
}

void AbstractOperationPlanNode::loadFromJSONObject(PlannerDomValue obj)
{
    m_target_table_name = obj.valueForKey("TARGET_TABLE_NAME").asStr();
    VoltDBEngine* engine = ExecutorContext::getEngine();
    m_tcd = engine->getTableDelegate(m_target_table_name);
    if ( ! m_tcd) {
        VOLT_ERROR("Failed to retrieve target table from execution engine for PlanNode '%s'",
                   debug().c_str());
        //TODO: throw something
    }
}

 ParameterValueExpression::ParameterValueExpression(int value_idx)
     : AbstractExpression(EXPRESSION_TYPE_VALUE_PARAMETER),
     m_valueIdx(value_idx), m_paramValue()
 {
     VOLT_TRACE("ParameterValueExpression %d", value_idx);
     ExecutorContext* context = ExecutorContext::getExecutorContext();
     VoltDBEngine* engine = context->getEngine();
     assert(engine != NULL);
     NValueArray& params = engine->getParameterContainer();
     assert(value_idx < params.size());
     m_paramValue = &params[value_idx];
 };

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeActivateTableStream
 * Signature: (JII)Z
 */
SHAREDLIB_JNIEXPORT jboolean JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeActivateTableStream
  (JNIEnv *env, jobject obj, jlong engine_ptr, jint tableId, jint streamType) {
    VOLT_DEBUG("nativeActivateTableStream in C++ called");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    try {
        return engine->activateTableStream(tableId, static_cast<voltdb::TableStreamType>(streamType));
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return false;
}

/**
 * Releases all resources held in the execution engine.
 * @param engine_ptr the VoltDBEngine pointer to be destroyed
 * @return error code
*/
SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeDestroy(
    JNIEnv *env, jobject obj,
    jlong engine_ptr) {
    VoltDBEngine *engine = castToEngine(engine_ptr);
    static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    updateJNILogProxy(engine); //JNIEnv pointer can change between calls, must be updated
    if (engine == NULL) {
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    }
    delete engine;
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_SUCCESS;
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeQuiesce
 * Signature: (JJ)V
 *
 * Called to instruct the EE to reach an idle steady state.
 */
SHAREDLIB_JNIEXPORT void JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeQuiesce
  (JNIEnv *env, jobject obj, jlong engine_ptr, jlong lastCommittedTxnId)
{
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    try {
        // JNIEnv pointer can change between calls, must be updated
        updateJNILogProxy(engine);
        engine->quiesce(lastCommittedTxnId);
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
}

void AbstractPlanNode::setOutputTable(Table* table)
{
    PersistentTable* persistentTable = dynamic_cast<PersistentTable*>(table);
    if (persistentTable) {
        VoltDBEngine* engine = ExecutorContext::getEngine();
        TableCatalogDelegate* tcd = engine->getTableDelegate(persistentTable->name());
        m_outputTable.setTable(tcd);
    } else {
        TempTable* tempTable = dynamic_cast<TempTable*>(table);
        assert(tempTable);
        m_outputTable.setTable(tempTable);
    }
}

/**
 * Initializes the execution engine with given parameter.
 * @param enginePtr the VoltDBEngine pointer to be initialized
 * @param clusterId id of the cluster the execution engine belongs to
 * @param nodeId this id will be set to the execution engine
 * @return error code
*/
SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeInitialize(
    JNIEnv *env, jobject obj,
    jlong enginePtr,
    jint clusterIndex,
    jlong siteId,
    jint partitionId,
    jint hostId,
    jbyteArray hostname,
    jlong tempTableMemory,
    jint hashinatorType,
    jbyteArray hashinatorConfig)
{
    VOLT_DEBUG("nativeInitialize() start");
    VoltDBEngine *engine = castToEngine(enginePtr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    if (engine == NULL) {
        VOLT_ERROR("engine_ptr was NULL or invalid pointer");
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    }
    try {
        updateJNILogProxy(engine); //JNIEnv pointer can change between calls, must be updated

        jbyte *hostChars = env->GetByteArrayElements( hostname, NULL);
        std::string hostString(reinterpret_cast<char*>(hostChars), env->GetArrayLength(hostname));
        env->ReleaseByteArrayElements( hostname, hostChars, JNI_ABORT);
        jbyte *hashinatorConfigData = env->GetByteArrayElements(hashinatorConfig, NULL);
        // initialization is separated from constructor so that constructor
        // never fails.
        VOLT_DEBUG("calling initialize...");
        bool success =
                engine->initialize(clusterIndex,
                                   siteId,
                                   partitionId,
                                   hostId,
                                   hostString,
                                   tempTableMemory,
                                   (HashinatorType)hashinatorType,
                                   (char*)hashinatorConfigData);
        env->ReleaseByteArrayElements( hashinatorConfig, hashinatorConfigData, JNI_ABORT);
        if (success) {
            VOLT_DEBUG("initialize succeeded");
            return org_voltdb_jni_ExecutionEngine_ERRORCODE_SUCCESS;
        } else {
            throwFatalException("initialize failed");
            return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
        }
    } catch (const FatalException &e) {
        topend->crashVoltDB(e);
    }
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

/**
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeGetStats
 * Signature: (I[I)Z
 *
 * Called to retrieve statistics
 *
 * @param env Pointer to the JNIEnv for this thread
 * @param obj Pointer to the object on which this method was called
 * @param engine_ptr Pointer to a VoltDBEngine instance
 * @param selector Ordinal value from StatisticsSelectorType enum indicating the type of stats to retrieve
 * @param locatorsArray Java array of CatalogIds indicating what set of sources should the statistics be retrieved from.
 * @return Number of result tables, 0 on no results, -1 on failure.
 */
SHAREDLIB_JNIEXPORT jint JNICALL
Java_org_voltdb_jni_ExecutionEngine_nativeGetStats(JNIEnv *env, jobject obj,
                                                   jlong pointer, jint selector,
                                                   jintArray locatorsArray,
                                                   jboolean jinterval,
                                                   jlong now) {
    VoltDBEngine *engine = castToEngine(pointer);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    /*
     * Can't use the standard JNI EE error code here because this method
     * actually uses that integer to indicate the number of result tables.
     */
    int result = -1;

    //JNIEnv pointer can change between calls, must be updated
    updateJNILogProxy(engine);
    engine->resetReusedResultOutputBuffer();

    /*
     * Retrieve locators if any
     */
    int *locators = NULL;
    int numLocators = 0;
    if (locatorsArray != NULL) {
        locators = env->GetIntArrayElements(locatorsArray, NULL);
        if (locators == NULL) {
            env->ExceptionDescribe();
            return JNI_FALSE;
        }
        numLocators = env->GetArrayLength(locatorsArray);
        if (env->ExceptionCheck()) {
            env->ExceptionDescribe();
            env->ReleaseIntArrayElements(locatorsArray, locators, JNI_ABORT);
            return JNI_FALSE;
        }
    }
    const bool interval = jinterval == JNI_TRUE ? true : false;

    try {
        result = engine->getStats(static_cast<int>(selector), locators,
                                  numLocators, interval, now);
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }

    env->ReleaseIntArrayElements(locatorsArray, locators, JNI_ABORT);

    return static_cast<jint>(result);
}

void SwapTablesPlanNode::loadFromJSONObject(PlannerDomValue obj) {
    AbstractOperationPlanNode::loadFromJSONObject(obj);
    m_otherTargetTableName = obj.valueForKey("OTHER_TARGET_TABLE_NAME").asStr();
    loadStringArrayFromJSONObject("INDEXES", obj, m_theIndexes);
    loadStringArrayFromJSONObject("OTHER_INDEXES", obj, m_otherIndexes);

    VoltDBEngine* engine = ExecutorContext::getEngine();
    m_otherTcd = engine->getTableDelegate(m_otherTargetTableName);
    if ( ! m_otherTcd) {
        VOLT_ERROR("Failed to retrieve second target table from execution engine for PlanNode: %s",
                   debug().c_str());
        //TODO: throw something
    }

}

/**
 * Initializes the execution engine with given parameter.
 * @param enginePtr the VoltDBEngine pointer to be initialized
 * @param clusterId id of the cluster the execution engine belongs to
 * @param nodeId this id will be set to the execution engine
 * @return error code
*/
SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeInitialize(
    JNIEnv *env, jobject obj,
    jlong enginePtr,
    jint clusterIndex,
    jlong siteId,
    jint partitionId,
    jint hostId,
    jstring hostname,
    jlong tempTableMemory,
    jint totalPartitions)
{
    VOLT_DEBUG("nativeInitialize() start");
    VoltDBEngine *engine = castToEngine(enginePtr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    if (engine == NULL) {
        VOLT_ERROR("engine_ptr was NULL or invalid pointer");
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    }
    try {
        updateJNILogProxy(engine); //JNIEnv pointer can change between calls, must be updated

        const char *hostChars = env->GetStringUTFChars( hostname, NULL);
        std::string hostString(hostChars);
        env->ReleaseStringUTFChars( hostname, hostChars);
        // initialization is separated from constructor so that constructor
        // never fails.
        VOLT_DEBUG("calling initialize...");
        bool success =
                engine->initialize(clusterIndex,
                                   siteId,
                                   partitionId,
                                   hostId,
                                   hostString,
                                   tempTableMemory,
                                   totalPartitions);

        if (success) {
            VOLT_DEBUG("initialize succeeded");
            return org_voltdb_jni_ExecutionEngine_ERRORCODE_SUCCESS;
        } else {
            throwFatalException("initialize failed");
            return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
        }
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

void AbstractPlanNode::setInputTables(const vector<Table*>& val)
{
    size_t ii = val.size();
    m_inputTables.resize(ii);
    while (ii--) {
        PersistentTable* persistentTable = dynamic_cast<PersistentTable*>(val[ii]);
        if (persistentTable) {
            VoltDBEngine* engine = ExecutorContext::getEngine();
            assert(engine);
            TableCatalogDelegate* tcd = engine->getTableDelegate(persistentTable->name());
            m_inputTables[ii].setTable(tcd);
        } else {
            TempTable* tempTable = dynamic_cast<TempTable*>(val[ii]);
            assert(tempTable);
            m_inputTables[ii].setTable(tempTable);
        }
    }
}

SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeAntiCacheMergeBlocks (
        JNIEnv *env,
        jobject obj,
        jlong engine_ptr,
        jint tableId) {

    int retval = org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    VOLT_DEBUG("nativeAntiCacheMergeBlocks() start");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    if (engine == NULL) return (retval);

    try {
        retval = engine->antiCacheMergeBlocks(static_cast<int32_t>(tableId));
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return (retval);
}