/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeExportAction
 *
 * @param ackAction  true if this call contains an ack
 * @param pollAction true if this call requests a poll
 * @param syncAction true if the stream offset being set for a table
 * @param ackOffset  if acking, the universal stream offset being acked/released
 * @param tableSignature    Signature of the table to which the Export action applies
 *
 * @return the universal stream offset for the last octet in any
 * returned poll results (returned via the query results buffer).  On
 * any error this will be less than 0.  For any call with no
 * pollAction, any value >= 0 may be ignored.
 */
SHAREDLIB_JNIEXPORT jlong JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeExportAction
  (JNIEnv *env,
   jobject obj,
   jlong engine_ptr,
   jboolean syncAction,
   jlong ackOffset,
   jlong seqNo,
   jbyteArray tableSignature) {
    VOLT_DEBUG("nativeExportAction 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 {
        try {
            engine->resetReusedResultOutputBuffer();
            return engine->exportAction(syncAction,
                                        static_cast<int64_t>(ackOffset),
                                        static_cast<int64_t>(seqNo),
                                        signature);
        } catch (SQLException e) {
            throwFatalException("%s", e.message().c_str());
        }
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return 0;
}

/*
 * Serialize more tuples to one or more output streams.
 * Returns a long for the remaining tuple count, -1 when done, or -2 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, but -2 is an error.
    return -2;
}

/**
 * Executes a plan fragment with the given parameter set.
 * @param engine_ptr the VoltDBEngine pointer
 * @param plan_fragment_id ID of the plan fragment to be executed.
 * @return error code
*/
SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeExecutePlanFragment (
        JNIEnv *env,
        jobject obj,
        jlong engine_ptr,
        jlong plan_fragment_id,
        jint outputDependencyId,
        jint inputDependencyId,
        jlong txnId,
        jlong lastCommittedTxnId,
        jlong undoToken) {
    VOLT_DEBUG("nativeExecutePlanFragment() start");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    assert(engine);
    try {
        updateJNILogProxy(engine); //JNIEnv pointer can change between calls, must be updated
        engine->setUndoToken(undoToken);
        engine->resetReusedResultOutputBuffer();
        NValueArray &params = engine->getParameterContainer();
        Pool *stringPool = engine->getStringPool();
        const int paramcnt = deserializeParameterSet(engine->getParameterBuffer(), engine->getParameterBufferCapacity(), params, engine->getStringPool());
        engine->setUsedParamcnt(paramcnt);
        const int retval = engine->executeQuery(plan_fragment_id, outputDependencyId, inputDependencyId, params, txnId, lastCommittedTxnId, true, true);
        stringPool->purge();
        return retval;
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

/**
 * 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 (FatalException e) {
        topend->crashVoltDB(e);
    }
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeAntiCacheReadBlocks (
        JNIEnv *env,
        jobject obj,
        jlong engine_ptr,
        jint tableId,
        jshortArray blockIdsArray) {
    
    int retval = org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    VOLT_DEBUG("nativeAntiCacheReadBlocks() start");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    if (engine == NULL) return (retval);
    
    try {
        jsize numBlockIds = env->GetArrayLength(blockIdsArray);
        jshort *_blockIds = env->GetShortArrayElements(blockIdsArray, NULL);
        if (_blockIds == NULL) {
            VOLT_ERROR("No evicted blockIds were given to the EE");
            return (retval);
        }
        // XXX: Is this necessary?
        uint16_t *blockIds = new uint16_t[numBlockIds];
        for (int ii = 0; ii < numBlockIds; ii++) {
            blockIds[ii] = _blockIds[ii];
        } // FOR
        
        retval = engine->antiCacheReadBlocks(static_cast<int32_t>(tableId), static_cast<int>(numBlockIds), blockIds);
        
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return (retval);
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeExportAction
 *
 * @param ackAction  true if this call contains an ack
 * @param pollAction true if this call requests a poll
 * @param syncAction true if the stream offset being set for a table
 * @param ackOffset  if acking, the universal stream offset being acked/released
 * @param tableId    the table ID to which the Export action applies
 *
 * @return the universal stream offset for the last octet in any
 * returned poll results (returned via the query results buffer).  On
 * any error this will be less than 0.  For any call with no
 * pollAction, any value >= 0 may be ignored.
 */
SHAREDLIB_JNIEXPORT jlong JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeExportAction
  (JNIEnv *env,
   jobject obj,
   jlong engine_ptr,
   jboolean ackAction,
   jboolean pollAction,
   jboolean resetAction,
   jboolean syncAction,
   jlong ackOffset,
   jlong seqNo,
   jlong tableId) {
    VOLT_DEBUG("nativeExportAction in C++ called");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    try {
        try {
            engine->resetReusedResultOutputBuffer();
            return engine->exportAction(ackAction, pollAction, resetAction, syncAction,
                                        static_cast<int64_t>(ackOffset),
                                        static_cast<int64_t>(seqNo),
                                        static_cast<int64_t>(tableId));
        } catch (SQLException e) {
            throwFatalException("%s", e.message().c_str());
        }
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return 0;
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeActivateTableStream
 * Signature: (JIII[B)Z
 */
SHAREDLIB_JNIEXPORT jboolean JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeActivateTableStream(
        JNIEnv *env, jobject obj, jlong engine_ptr, jint tableId, jint streamType,
        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, 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;
}

/*
 * 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);
    }
}

/*
 * 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);
    }
}

/**
 * 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);
}

/*
 * Executes a plan fragment of an adhoc query.
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeExecuteCustomPlanFragment
 * Signature: (JLjava/lang/String;JJJ)I
 */
SHAREDLIB_JNIEXPORT jint JNICALL
Java_org_voltdb_jni_ExecutionEngine_nativeExecuteCustomPlanFragment (
    JNIEnv *env,
    jobject obj,
    jlong engine_ptr,
    jstring plan,
    jint outputDependencyId,
    jint inputDependencyId,
    jlong txnId,
    jlong lastCommittedTxnId,
    jlong undoToken) {
    int retval = org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;

    VOLT_DEBUG("nativeExecuteCustomPlanFragment() 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);
    engine->resetReusedResultOutputBuffer();
    engine->setUndoToken(undoToken);
    static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    Pool *stringPool = engine->getStringPool();

    // convert java plan string to stdc++ string plan
    const char *str = static_cast<const char*>(env->GetStringUTFChars(plan,
                                                                      NULL));
    assert(str);
    string cppplan = str;
    env->ReleaseStringUTFChars(plan, str);

    // execute
    engine->setUsedParamcnt(0);
    try {
        retval = engine->executePlanFragment(cppplan, outputDependencyId,
                                             inputDependencyId, txnId,
                                             lastCommittedTxnId);
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    // cleanup
    stringPool->purge();

    return retval;
}

/**
 * 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 signalHandler(int signum, siginfo_t *info, void *context) {
    if (currentVM == NULL || currentEngine == NULL)
        return;

    char err_msg[128];
    snprintf(err_msg, 128, "SIGSEGV caught: signal number %d, error value %d,"
             " signal code %d\n\n", info->si_signo, info->si_errno,
             info->si_code);
    std::string message = err_msg;
    message.append(currentEngine->debug());

    JNIEnv *env;
    if (currentVM->AttachCurrentThread((void **)(void *)&env, NULL) != 0)
        exit(-1);
    Topend *topend = static_cast<JNITopend*>(currentEngine->getTopend())->updateJNIEnv(env);
    topend->crashVoltDB(SegvException(message.c_str(), context, __FILE__, __LINE__));
    currentVM->DetachCurrentThread();
}

/**
 * This method is called to initially load table data.
 * @param pointer the VoltDBEngine pointer
 * @param table_id catalog ID of the table
 * @param serialized_table the table data to be loaded
*/
SHAREDLIB_JNIEXPORT jint JNICALL
Java_org_voltdb_jni_ExecutionEngine_nativeLoadTable (
    JNIEnv *env, jobject obj, jlong engine_ptr, jint table_id,
    jbyteArray serialized_table, jlong spHandle, jlong lastCommittedSpHandle,
    jboolean returnUniqueViolations, jlong undoToken)
{
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    if (engine == NULL) {
        return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
    }

    engine->resetReusedResultOutputBuffer();
    engine->setUndoToken(undoToken);

    //JNIEnv pointer can change between calls, must be updated
    updateJNILogProxy(engine);
    VOLT_DEBUG("loading table %d in C++...", table_id);

    // deserialize dependency.
    jsize length = env->GetArrayLength(serialized_table);
    VOLT_DEBUG("deserializing %d bytes ...", (int) length);
    jbyte *bytes = env->GetByteArrayElements(serialized_table, NULL);
    ReferenceSerializeInput serialize_in(bytes, length);
    try {
        try {
            bool success = engine->loadTable(table_id, serialize_in,
                                             spHandle, lastCommittedSpHandle,
                                             returnUniqueViolations);
            env->ReleaseByteArrayElements(serialized_table, bytes, JNI_ABORT);
            VOLT_DEBUG("deserialized table");

            if (success)
                return org_voltdb_jni_ExecutionEngine_ERRORCODE_SUCCESS;
        } catch (const SerializableEEException &e) {
            engine->resetReusedResultOutputBuffer();
            e.serialize(engine->getExceptionOutputSerializer());
        }
    } catch (const FatalException &e) {
        topend->crashVoltDB(e);
    }

    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

/*
 * Class:     org_voltdb_jni_ExecutionEngine
 * Method:    nativeUpdateHashinator
 * Signature: (JI)I
 */
SHAREDLIB_JNIEXPORT void JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeUpdateHashinator(JNIEnv *env, jobject obj, jlong engine_ptr)
{
    VOLT_DEBUG("nativeUpdateHashinator in C++ called");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    assert(engine);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    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[0]));
        const char *configValue = static_cast<const char*>(voltdb::ValuePeeker::peekObjectValue(params[1]));
        engine->updateHashinator(hashinatorType, configValue);
        stringPool->purge();
    } catch (const FatalException &e) {
        topend->crashVoltDB(e);
    }
}

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);
}

/**
 * Load the system catalog for this engine.
 * @param engine_ptr the VoltDBEngine pointer
 * @param serialized_catalog the root catalog object serialized as text strings.
 * human-readable text strings separated by line feeds.
 * @return error code
*/
SHAREDLIB_JNIEXPORT jint JNICALL
Java_org_voltdb_jni_ExecutionEngine_nativeUpdateCatalog(
    JNIEnv *env, jobject obj,
    jlong engine_ptr, jlong timestamp, jbyteArray catalog_diffs) {
    VOLT_DEBUG("nativeUpdateCatalog() start");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    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;
    }

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

    //copy to std::string. utf_chars may or may not by a copy of the string
    jbyte* utf_chars = env->GetByteArrayElements(catalog_diffs, NULL);
    string str(reinterpret_cast<char*>(utf_chars), env->GetArrayLength(catalog_diffs));
    env->ReleaseByteArrayElements(catalog_diffs, utf_chars, JNI_ABORT);
    VOLT_DEBUG("calling loadCatalog...");

    try {
        bool success = engine->updateCatalog( timestamp, str);

        if (success) {
            VOLT_DEBUG("updateCatalog succeeded");
            return org_voltdb_jni_ExecutionEngine_ERRORCODE_SUCCESS;
        }
    } catch (const SerializableEEException &e) {
        engine->resetReusedResultOutputBuffer();
        e.serialize(engine->getExceptionOutputSerializer());
    } catch (const FatalException &fe) {
        if (topend != NULL) {
            topend->crashVoltDB(fe);
        }
    }

    VOLT_ERROR("updateCatalog failed");
    return org_voltdb_jni_ExecutionEngine_ERRORCODE_ERROR;
}

SHAREDLIB_JNIEXPORT jint JNICALL Java_org_voltdb_jni_ExecutionEngine_nativeAntiCacheEvictBlock (
        JNIEnv *env,
        jobject obj,
        jlong engine_ptr,
        jint tableId,
        jlong blockSize) {
         
    int retval = -1;
    VOLT_DEBUG("nativeAntiCacheEvictBlocks() start");
    VoltDBEngine *engine = castToEngine(engine_ptr);
    Topend *topend = static_cast<JNITopend*>(engine->getTopend())->updateJNIEnv(env);
    if (engine == NULL) return (retval);

    engine->resetReusedResultOutputBuffer();
    
    try {
        retval = engine->antiCacheEvictBlock(static_cast<int32_t>(tableId), static_cast<long>(blockSize));
    } catch (FatalException e) {
        topend->crashVoltDB(e);
    }
    return (retval);
}