#ifndef DOLPHINDB_SWORDFISH_H_
#define DOLPHINDB_SWORDFISH_H_

#include "CoreConcept.h"
#include "Concepts.h"
#include "DolphinString.h"
#include "SysIO.h"

#include <exception>
#include <memory>
#include <string>
#include <unordered_map>
#include <vector>
#include <type_traits>

namespace ddb {

/**
 * @defgroup Version Version Macros
 * @{
 */
/** Library major version. */
#define SWORDFISH_VERSION_MAJOR 3
/** Library minor version. */
#define SWORDFISH_VERSION_MINOR 3
/** Library patch version. */
#define SWORDFISH_VERSION_PATCH 0
/** @} */

//==============================================================================
// DolphinDB Runtime Utilities
//==============================================================================

class SWORDFISH_API DolphinDBLib {
public:
    /**
     * @brief Initialize DolphinDB runtime.
     *
     * @warning This function (or the following one) MUST be called before
     * using any other API functions.
     */
    static void initializeRuntime();
    /**
     * @brief Initialize DolphinDB runtime with command line arguments.
     */
    static void initializeRuntime(int argc, char **argv);

    /**
     * @brief Finalize DolphinDB runtime.
     *
     * @warning If DolphinDB runtime is initialized, MUST call this function to
     * finalize it. And once this function is called, can NOT call other API
     * functions (include `initializeRuntime`).
     */
    static void finalizeRuntime();

    /** Get a C-style string representation of the library version. */
    static const char *getVersionString();

    /**
     * @brief Create a Session, then can use it to run DolphinDB builtin functions.
     */
    static SessionSP createSession();

    /**
     * @brief Run a piece of DolphinDB script.
     */
    static ConstantSP execute(const std::string &script);

    /**
     * @brief Run a piece of DolphinDB script in the given session.
     */
    static ConstantSP execute(SessionSP session, const std::string &script);

    /**
     * @brief The following two functions are basically the same as the above ones,
     * except that they provide an @c env parameter. The free variables used in @c script
     * will be looked up in @c env
     *
     * @param env A map used to represent free variables in @c script. The key is variable name.
     */
    static ConstantSP execute(const std::string &script, const std::unordered_map<std::string, ConstantSP> &env);
    static ConstantSP execute(SessionSP session, const std::string &script,
                              const std::unordered_map<std::string, ConstantSP> &env);

    /**
     * @brief Run a builtin function of DolphinDB.
     */
    static ConstantSP execute(const std::string &funcName, std::vector<ConstantSP> &arguments);

     /**
     * @brief Run a function of DolphinDB in the given session.
     */
    static ConstantSP execute(SessionSP session, const std::string &funcName, std::vector<ConstantSP> &arguments);
};

class ReactiveState {
public:
	virtual ~ReactiveState(){}
	virtual IO_ERR snapshotState(const DataOutputStreamSP& out) = 0;
	virtual IO_ERR restoreState(const DataInputStreamSP& in) = 0;
	virtual void append(Heap* heap, const ConstantSP& keyIndex) = 0;
	virtual void addKeys(int count) = 0;
	virtual void removeKeys(const vector<int>& keyIndices) = 0;
	virtual void reserveKeys(int count) = 0;
	virtual void getMemoryUsed(long long& fixedMemUsed, long long& variableMemUsedPerKey) = 0;
	// clearkeys() will NOT remove keys, but reset their state to initial state, used in segmentby
	// add clearKeys for [cumsum, cummax, cummin, cumcount, cumavg, cumstd, cumvar, cumstdp, cumvarp], others not support.
	// for those ReactiveState which NOT support segmentby, throw an execption
	virtual void clearKeys(const ConstantSP& keyIndex) {
		throw RuntimeException("ReactiveState NOT supported clearKeys()");
	}
	virtual void setTable(const TableSP& table){
		table_ = table;
	}
	TableSP getTable() const {
		return table_;
	}
	virtual void setSession(const SessionSP& session){
		session_ = session;
	}

	virtual long long getAdditionalMemoryUsed(){ return 0; }
    virtual bool isTimeMovingFunction() {
        return false;
    }
    virtual void getStateDetails(int key, DolphinString &str) {}
protected:
	template<class T>
	void setData(int outputColIndex, INDEX* indices, int count, T* buf){
		ConstantSP result = table_->getColumn(outputColIndex);
		if(LIKELY(result->isFastMode())){
			T* data = (T*)result->getDataArray();
			for(int i=0; i<count; ++i){
				data[indices[i]] = buf[i];
			}
		}
		else{
			T** dataSegment = (T**)result->getDataSegment();
			int segmentSizeInBit = result->getSegmentSizeInBit();
			int segmentMask = (1<<segmentSizeInBit) - 1;
			for(int i=0; i<count; ++i){
				dataSegment[indices[i] >> segmentSizeInBit][indices[i] & segmentMask] = buf[i];
			}
		}
	}

	template<class T>
	void setData(int outputColIndex, INDEX* indices, int count, const T& value){
		ConstantSP result = table_->getColumn(outputColIndex);
		if(LIKELY(result->isFastMode())){
			T* data = (T*)result->getDataArray();
			for(int i=0; i<count; ++i){
				data[indices[i]] = value;
			}
		}
		else {
			T** dataSegment = (T**)result->getDataSegment();
			int segmentSizeInBit = result->getSegmentSizeInBit();
			int segmentMask = (1<<segmentSizeInBit) - 1;
			for(int i=0; i<count; ++i){
				dataSegment[indices[i] >> segmentSizeInBit][indices[i] & segmentMask] = value;
			}
		}
	}

	void setString(int outputColIndex, INDEX* indices, int count, DolphinString* buf);
	void setString(int outputColIndex, INDEX* indices, int count, const DolphinString& value);

	template<class T>
	inline void reserveElements(vector<T>& data, int keyIndices) {
		data.reserve(keyIndices);
	}

	template<class T>
	void removeElements(vector<T>& data, const vector<int>& keyIndices){
		if(data.empty())
			return;
	    size_t dstCursor = keyIndices[0];
	    size_t srcCursor = keyIndices[0] + 1;
	    size_t indicesCursor = 1;
	    size_t count = data.size();
	    size_t indicesCount = keyIndices.size();
	    while(srcCursor < count){
	        if (indicesCursor < indicesCount) {
	            while (srcCursor < (size_t)keyIndices[indicesCursor]) {
	                data[dstCursor++] = std::move(data[srcCursor++]);
	            }
	            ++srcCursor;
	            ++indicesCursor;
	        }
	        else {
	            while (srcCursor < count) {
	                data[dstCursor++] = std::move(data[srcCursor++]);
	            }
	        }
	    }
		data.erase(data.begin() + count - indicesCount, data.end());
	    //data.resize(count - indicesCount);
	}

protected:
	SessionSP session_;
	TableSP table_;
};
typedef SmartPointer<ReactiveState> ReactiveStateSP;

struct SWORDFISH_API ReactiveStateFactory {
    static ReactiveStateSP createTransformReactiveState(const SQLContextSP &context, const ObjectSP &obj,
                                                        vector<int> outputColIndex, Heap *heap, ConstantSP *result);
    static ReactiveStateSP createConstantReactiveState(int outputColIndex, const ObjectSP& cons);
    static ReactiveStateSP createTalibNullReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTalibReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createKamaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createSmaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createWmaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createDemaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTemaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTrimaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createT3ReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createLinearTimeTrendReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createEmaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createWilderReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createGemaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMoveReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createPrevReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createDeltasReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createRatiosReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createPercentChangeReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createFfillReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createIterateReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createEwmMeanReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createEwmVarReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createEwmStdReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createEwmCovReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createEwmCorrReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMcountReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMavgReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createStateMavgReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMsumReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMprodReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMvarReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMvarpReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMstdReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMstdpReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMskewReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMkurtosisReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMminReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMmaxReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMiminReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMimaxReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMfirstReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMlastReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMmedReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMpercentileReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMrankReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMcorrReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMcovarReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMbetaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMwsumReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMwavgReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMslrReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumavgReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumsumReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumprodReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumcountReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumvarReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumminReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCummaxReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumvarpReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumstdReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumstdpReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumcorrReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumcovarReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumbetaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumwsumReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumwavgReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumfirstNotReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumlastNotReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCummedReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumpercentileReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumPositiveStreakReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumnuniqueReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createNewCumnuniqueReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMovingReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices, SQLContextSP &context, Heap *heap);
    static ReactiveStateSP createWindowReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMoveReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingsumReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingsum2ReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingminReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingmaxReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingrankReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingmedReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices, SQLContextSP &context, Heap *heap);
    static ReactiveStateSP createTMovingpercentileReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices, SQLContextSP &context, Heap *heap);
    static ReactiveStateSP createTMovingprodReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingcorrReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingbetaReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingcovarReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingwsumReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingwavgReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingskewReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingkurtosisReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingavgReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingcountReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingvarReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingvarpReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingstdReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingstdpReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingfirstReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovinglastReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMmadReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumsumTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumavgTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumstdTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumstdpTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumvarTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumvarpTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumbetaTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumcorrTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumcovarTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumwsumTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumskewTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createCumkurtosisTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMsumTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMavgTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMstdpTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMstdTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMvarpTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMvarTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMcorrTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMbetaTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMcovarTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMwsumTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMskewTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMkurtosisTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingsumTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingavgTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingstdpTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingstdTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingvarpTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingvarTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingskewTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingkurtosisTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingcorrTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingbetaTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingcovarTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingwsumTopNReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createDynamicGroupCumcountReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createDynamicGroupCumsumReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createSegmentbyReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createPrevStateReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTopRangeReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createLowRangeReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMmaxPositiveStreakReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createSumbarsReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTrueRangeReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createConditionalIterateReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createStateIterateReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createMovingWindowDataReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createTMovingWindowDataReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createGenericStateIterateReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createGenericTStateIterateReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
    static ReactiveStateSP createAccumulateReactiveState(const vector<ObjectSP>& args, const vector<int>& inputColIndices, const vector<DATA_TYPE>& inputColTypes, const vector<int>& outputColIndices);
};

namespace oltp {

using std::string;
using std::vector;
using std::shared_ptr;
using std::pair;

class DBImpl;
class ConnectionContext;

class StringView;
struct RawTableBuffer;
struct TableDescription;
typedef SmartPointer<TableDescription> TableDescriptionSP;

//==============================================================================
// Database
//==============================================================================

struct DBOption {
    /**
     * If true, will open the database in read-only mode, means only allow to
     * query, not allow to write data (insert/delete/update) or create/drop
     * table.
     *
     * A database can open in read-only mode multiple times simultaneously
     * (in the same process or a different process). But once opened in write
     * mode, no one else can open the database at the same time.
     *
     * @attention If not necessary, do NOT open a database simultaneously
     * (in read-only mode) in the same process, because every time the database
     * is opened, all data is loaded into memory.
     */
    bool readOnly = false;

    /**
     * If false, all data is stored in memory and will be lost once the database
     * is closed.
     * If true, will turn on write-ahead logging for persistence.
     *
     * @attention If do not want to lose any data, MUST enable WAL.
     */
    bool enableWAL = true;
    /**
     * This option is meaningful only when `enableWAL` is true.
     *
     * If true, before a write-transaction commit, MUST wait until all
     * write-ahead logs of this transaction are persisted.
     *
     * If false, database can survive from process crash, but can not survive
     * from OS crash. In this mode, performance is better typically.
     *
     * @attention If do not want to lose any data even when OS crash, set this
     * option to true. If wants better performance and do not care OS crash, set
     * this option to false;
     */
    bool syncOnTxnCommit = false;

    /**
     * If true, will do checkpoint automatically according to
     * `checkpointThreshold` and `checkpointInterval`.
     */
    bool enableCheckpoint = true;
    /**
     * Force to do checkpoint when the size of write-ahead logs exceed # MiB.
     */
    size_t checkpointThreshold = 100;
    /**
     * Force to do checkpoint every # seconds.
     */
    size_t checkpointInterval = 60;
};

/**
 * The database object. Open (or create) a database, then connect to the
 * database to operate it.
 *
 * @attention Before closing a DB (call destructor), make sure all clients
 * (Connection) which connect to the DB are inactive.
 */
class SWORDFISH_API DB {
public:
    /**
     * @brief Open a database with specific option.
     * @param path The directory of the database.
     */
    explicit DB(const string &path, const DBOption& option);

    ~DB();

    DB(const DB &) = delete;
    DB& operator=(const DB &) = delete;

private:
    friend class Connection;
    shared_ptr<DBImpl> impl_;
};

//==============================================================================
// Connection
//==============================================================================

/**
 * A connection is a client, connect to a database.
 *
 * @warning Do NOT use a same connection object in multiple threads.
 *
 * @attention Make sure the DB object is valid when the connection is active.
 */
class SWORDFISH_API Connection {
public:
    explicit Connection(DB &db);

    ~Connection();

    Connection(const Connection &) = delete;
    Connection& operator=(const Connection &) = delete;

    /**
     * Execute a SQL or DolphinDB script, e.g.,
     * execute("select * from table order by a"),
     * execute("a = 1; b = 2; a + b").
     *
     * @warning Do NOT use this API to create/drop table, use DDL API instead.
     */
    ConstantSP execute(const string &script);

    //--------------------------- Transaction API ---------------------------//

    /**
     * Begin a new transaction, then can execute multiple DML operations
     * (query/insert/delete/update) atomically. All DML in a transaction context
     * will succeed (if commit), or fail as nothing happened (if rollback).
     *
     * @warning Can NOT execute any DDL operations (create/drop table) in a
     * transaction context.
     *
     * @warning If already in an active transaction context, can NOT begin a new
     * transaction, otherwise RuntimeException will be thrown.
     */
    void beginTransaction();
    void commit();
    void rollback();

    /**
     * A utility method used to execute mutiple DML operations in the same
     * transaction.
     *
     * Usage:
     * @code {.cpp}
     *   conn.transaction([&]() {
     *       conn.insert(...);
     *       conn.update(...);
     *       conn.update(...);
     *   });
     * @endcode
     *
     * @tparam Code A callable type, e.g., lambda.
     * @param code DML operations want to execute.
     *
     * @note No need to catch `OLTPNeedRetryException` inside `code`.
     */
    template <typename Code>
    void transaction(Code code);

    /** Check whether is in an active transaction context. */
    bool isInActiveTransactionContext();

    //------------------------------- DDL API -------------------------------//

    /**
     * @brief Create a table in current database.
     *
     * @param tableName Table name.
     * @param colDesc Columns description.
     * @param primaryKey Columns use as primary index key.
     * @param secondaryKeys Secondary index keys.
     *
     * @note Primary index is required, secondary index is optional and can have
     * multiple secondary indexes.
     *
     * @note Description of secondary index key `pair<bool, vector<string>>`:
     * first => whether it is a unique index. second => columns use as index
     * key.
     */
    void createTable(const string &tableName, const vector<ColumnDesc> &colDesc,
                     const vector<string> &primaryKey,
                     const vector<pair<bool, vector<string>>> &secondaryKeys);

    /**
     * @brief Drop a table of current database.
     *
     * @param tableName Which table.
     */
    void dropTable(const string &tableName);

    //------------------------- High Level DML API -------------------------//

    /**
     * @brief Query data from the table.
     *
     * @param tableName Which table.
     * @param columnNames Columns (all columns if it's empty) wants to select.
     * @param filters Where clause in SQL, select all if empty. Use
     * `makeFilters` to generate filters, e.g., `makeFilters("a = 1")`,
     * `makeFilters("a = 1, b > 2")`.
     * @return A `Table` object represent the query result.
     *
     * @throw OLTPNeedRetryException
     * @throw OLTPNotRetryException
     *
     * @note This API equivalent to a simple SQL like
     * "select xxx form table where a = 2". If wants to execute complex SQL,
     * e.g., has "order by", "group by" and so on, use `execute` API.
     */
    TableSP query(const string &tableName, const vector<string> &columnNames,
                  const vector<ObjectSP> &filters);

    /**
     * @brief Insert data into the table.
     *
     * @param tableName Which table.
     * @param data A `Table` object represent data wants to insert.
     * @return How many rows were inserted.
     *
     * @throw OLTPNeedRetryException
     * @throw OLTPNotRetryException
     */
    size_t insert(const string &tableName, const TableSP &data);

    /**
     * @brief Delete data from the table.
     *
     * @param tableName Which table.
     * @param filters Where clause in SQL, select all if empty. Use
     * `makeFilters` to generate filters.
     * @return How many rows were deleted.
     *
     * @throw OLTPNeedRetryException
     * @throw OLTPNotRetryException
     */
    size_t remove(const string &tableName, const vector<ObjectSP> &filters);

    /**
     * @brief Update data in the table.
     *
     * @param tableName Which table.
     * @param updateExpr How to update it. Use `makeColumnUpdateDefs` to
     * generate update expression, e.g., `makeColumnUpdateDefs("a = a +1")`.
     * @param filters Where clause in SQL, select all if empty. Use
     * `makeFilters` to generate filters.
     * @return How many rows were updated.
     *
     * @throw OLTPNeedRetryException
     * @throw OLTPNotRetryException
     */
    size_t update(const string &tableName,
                  const vector<ColumnDefSP> &updateExpr,
                  const vector<ObjectSP> &filters);

    //------------------------- Low Level DML API -------------------------//

    /**
     * This is low level query API. Caller allocates memory
     * (see `RawTableBuffer`) in advance, and will write result to this memory
     * region. `RawTableBuffer::rows` indicates maximum number of rows to
     * return. Caller MUST make sure a tuple in `RawTableBuffer` is big enough
     * to store a row of the query result.
     *
     * @note This API has ~20% performance improvement compared to high level
     * API. If not necessary, this API is NOT recommended.
     *
     * @param columnNames Columns (all columns if it's empty) wants to select.
     * @param result The buffer use to store query result.
     * @return How many rows were wrote to the buffer.
     *
     * @throw OLTPNeedRetryException
     * @throw OLTPNotRetryException
     *
     * @note This API equivalent to a simple SQL like
     * "select xxx form table where a = 2". If wants to execute complex SQL,
     * e.g., has "order by", "group by" and so on, use `execute` API.
     */
    size_t query(const string &tableName, const vector<string> &columnNames,
                 const vector<ObjectSP> &filters, const RawTableBuffer &result);

    /**
     * This is low level insert API. Caller MUST make sure the schema of data
     * passed in matches the schema of the table.
     *
     * @warning This API has NO performance improvement compared to high level
     * API. If not necessary, this API is NOT recommended.
     *
     * @throw OLTPNeedRetryException
     * @throw OLTPNotRetryException
     */
    size_t insert(const string &tableName, const RawTableBuffer &data);

    //-------------------------------- Misc --------------------------------//

    /**
     * @brief Request to do checkpoint.
     *
     * @param force Force checkpoint, even if no changes have been made.
     * @param wait Whether wait checkpoint finish.
     */
    void requestCheckpoint(bool force, bool wait);

    SessionSP getCurrentSession() const;

    /**
     * Generate filters used in SQL `where` clause, e.g., "a = 1",
     * "a > 3, b = 2".
     *
     * @param text String text. Select all if empty.
     *
     * @note Do NOT use "a > 3 and b = 2", instead, use "a > 3, b = 2".
     */
    vector<ObjectSP> makeFilters(const string &text);

    /**
     * Generate update expression used in SQL update, e.g., "a = 1",
     * "a = a + 1, b = b - 1".
     */
    vector<ColumnDefSP> makeColumnUpdateDefs(const string &text);

    /** Get name of all tables in current database. */
    vector<string> listAllTable();

    /**
     * Get the table info of a specific table, or NULL-SP if it cannot be found.
     */
    TableDescriptionSP getTableInfo(const string &tableName);

private:
    shared_ptr<ConnectionContext> context_;
};

template <typename Code>
void Connection::transaction(Code code) {
    for (;;) {
        try {
            this->beginTransaction();
            code();
            this->commit();
            break;
        } catch (const OLTPNeedRetryException &ex) {
            // sleep a while to avoid conflict again
            Util::sleep(/*milliSeconds*/2);
            continue;
        } catch (...) {
            this->rollback();
            throw;
        }
    }
}

//==============================================================================
// RawTupleBuffer / RawTableBuffer
//==============================================================================

/**
 * @attention Use `RawTupleBufferGenerator` to generator a `RawTupleBuffer`.
 */
struct RawTupleBuffer {
    char *buffer;
};

/**
 * @attention Use `RawTableBufferGenerator` to generator a `RawTableBuffer`.
 */
struct RawTableBuffer {
    /** Maximum tuples this table can store. */
    int rows;
    RawTupleBuffer *tuples;
};

class SWORDFISH_API RawTupleBufferReader {
public:
    RawTupleBufferReader();
    RawTupleBufferReader(const RawTupleBuffer &tuple);
    ~RawTupleBufferReader();

    void reset(const RawTupleBuffer &tuple);

    char readBool();
    char readChar();
    short readShort();
    int readInt();
    long long readLong();
    float readFloat();
    double readDouble();
    StringView readString();

private:
    template <typename T>
    T read();

    const char *buffer_ = nullptr;
    int cursor_ = 0;
};

/// @cond
/**
 * @attention Do NOT use this class, use `RawTableBufferWriter` instead.
 */
class SWORDFISH_API RawTupleBufferWriter {
    template <typename T>
    struct remove_cv_ref {
        typedef typename std::remove_cv<typename std::remove_reference<T>::type>::type type;
    };
    template <typename T>
    using remove_cv_ref_t = typename remove_cv_ref<T>::type;
public:
    RawTupleBufferWriter();
    RawTupleBufferWriter(const RawTupleBuffer &tuple);
    ~RawTupleBufferWriter();

    void reset(const RawTupleBuffer &tuple);

    RawTupleBufferWriter& writeBool(char value);
    RawTupleBufferWriter& writeChar(char value);
    RawTupleBufferWriter& writeShort(short value);
    RawTupleBufferWriter& writeInt(int value);
    RawTupleBufferWriter& writeLong(long long value);
    RawTupleBufferWriter& writeFloat(float value);
    RawTupleBufferWriter& writeDouble(double value);

    RawTupleBufferWriter& writeString(const std::string &value);
    RawTupleBufferWriter& writeString(const char *str);
    RawTupleBufferWriter& writeString(const char *str, int len);
    RawTupleBufferWriter& writeString(const StringView &value);
    template <typename T,
              typename U = typename std::enable_if<std::is_arithmetic<remove_cv_ref_t<T>>::value>::type>
    RawTupleBufferWriter& writeString(T value) {
        throw RuntimeException("Can not convert arithmetic type to string");
    }

    RawTupleBufferWriter& writeDate(int value);
    RawTupleBufferWriter& writeMonth(int value);
    RawTupleBufferWriter& writeTime(int value);
    RawTupleBufferWriter& writeMinute(int value);
    RawTupleBufferWriter& writeSecond(int value);
    RawTupleBufferWriter& writeDatetime(int value);
    RawTupleBufferWriter& writeTimestamp(long long value);
    RawTupleBufferWriter& writeNanoTime(long long value);
    RawTupleBufferWriter& writeNanoTimestamp(long long value);
    RawTupleBufferWriter& writeDateHour(int value);

    RawTupleBufferWriter& writeBool(ConstantSP value);
    RawTupleBufferWriter& writeChar(ConstantSP value);
    RawTupleBufferWriter& writeShort(ConstantSP value);
    RawTupleBufferWriter& writeInt(ConstantSP value);
    RawTupleBufferWriter& writeLong(ConstantSP value);
    RawTupleBufferWriter& writeFloat(ConstantSP value);
    RawTupleBufferWriter& writeDouble(ConstantSP value);
    RawTupleBufferWriter& writeString(ConstantSP value);

    RawTupleBufferWriter& writeDate(ConstantSP value);
    RawTupleBufferWriter& writeMonth(ConstantSP value);
    RawTupleBufferWriter& writeTime(ConstantSP value);
    RawTupleBufferWriter& writeMinute(ConstantSP value);
    RawTupleBufferWriter& writeSecond(ConstantSP value);
    RawTupleBufferWriter& writeDatetime(ConstantSP value);
    RawTupleBufferWriter& writeTimestamp(ConstantSP value);
    RawTupleBufferWriter& writeNanoTime(ConstantSP value);
    RawTupleBufferWriter& writeNanoTimestamp(ConstantSP value);
    RawTupleBufferWriter& writeDateHour(ConstantSP value);

private:
    template <typename T>
    void write(T value);

    char *buffer_ = nullptr;
    int cursor_ = 0;
};
/// @endcond

class SWORDFISH_API RawTableBufferWriter {
public:
    RawTableBufferWriter(const RawTableBuffer &table,
                         const vector<ColumnDesc> &schema);

    ~RawTableBufferWriter();

    void reset(const RawTableBuffer &table);
    void reset(const RawTableBuffer &table, const vector<ColumnDesc> &schema);

    /**
     * Usage:
     * @code {.cpp}
     *   RawTableBufferWriter writer(table, schema);
     *
     *   writer.append(1, false, 1.0, "a")
     *         .append(2,  true, 2.0, "b")
     *         .append(3, false, 3.0, "c");
     * @endcode
     */
    template<typename ...Args>
    RawTableBufferWriter& append(Args... args) {
        if (row_ >= table_.rows) {
            throw RuntimeException("Table buffer is full");
        }

        if (schema_.size() != sizeof...(args)) {
            throw RuntimeException("Schema not match: expect " +
                    std::to_string(schema_.size()) + " columns, but given " +
                    std::to_string(sizeof...(args)) + " columns");
        }

        {
            RawTupleBufferWriter writer(table_.tuples[row_]);
            int i = 0;
            std::initializer_list<int> {
                (write(writer, schema_[i].getType(), args), ++i)...
            };
        }

        row_ += 1;
        return *this;
    }

private:
    template <typename T>
    void write(RawTupleBufferWriter &writer, DATA_TYPE type, const T &data);

    void write(RawTupleBufferWriter &writer, DATA_TYPE type,
               const string &data);

    void write(RawTupleBufferWriter &writer, DATA_TYPE type, const char *data);

    RawTableBuffer table_;
    vector<ColumnDesc> schema_;
    int row_ = 0;
};

template <typename T>
void RawTableBufferWriter::write(RawTupleBufferWriter &writer, DATA_TYPE type,
                                 const T &data) {
    switch (type) {
        case DT_BOOL: writer.writeBool(data); break;
        case DT_CHAR: writer.writeChar(data); break;
        case DT_SHORT: writer.writeShort(data); break;
        case DT_INT: writer.writeInt(data); break;
        case DT_LONG: writer.writeLong(data); break;
        case DT_FLOAT: writer.writeFloat(data); break;
        case DT_DOUBLE: writer.writeDouble(data); break;
        case DT_DATE: writer.writeDate(data); break;
        case DT_MONTH: writer.writeMonth(data); break;
        case DT_TIME: writer.writeTime(data); break;
        case DT_MINUTE: writer.writeMinute(data); break;
        case DT_SECOND: writer.writeSecond(data); break;
        case DT_DATETIME: writer.writeDatetime(data); break;
        case DT_TIMESTAMP: writer.writeTimestamp(data); break;
        case DT_NANOTIME: writer.writeNanoTime(data); break;
        case DT_NANOTIMESTAMP: writer.writeNanoTimestamp(data); break;
        case DT_DATEHOUR: writer.writeDateHour(data); break;
        case DT_STRING: writer.writeString(data); break;
        default:
            throw RuntimeException("Not support data type: " +
                    Util::getDataTypeString(type));
    }
}

/// @cond
class SWORDFISH_API RawTupleBufferGenerator {
public:
    RawTupleBufferGenerator(size_t bytes_per_row);

    ~RawTupleBufferGenerator();

    RawTupleBufferGenerator(const RawTupleBufferGenerator&) = delete;
    RawTupleBufferGenerator& operator=(const RawTupleBufferGenerator&) = delete;

    const RawTupleBuffer& get() const;

private:
    RawTupleBuffer tuple_;
    char *buffer_ = nullptr;
};
/// @endcond

class SWORDFISH_API RawTableBufferGenerator {
public:
    RawTableBufferGenerator(int rows, size_t bytes_per_row);

    ~RawTableBufferGenerator();

    RawTableBufferGenerator(const RawTableBufferGenerator&) = delete;
    RawTableBufferGenerator& operator=(const RawTableBufferGenerator&) = delete;

    const RawTableBuffer& get() const;

private:
    RawTableBuffer table_;
    RawTupleBuffer *tuples_ = nullptr;
};

//==============================================================================
// RawTupleBuffer / RawTableBuffer
//==============================================================================

struct IndexDescription {
    /** Index name. */
    string name;
    /** Whether is primary index. */
    bool primary;
    /** Whether is unique index. */
    bool unique;
    /** Columns use as index key. */
    vector<ColumnDesc> key;
};

struct TableDescription {
    /** Table name. */
    string name;
    /** Columns description. */
    vector<ColumnDesc> columns;
    /** Indexes of this table. */
    vector<IndexDescription> indexes;
};

//==============================================================================
// Util
//==============================================================================

class SWORDFISH_API StringView {
public:
    StringView() = default;

    StringView(const char *str, int size) : str_(str), size_(size) {
    }

    const char *str() const { return str_; }

    int size() const { return size_; }

    std::string toString() const {
        return std::string(str_, size_);
    }

    bool empty() const { return size_ == 0; }

private:
    const char *str_ = "";
    int size_ = 0;
};

/**
 * @brief Compute how many bytes is enough to store the data type.
 * @param max_bytes_for_varlen Max length of varlen type.
 * @return Bytes.
 */
SWORDFISH_API size_t computeBytes(DATA_TYPE type, size_t max_bytes_for_varlen);
SWORDFISH_API size_t computeBytes(const std::vector<DATA_TYPE> &types,
                    size_t max_bytes_for_varlen);
SWORDFISH_API size_t computeBytes(const std::vector<ColumnDesc> &descs,
                                  size_t max_bytes_for_varlen);

} // namespace oltp


} // namespace ddb

#endif  // DOLPHINDB_SWORDFISH_H_
