SessionImpl.java

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

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
*/

package com.mysql.clusterj.core;

import com.mysql.clusterj.ClusterJException;
import com.mysql.clusterj.ClusterJFatalInternalException;
import com.mysql.clusterj.ClusterJUserException;
import com.mysql.clusterj.DynamicObject;
import com.mysql.clusterj.LockMode;
import com.mysql.clusterj.Query;
import com.mysql.clusterj.Transaction;

import com.mysql.clusterj.core.spi.DomainTypeHandler;
import com.mysql.clusterj.core.spi.ValueHandler;

import com.mysql.clusterj.core.query.QueryDomainTypeImpl;
import com.mysql.clusterj.core.query.QueryBuilderImpl;
import com.mysql.clusterj.core.query.QueryImpl;

import com.mysql.clusterj.core.spi.SessionSPI;

import com.mysql.clusterj.core.store.ClusterTransaction;
import com.mysql.clusterj.core.store.Db;
import com.mysql.clusterj.core.store.Dictionary;
import com.mysql.clusterj.core.store.Index;
import com.mysql.clusterj.core.store.IndexOperation;
import com.mysql.clusterj.core.store.IndexScanOperation;
import com.mysql.clusterj.core.store.Operation;
import com.mysql.clusterj.core.store.PartitionKey;
import com.mysql.clusterj.core.store.ResultData;
import com.mysql.clusterj.core.store.ScanOperation;
import com.mysql.clusterj.core.store.Table;

import com.mysql.clusterj.core.util.I18NHelper;
import com.mysql.clusterj.core.util.Logger;
import com.mysql.clusterj.core.util.LoggerFactoryService;

import com.mysql.clusterj.query.QueryBuilder;
import com.mysql.clusterj.query.QueryDefinition;
import com.mysql.clusterj.query.QueryDomainType;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class SessionImpl implements SessionSPI, CacheManager, StoreManager {

    static final I18NHelper local = I18NHelper.getInstance(SessionImpl.class);

    static final Logger logger = LoggerFactoryService.getFactory().getInstance(SessionImpl.class);

    protected SessionFactoryImpl factory;

    protected Db db;

    protected Dictionary dictionary;

    protected TransactionImpl transactionImpl;

    protected PartitionKey partitionKey = null;

    protected boolean rollbackOnly = false;

    protected ClusterTransaction clusterTransaction;

    protected String joinTransactionId = null;

    protected Map properties;

    protected final int RESULT_READY = 0;
    protected final int SCAN_FINISHED = 1;
    protected final int CACHE_EMPTY = 2;

    protected List<StateManager> changeList = new ArrayList<StateManager>();

    protected List<Runnable> postExecuteOperations = new ArrayList<Runnable>();

    protected TransactionState transactionState;

    private ClusterJException transactionException;

    protected int nestedAutoTransactionCounter = 0;

    // TODO get this from properties
    protected int numberOfRetries = 5;

    private LockMode lockmode = LockMode.READ_COMMITTED;

    private Runnable postExecuteCallbackHandler = new Runnable() {
        public void run() {
            for (Runnable postExecuteCallback: postExecuteOperations) {
                postExecuteCallback.run();
            }
            postExecuteOperations.clear();
        }
    };

    SessionImpl(SessionFactoryImpl factory, Map properties, 
            Db db, Dictionary dictionary) {
        this.factory = factory;
        this.db = db;
        this.dictionary = dictionary;
        this.properties = properties;
        transactionImpl = new TransactionImpl(this);
        transactionState = transactionStateNotActive;
    }

    public <T> Query<T> createQuery(QueryDefinition<T> qd) {
        if (!(qd instanceof QueryDomainTypeImpl)) {
            throw new ClusterJUserException(
                    local.message("ERR_Exception_On_Method", "createQuery"));
        }
        return new QueryImpl<T>(this, (QueryDomainTypeImpl<T>)qd);
    }

    public <T> T find(Class<T> cls, Object key) {
        DomainTypeHandler<T> domainTypeHandler = getDomainTypeHandler(cls);
        T instance = (T) factory.newInstance(cls, dictionary);
        ValueHandler keyHandler = domainTypeHandler.createKeyValueHandler(key);
        ValueHandler instanceHandler = domainTypeHandler.getValueHandler(instance);
        // initialize from the database using the key
        return (T) initializeFromDatabase(
                domainTypeHandler, instance, instanceHandler, keyHandler);
    }

    public <T> T initializeFromDatabase(DomainTypeHandler<T> domainTypeHandler,
            T instance,
            ValueHandler instanceHandler, ValueHandler keyHandler) {
        startAutoTransaction();
        try {
            ResultData rs = selectUnique(domainTypeHandler, keyHandler, null);
            if (rs.next()) {
                // we have a result; initialize the instance
                if (instanceHandler == null) {
                    if (logger.isDetailEnabled()) logger.detail("Creating instanceHandler for class " + domainTypeHandler.getName() + " table: " + domainTypeHandler.getTableName() + keyHandler.pkToString(domainTypeHandler));
                    // we need both a new instance and its handler
                    instance = domainTypeHandler.newInstance();
                    instanceHandler = domainTypeHandler.getValueHandler(instance);
                } else if (instance == null) {
                if (logger.isDetailEnabled()) logger.detail("Creating instance for class " + domainTypeHandler.getName() + " table: " + domainTypeHandler.getTableName() + keyHandler.pkToString(domainTypeHandler));
                    // we have a handler but no instance
                    instance = domainTypeHandler.getInstance(instanceHandler);
                }
                // found the instance in the datastore
                instanceHandler.found(Boolean.TRUE);
                // put the results into the instance
                domainTypeHandler.objectSetValues(rs, instanceHandler);
                // set the cache manager to track updates
                domainTypeHandler.objectSetCacheManager(this, instanceHandler);
                // reset modified bits in instance
                domainTypeHandler.objectResetModified(instanceHandler);
            } else {
                if (logger.isDetailEnabled()) logger.detail("No instance found in database for class " + domainTypeHandler.getName() + " table: " + domainTypeHandler.getTableName() + keyHandler.pkToString(domainTypeHandler));
                // no instance found in database
                if (instanceHandler != null) {
                    // mark the handler as not found
                    instanceHandler.found(Boolean.FALSE);
                }
                endAutoTransaction();
                return null;
            }
        } catch (ClusterJException ex) {
            failAutoTransaction();
            throw ex;
        }
        endAutoTransaction();
        return instance;
    }

    private void setPartitionKey(DomainTypeHandler<?> domainTypeHandler,
            ValueHandler keyHandler) {
        if (!isEnlisted()) {
            // there is still time to set the partition key
            PartitionKey partitionKey = 
                domainTypeHandler.createPartitionKey(keyHandler);
            clusterTransaction.setPartitionKey(partitionKey);
        }
    }

    public <T> T newInstance(Class<T> cls) {
        return factory.newInstance(cls, dictionary);
    }

    public <T> T newInstance(Class<T> cls, Object key) {
        DomainTypeHandler<T> domainTypeHandler = getDomainTypeHandler(cls);
        T instance = factory.newInstance(cls, dictionary);
        domainTypeHandler.objectSetKeys(key, instance);
        return instance;
    }

    public <T> T load(final T object) {
        if (object == null) {
            return null;
        }
        if (Iterable.class.isAssignableFrom(object.getClass())) {
            Iterable<?> instances = (Iterable<?>)object;
            for (Object instance:instances) {
                load(instance);
            }
            return object;
        }
        if (object.getClass().isArray()) {
            Object[] instances = (Object[])object;
            for (Object instance:instances) {
                load(instance);
            }
            return object;
        }
        // a transaction must already be active (autocommit is not supported)
        assertActive();
        final DomainTypeHandler<?> domainTypeHandler = getDomainTypeHandler(object);
        final ValueHandler instanceHandler = domainTypeHandler.getValueHandler(object);
        setPartitionKey(domainTypeHandler, instanceHandler);
        Table storeTable = domainTypeHandler.getStoreTable();
        // perform a primary key operation
        final Operation op = clusterTransaction.getSelectOperation(storeTable);
        // set the keys into the operation
        domainTypeHandler.operationSetKeys(instanceHandler, op);
        // set the expected columns into the operation
        domainTypeHandler.operationGetValues(op);
        final ResultData rs = op.resultData(false);
        final SessionImpl cacheManager = this;
        // defer execution of the key operation until the next find, flush, or query
        Runnable postExecuteOperation = new Runnable() {
            public void run() {
                if (rs.next()) {
                    // found row in database
                    instanceHandler.found(Boolean.TRUE);
                   // put the results into the instance
                    domainTypeHandler.objectSetValues(rs, instanceHandler);
                    // set the cache manager to track updates
                    domainTypeHandler.objectSetCacheManager(cacheManager, instanceHandler);
                    // reset modified bits in instance
                    domainTypeHandler.objectResetModified(instanceHandler);
                } else {
                    // mark instance as not found
                    instanceHandler.found(Boolean.FALSE);
                }
                
            }
        };
        postExecuteOperations.add(postExecuteOperation);
        return object;
    }

    public Boolean found(Object instance) {
        if (instance == null) {
            return null;
        }
        if (instance instanceof DynamicObject) {
            return ((DynamicObject)instance).found();
        }
        // make sure the instance is a persistent type
        getDomainTypeHandler(instance);
        return true;
    }
    
    public <T> T makePersistent(T object) {
        if (object == null) {
            return null;
        }
        if (Iterable.class.isAssignableFrom(object.getClass())) {
            startAutoTransaction();
            Iterable<?> instances = (Iterable<?>)object;
            for (Object instance:instances) {
                makePersistent(instance);
            }
            endAutoTransaction();
            return object;
        }
        if (object.getClass().isArray()) {
            startAutoTransaction();
            Object[] instances = (Object[])object;
            for (Object instance:instances) {
                makePersistent(instance);
            }
            endAutoTransaction();
            return object;
        }
        DomainTypeHandler<T> domainTypeHandler = getDomainTypeHandler(object);
        ValueHandler valueHandler = domainTypeHandler.getValueHandler(object);
        insert(domainTypeHandler, valueHandler);
        return object;
    }

    public Operation insert(
            DomainTypeHandler<?> domainTypeHandler, ValueHandler valueHandler) {
        startAutoTransaction();
        setPartitionKey(domainTypeHandler, valueHandler);
        Operation op = null;
        Table storeTable = null;
        try {
            storeTable = domainTypeHandler.getStoreTable();
            op = clusterTransaction.getInsertOperation(storeTable);
            // set all values in the operation, keys first
            domainTypeHandler.operationSetKeys(valueHandler, op);
            domainTypeHandler.operationSetModifiedNonPKValues(valueHandler, op);
            // reset modified bits in instance
            domainTypeHandler.objectResetModified(valueHandler);
        } catch (ClusterJUserException cjuex) {
            failAutoTransaction();
            throw cjuex;
        } catch (ClusterJException cjex) {
            failAutoTransaction();
            logger.error(local.message("ERR_Insert", storeTable.getName()));
            throw new ClusterJException(
                    local.message("ERR_Insert", storeTable.getName()), cjex);
        } catch (RuntimeException rtex) {
            failAutoTransaction();
            logger.error(local.message("ERR_Insert", storeTable.getName()));
            throw new ClusterJException(
                    local.message("ERR_Insert", storeTable.getName()), rtex);
        }
        endAutoTransaction();
        return op;
    }

    public Iterable makePersistentAll(Iterable instances) {
        startAutoTransaction();
        List<Object> result = new ArrayList<Object>();
        for (Object instance:instances) {
            result.add(makePersistent(instance));
            }
        endAutoTransaction();
        return result;
    }

    public <T> void deletePersistent(Class<T> cls, Object key) {
        DomainTypeHandler<T> domainTypeHandler = getDomainTypeHandler(cls);
        ValueHandler keyValueHandler = domainTypeHandler.createKeyValueHandler(key);
        delete(domainTypeHandler, keyValueHandler);
    }

    public void deletePersistent(Object object) {
        if (object == null) {
            return;
        }
        DomainTypeHandler domainTypeHandler = getDomainTypeHandler(object);
        ValueHandler valueHandler = domainTypeHandler.getValueHandler(object);
        delete(domainTypeHandler, valueHandler);
    }

    public Operation delete(DomainTypeHandler domainTypeHandler, ValueHandler valueHandler) {
        startAutoTransaction();
        Table storeTable = domainTypeHandler.getStoreTable();
        setPartitionKey(domainTypeHandler, valueHandler);
        Operation op = null;
        try {
            op = clusterTransaction.getDeleteOperation(storeTable);
            domainTypeHandler.operationSetKeys(valueHandler, op);
        } catch (ClusterJException ex) {
            failAutoTransaction();
            throw new ClusterJException(
                    local.message("ERR_Delete", storeTable.getName()), ex);
        }
        endAutoTransaction();
        return op;
    }

    public void deletePersistentAll(Iterable objects) {
        startAutoTransaction();
        for (Iterator it = objects.iterator(); it.hasNext();) {
            deletePersistent(it.next());
        }
        endAutoTransaction();
    }

    public <T> int deletePersistentAll(Class<T> cls) {
        DomainTypeHandler<T> domainTypeHandler = getDomainTypeHandler(cls);
        return deletePersistentAll(domainTypeHandler);
    }

    public int deletePersistentAll(DomainTypeHandler<?> domainTypeHandler) {
        startAutoTransaction();
        Table storeTable = domainTypeHandler.getStoreTable();
        String tableName = storeTable.getName();
        ScanOperation op = null;
        int count = 0;
        try {
            op = clusterTransaction.getTableScanOperationLockModeExclusiveScanFlagKeyInfo(storeTable);
            count = deletePersistentAll(op, true);
        } catch (ClusterJException ex) {
            failAutoTransaction();
            // TODO add table name to the error message
            throw new ClusterJException(
                    local.message("ERR_Delete_All", tableName), ex);
        }
        endAutoTransaction();
        return count;
    }

    public int deletePersistentAll(ScanOperation op, boolean abort) {
        int cacheCount = 0;
        int count = 0;
        boolean done = false;
        boolean fetch = true;
        // cannot use early autocommit optimization here
        clusterTransaction.setAutocommit(false);
        // execute the operation
        clusterTransaction.executeNoCommit(true, true);
        while (!done ) {
            int result = op.nextResult(fetch);
            switch (result) {
                case RESULT_READY:
                    op.deleteCurrentTuple();
                    ++count;
                    ++cacheCount;
                    fetch = false;
                    break;
                case SCAN_FINISHED:
                    done = true;
                    if (cacheCount != 0) {
                        clusterTransaction.executeNoCommit(abort, true);
                    }
                    op.close();
                    break;
                case CACHE_EMPTY:
                    clusterTransaction.executeNoCommit(abort, true);
                    cacheCount = 0;
                    fetch = true;
                    break;
                default: 
                    throw new ClusterJException(
                            local.message("ERR_Next_Result_Illegal", result));
            }
        }
        return count;
    }

    public ResultData selectUnique(DomainTypeHandler domainTypeHandler,
            ValueHandler keyHandler, BitSet fields) {
        assertActive();
        setPartitionKey(domainTypeHandler, keyHandler);
        Table storeTable = domainTypeHandler.getStoreTable();
        // perform a single select by key operation
        Operation op = clusterTransaction.getSelectOperation(storeTable);
        // set the keys into the operation
        domainTypeHandler.operationSetKeys(keyHandler, op);
        // set the expected columns into the operation
        domainTypeHandler.operationGetValues(op);
        // execute the select and get results
        ResultData rs = op.resultData();
        return rs;
    }

    public void updatePersistent(Object object) {
        if (object == null) {
            return;
        }
        DomainTypeHandler domainTypeHandler = getDomainTypeHandler(object);
        if (logger.isDetailEnabled()) logger.detail("UpdatePersistent on object " + object);
        ValueHandler valueHandler = domainTypeHandler.getValueHandler(object);
        update(domainTypeHandler, valueHandler);
    }

    public Operation update(DomainTypeHandler domainTypeHandler, ValueHandler valueHandler) {
        startAutoTransaction();
        setPartitionKey(domainTypeHandler, valueHandler);
        Table storeTable = null;
        Operation op = null;
        try {
            storeTable = domainTypeHandler.getStoreTable();
            op = clusterTransaction.getUpdateOperation(storeTable);
            domainTypeHandler.operationSetKeys(valueHandler, op);
            domainTypeHandler.operationSetModifiedNonPKValues(valueHandler, op);
            if (logger.isDetailEnabled()) logger.detail("Updated object " +
                    valueHandler);
        } catch (ClusterJException ex) {
            failAutoTransaction();
            throw new ClusterJException(
                    local.message("ERR_Update", storeTable.getName()) ,ex);
        }
        endAutoTransaction();
        return op;
    }

    public void updatePersistentAll(Iterable objects) {
        startAutoTransaction();
        for (Iterator it = objects.iterator(); it.hasNext();) {
            updatePersistent(it.next());
        }
        endAutoTransaction();
    }

    public <T> T savePersistent(T instance) {
        DomainTypeHandler domainTypeHandler = getDomainTypeHandler(instance);
        if (logger.isDetailEnabled()) logger.detail("UpdatePersistent on object " + instance);
        ValueHandler valueHandler = domainTypeHandler.getValueHandler(instance);
        startAutoTransaction();
        setPartitionKey(domainTypeHandler, valueHandler);
        Table storeTable = null;
        try {
            storeTable = domainTypeHandler.getStoreTable();
            Operation op = null;
            op = clusterTransaction.getWriteOperation(storeTable);
            domainTypeHandler.operationSetKeys(valueHandler, op);
            domainTypeHandler.operationSetModifiedNonPKValues(valueHandler, op);
            if (logger.isDetailEnabled()) logger.detail("Wrote object " +
                    valueHandler);
        } catch (ClusterJException ex) {
            failAutoTransaction();
            throw new ClusterJException(
                    local.message("ERR_Write", storeTable.getName()) ,ex);
        }
        endAutoTransaction();
        return instance;
    }

    public Iterable savePersistentAll(Iterable instances) {
        List<Object> result = new ArrayList<Object>();
        startAutoTransaction();
        for (Iterator it = instances.iterator(); it.hasNext();) {
            result.add(savePersistent(it.next()));
        }
        endAutoTransaction();
        return result;
    }

    public Transaction currentTransaction() {
        return transactionImpl;
    }

    public void close() {
        if (clusterTransaction != null) {
            clusterTransaction.close();
            clusterTransaction = null;
        }
        if (db != null) {
            db.close();
            db = null;
        }
    }

    public boolean isClosed() {
        return db==null;
    }

    protected void assertNotClosed() {
        if (isClosed()) {
            throw new ClusterJUserException(
                    local.message("ERR_Session_Closed"));
        }
    }

    public void begin() {
        if (logger.isDebugEnabled()) logger.debug("begin transaction.");
        transactionState = transactionState.begin();
        handleTransactionException();
    }

    protected void internalBegin() {
        try {
            clusterTransaction = db.startTransaction(joinTransactionId);
            clusterTransaction.setLockMode(lockmode);
            // if a transaction has already begun, tell the cluster transaction about the key
            if (partitionKey != null) {
                clusterTransaction.setPartitionKey(partitionKey);
            }
            // register our post-execute callback
            clusterTransaction.postExecuteCallback(postExecuteCallbackHandler);
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Ndb_Start"), ex);
        }
    }

    public void commit() {
        if (logger.isDebugEnabled()) logger.debug("commit transaction.");
        transactionState = transactionState.commit();
        handleTransactionException();
    }

    protected void internalCommit() {
        if (rollbackOnly) {
            try {
                internalRollback();
                throw new ClusterJException(
                        local.message("ERR_Transaction_Rollback_Only"));
            } catch (ClusterJException ex) {
                throw new ClusterJException(
                        local.message("ERR_Transaction_Rollback_Only"), ex);
            }
        }
        try {
            clusterTransaction.executeCommit(true, true);
        } finally {
            // always close the transaction
            clusterTransaction.close();
            clusterTransaction = null;
            partitionKey = null;
        }
    }

    public void rollback() {
        if (logger.isDebugEnabled()) logger.debug("roll back transaction.");
        transactionState = transactionState.rollback();
        handleTransactionException();
    }

    protected void internalRollback() {
        try {
                clusterTransaction.executeRollback();
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Transaction_Execute", "rollback"), ex);
        } finally {
            if (clusterTransaction != null) {
                clusterTransaction.close();
            }
            clusterTransaction = null;
            partitionKey = null;
        }
    }

    public void startAutoTransaction() {
        if (logger.isDebugEnabled()) logger.debug("start AutoTransaction");
        transactionState = transactionState.start();
        handleTransactionException();
    }

    public void endAutoTransaction() {
        if (logger.isDebugEnabled()) logger.debug("end AutoTransaction");
        transactionState = transactionState.end();
        handleTransactionException();
    }

    public void failAutoTransaction() {
        if (logger.isDebugEnabled()) logger.debug("fail AutoTransaction");
        transactionState = transactionState.fail();
    }

    protected void handleTransactionException() {
        if (transactionException == null) {
            return;
        } else {
            ClusterJException ex = transactionException;
            transactionException = null;
            throw ex;
        }
    }

    public void setRollbackOnly() {
        rollbackOnly = true;
    }

    public boolean getRollbackOnly() {
        return rollbackOnly;
    }

    protected interface TransactionState {
        boolean isActive();

        TransactionState begin();
        TransactionState commit();
        TransactionState rollback();

        TransactionState start();
        TransactionState end();
        TransactionState fail();
    }

    protected TransactionState transactionStateNotActive =
            new TransactionState() {

        public boolean isActive() {
            return false;
        }

        public TransactionState begin() {
            try {
                internalBegin();
                return transactionStateActive;
            } catch (ClusterJException ex) {
                transactionException = ex;
                return transactionStateNotActive;
            }
        }

        public TransactionState commit() {
            transactionException = new ClusterJUserException(
                    local.message("ERR_Transaction_Must_Be_Active_For_Method",
                    "commit"));
            return transactionStateNotActive;
        }

        public TransactionState rollback() {
            transactionException = new ClusterJUserException(
                    local.message("ERR_Transaction_Must_Be_Active_For_Method",
                    "rollback"));
            return transactionStateNotActive;
        }

        public TransactionState start() {
            try {
                internalBegin();
                clusterTransaction.setAutocommit(true);
                nestedAutoTransactionCounter = 1;
                return transactionStateAutocommit;
            } catch (ClusterJException ex) {
                transactionException = ex;
                return transactionStateNotActive;
            }
        }

        public TransactionState end() {
            throw new ClusterJFatalInternalException(
                    local.message("ERR_Transaction_Auto_Start", "end"));
        }

        public TransactionState fail() {
            throw new ClusterJFatalInternalException(
                    local.message("ERR_Transaction_Auto_Start", "end"));
        }

    };

    protected TransactionState transactionStateActive =
            new TransactionState() {

        public boolean isActive() {
            return true;
        }

        public TransactionState begin() {
            transactionException = new ClusterJUserException(
                    local.message("ERR_Transaction_Must_Not_Be_Active_For_Method",
                    "begin"));
            return transactionStateActive;
        }

        public TransactionState commit() {
            try {
                // flush unwritten changes
                flush();
                internalCommit();
            } catch (ClusterJException ex) {
                transactionException = ex;
            }
            return transactionStateNotActive;
        }

        public TransactionState rollback() {
            try {
                internalRollback();
                return transactionStateNotActive;
            } catch (ClusterJException ex) {
                transactionException = ex;
                return transactionStateNotActive;
            }
        }

        public TransactionState start() {
            // nothing to do
            return transactionStateActive;
        }

        public TransactionState end() {
            // nothing to do
            return transactionStateActive;
        }

        public TransactionState fail() {
            // nothing to do
            return transactionStateActive;
        }

    };

    protected TransactionState transactionStateAutocommit =
            new TransactionState() {

        public boolean isActive() {
            return true;
        }

        public TransactionState begin() {
            throw new ClusterJFatalInternalException(
                    local.message("ERR_Transaction_Auto_End", "begin"));
        }

        public TransactionState commit() {
            throw new ClusterJFatalInternalException(
                    local.message("ERR_Transaction_Auto_End", "commit"));
        }

        public TransactionState rollback() {
            throw new ClusterJFatalInternalException(
                    local.message("ERR_Transaction_Auto_End", "rollback"));
        }

        public TransactionState start() {
            // nested start; increment counter
            nestedAutoTransactionCounter++;
            return transactionStateAutocommit;
        }

        public TransactionState end() {
            if (--nestedAutoTransactionCounter > 0) {
                return transactionStateAutocommit;
            } else if (nestedAutoTransactionCounter == 0) {
                try {
                    internalCommit();
                } catch (ClusterJException ex) {
                    transactionException = ex;
                }
                return transactionStateNotActive;
            } else {
            throw new ClusterJFatalInternalException(
                    local.message("ERR_Transaction_Auto_Start", "end"));
            }
        }

        public TransactionState fail() {
            try {
                nestedAutoTransactionCounter = 0;
                internalRollback();
                return transactionStateNotActive;
            } catch (ClusterJException ex) {
                // ignore failures caused by internal rollback
                return transactionStateNotActive;
            }
        }

    };

    protected synchronized <T> DomainTypeHandler<T> getDomainTypeHandler(T object) {
        DomainTypeHandler<T> domainTypeHandler =
                factory.getDomainTypeHandler(object, dictionary);
        return domainTypeHandler;
    }

    public synchronized <T> DomainTypeHandler<T> getDomainTypeHandler(Class<T> cls) {
        DomainTypeHandler<T> domainTypeHandler =
                factory.getDomainTypeHandler(cls, dictionary);
        return domainTypeHandler;
    }

    public Dictionary getDictionary() {
        return dictionary;
    }

    boolean isActive() {
        return transactionState.isActive();
    }

    public boolean isEnlisted() {
        return clusterTransaction==null?false:clusterTransaction.isEnlisted();
    }

    private void assertActive() {
        if (!transactionState.isActive()) {
            throw new ClusterJUserException(
                    local.message("ERR_Transaction_Must_Be_Active"));
        }
    }

    private void assertNotActive(String methodName) {
        if (transactionState.isActive()) {
            throw new ClusterJUserException(
                    local.message("ERR_Transaction_Must_Not_Be_Active_For_Method",
                    methodName));
        }
    }

    public Query createQuery(Class cls) {
        throw new UnsupportedOperationException(
                local.message("ERR_NotImplemented"));
    }

    public QueryBuilder getQueryBuilder() {
        return new QueryBuilderImpl(this);
    }

    public IndexScanOperation getIndexScanOperation(Index storeIndex, Table storeTable) {
        assertActive();
        try {
            IndexScanOperation result = clusterTransaction.getIndexScanOperation(storeIndex, storeTable);
            return result;
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Index_Scan", storeTable.getName(), storeIndex.getName()), ex);
        }
    }

    public IndexScanOperation getIndexScanOperationMultiRange(Index storeIndex, Table storeTable) {
        assertActive();
        try {
            IndexScanOperation result = clusterTransaction.getIndexScanOperationMultiRange(storeIndex, storeTable);
            return result;
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Index_Scan", storeTable.getName(), storeIndex.getName()), ex);
        }
    }

    public IndexScanOperation getIndexScanDeleteOperation(Index storeIndex, Table storeTable) {
        assertActive();
        try {
            IndexScanOperation result = clusterTransaction.getIndexScanOperationLockModeExclusiveScanFlagKeyInfo(storeIndex, storeTable);
            return result;
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Index_Scan", storeTable.getName(), storeIndex.getName()), ex);
        }
    }

    public ScanOperation getTableScanOperation(Table storeTable) {
        assertActive();
        try {
            ScanOperation result = clusterTransaction.getTableScanOperation(storeTable);
            return result;
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Table_Scan", storeTable.getName()), ex);
        }
    }

   public ScanOperation getTableScanDeleteOperation(Table storeTable) {
       assertActive();
       try {
           ScanOperation result = clusterTransaction.getTableScanOperationLockModeExclusiveScanFlagKeyInfo(storeTable);
           return result;
       } catch (ClusterJException ex) {
           throw new ClusterJException(
                   local.message("ERR_Table_Scan", storeTable.getName()), ex);
       }
   }

    public IndexOperation getUniqueIndexOperation(Index storeIndex, Table storeTable) {
        assertActive();
        try {
            IndexOperation result = clusterTransaction.getUniqueIndexOperation(storeIndex, storeTable);
            return result;
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Unique_Index", storeTable.getName(), storeIndex.getName()), ex);
        }
    }

    public Operation getSelectOperation(Table storeTable) {
        assertActive();
        try {
            Operation result = clusterTransaction.getSelectOperation(storeTable);
            return result;
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Select", storeTable), ex);
        }
    }

    public Operation getDeleteOperation(Table storeTable) {
        assertActive();
        try {
            Operation result = clusterTransaction.getDeleteOperation(storeTable);
            return result;
        } catch (ClusterJException ex) {
            throw new ClusterJException(
                    local.message("ERR_Delete", storeTable), ex);
        }
    }

   public IndexOperation getUniqueIndexDeleteOperation(Index storeIndex, Table storeTable) {
       assertActive();
       try {
           IndexOperation result = clusterTransaction.getUniqueIndexDeleteOperation(storeIndex, storeTable);
           return result;
       } catch (ClusterJException ex) {
           throw new ClusterJException(
                   local.message("ERR_Unique_Index_Delete", storeTable.getName(), storeIndex.getName()), ex);
       }
   }

    public void flush() {
        if (logger.isDetailEnabled()) logger.detail("flush changes with changeList size: " + changeList.size());
        if (!changeList.isEmpty()) {
            for (StateManager sm: changeList) {
                sm.flush(this);
            }
            changeList.clear();
        }
        // now flush changes to the back end
        if (clusterTransaction != null) {
            executeNoCommit();
        }
    }

    public List getChangeList() {
        return Collections.unmodifiableList(changeList);
    }

    public void persist(Object instance) {
        makePersistent(instance);
    }

    public void remove(Object instance) {
        deletePersistent(instance);
    }

    public void markModified(StateManager instance) {
        changeList.add(instance);
    }

    public void setPartitionKey(Class<?> domainClass, Object key) {
        DomainTypeHandler<?> domainTypeHandler = getDomainTypeHandler(domainClass);
        String tableName = domainTypeHandler.getTableName();
        // if transaction is enlisted, throw a user exception
        if (isEnlisted()) {
            throw new ClusterJUserException(
                    local.message("ERR_Set_Partition_Key_After_Enlistment", tableName));
        }
        // if a partition key has already been set, throw a user exception
        if (this.partitionKey != null) {
            throw new ClusterJUserException(
                    local.message("ERR_Set_Partition_Key_Twice", tableName));
        }
        ValueHandler handler = domainTypeHandler.createKeyValueHandler(key);
        this.partitionKey= domainTypeHandler.createPartitionKey(handler);
        // if a transaction has already begun, tell the cluster transaction about the key
        if (clusterTransaction != null) {
            clusterTransaction.setPartitionKey(partitionKey);
        }
    }

    public void markModified(Object instance, String fieldName) {
        DomainTypeHandler<?> domainTypeHandler = getDomainTypeHandler(instance);
        ValueHandler handler = domainTypeHandler.getValueHandler(instance);
        domainTypeHandler.objectMarkModified(handler, fieldName);
    }

    public void executeNoCommit(boolean abort, boolean force) {
        if (clusterTransaction != null) {
            clusterTransaction.executeNoCommit(abort, force);
        }
    }

    public void executeNoCommit() {
        executeNoCommit(true, true);
    }

    public <T> QueryDomainType<T> createQueryDomainType(DomainTypeHandler<T> domainTypeHandler) {
        QueryBuilderImpl builder = (QueryBuilderImpl)getQueryBuilder();
        return builder.createQueryDefinition(domainTypeHandler);
    }

    public String getCoordinatedTransactionId() {
        return clusterTransaction.getCoordinatedTransactionId();
    }

    public void setCoordinatedTransactionId(String coordinatedTransactionId) {
        clusterTransaction.setCoordinatedTransactionId(coordinatedTransactionId);
    }

    public void setLockMode(LockMode lockmode) {
        this.lockmode = lockmode;
        if (clusterTransaction != null) {
            clusterTransaction.setLockMode(lockmode);
        }
    }

}