package com.thinkaurelius.titan.diskstorage.berkeleyje;

public class BerkeleyJEKeyValueStore implements OrderedKeyValueStore {

    private static final Logger log = LoggerFactory.getLogger(BerkeleyJEKeyValueStore.class);

    /// DatabaseEntry 是 BerkeleyDB 记录的key和data的类型。
    private static final StaticBuffer.Factory<DatabaseEntry> ENTRY_FACTORY = new StaticBuffer.Factory<DatabaseEntry>() {
        @Override
        public DatabaseEntry get(byte[] array, int offset, int limit) {
            return new DatabaseEntry(array,offset,limit-offset);
        }
    };


    private final Database db; /// BerkeleyDB实例引用
    private final String name; 

    /// 引用了manager，主要是为了在close()的时候移除自己：manager.removeDatabase(this)
    private final BerkeleyJEStoreManager manager;

    private boolean isOpen;

    public BerkeleyJEKeyValueStore(String n, Database data, BerkeleyJEStoreManager m) {
        db = data;
        name = n;
        manager = m;
        isOpen = true;
    }

    public DatabaseConfig getConfiguration() throws BackendException {
        try {
            return db.getConfig();
        } catch (DatabaseException e) {
            throw new PermanentBackendException(e);
        }
    }

    @Override
    public String getName() {
        return name;
    }

    private static final Transaction getTransaction(StoreTransaction txh) {
        Preconditions.checkArgument(txh!=null);
        return ((BerkeleyJETx) txh).getTransaction();
    }

    @Override
    public synchronized void close() throws BackendException {
        try {
            if(isOpen) db.close();
        } catch (DatabaseException e) {
            throw new PermanentBackendException(e);
        }
        if (isOpen) manager.removeDatabase(this);
        isOpen = false;
    }

    @Override
    public StaticBuffer get(StaticBuffer key, StoreTransaction txh) throws BackendException {
        Transaction tx = getTransaction(txh);
        try {
            DatabaseEntry dbkey = key.as(ENTRY_FACTORY);
            DatabaseEntry data = new DatabaseEntry();

            log.trace("db={}, op=get, tx={}", name, txh);

            OperationStatus status = db.get(tx, dbkey, data, getLockMode(txh));

            if (status == OperationStatus.SUCCESS) {
                return getBuffer(data);
            } else {
                return null;
            }
        } catch (DatabaseException e) {
            throw new PermanentBackendException(e);
        }
    }

    @Override
    public boolean containsKey(StaticBuffer key, StoreTransaction txh) throws BackendException {
        return get(key,txh)!=null;
    }

    @Override
    public void acquireLock(StaticBuffer key, StaticBuffer expectedValue, StoreTransaction txh) throws BackendException {
        if (getTransaction(txh) == null) {
            log.warn("Attempt to acquire lock with transactions disabled");
        } //else we need no locking
    }

    @Override
    public RecordIterator<KeyValueEntry> getSlice(KVQuery query, StoreTransaction txh) throws BackendException {
        log.trace("beginning db={}, op=getSlice, tx={}", name, txh);
        Transaction tx = getTransaction(txh);
        Cursor cursor = null;
        final StaticBuffer keyStart = query.getStart();
        final StaticBuffer keyEnd = query.getEnd();
        final KeySelector selector = query.getKeySelector();
        final List<KeyValueEntry> result = new ArrayList<KeyValueEntry>();
        try {
            DatabaseEntry foundKey = keyStart.as(ENTRY_FACTORY);
            DatabaseEntry foundData = new DatabaseEntry();

            cursor = db.openCursor(tx, null);
            /// getSearchKeyRange将游标定位到跟foundKey最接近的地方，再用Cursor.getNext就能得到游标下一个
            OperationStatus status = cursor.getSearchKeyRange(foundKey, foundData, getLockMode(txh));
            //Iterate until given condition is satisfied or end of records
            while (status == OperationStatus.SUCCESS) {
                StaticBuffer key = getBuffer(foundKey); /// getSearchKeyRange返回的key值

                if (key.compareTo(keyEnd) >= 0)
                    break;

                if (selector.include(key)) {
                    result.add(new KeyValueEntry(key, getBuffer(foundData)));
                }

                if (selector.reachedLimit())
                    break;

                /// 调用cursor.getNext得到下一个key，因为是key有序的，所以得到的就是一个key range。
                status = cursor.getNext(foundKey, foundData, getLockMode(txh));
            }
            log.trace("db={}, op=getSlice, tx={}, resultcount={}", name, txh, result.size());

            return new RecordIterator<KeyValueEntry>() {
                private final Iterator<KeyValueEntry> entries = result.iterator();

                @Override
                public boolean hasNext() {
                    return entries.hasNext();
                }

                @Override
                public KeyValueEntry next() {
                    return entries.next();
                }

                @Override
                public void close() {
                }

                @Override
                public void remove() {
                    throw new UnsupportedOperationException();
                }
            };
        } catch (Exception e) {
            throw new PermanentBackendException(e);
        } finally {
            try {
                if (cursor != null) cursor.close();
            } catch (Exception e) {
                throw new PermanentBackendException(e);
            }
        }
    }

    @Override
    public Map<KVQuery,RecordIterator<KeyValueEntry>> getSlices(List<KVQuery> queries, StoreTransaction txh) throws BackendException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void insert(StaticBuffer key, StaticBuffer value, StoreTransaction txh) throws BackendException {
        insert(key, value, txh, true);
    }

    public void insert(StaticBuffer key, StaticBuffer value, StoreTransaction txh, boolean allowOverwrite) throws BackendException {
        Transaction tx = getTransaction(txh);
        try {
            OperationStatus status;

            log.trace("db={}, op=insert, tx={}", name, txh);

            if (allowOverwrite)
                status = db.put(tx, key.as(ENTRY_FACTORY), value.as(ENTRY_FACTORY));
            else
                status = db.putNoOverwrite(tx, key.as(ENTRY_FACTORY), value.as(ENTRY_FACTORY));

            if (status != OperationStatus.SUCCESS) {
                if (status == OperationStatus.KEYEXIST) {
                    throw new PermanentBackendException("Key already exists on no-overwrite.");
                } else {
                    throw new PermanentBackendException("Could not write entity, return status: " + status);
                }
            }
        } catch (DatabaseException e) {
            throw new PermanentBackendException(e);
        }
    }


    @Override
    public void delete(StaticBuffer key, StoreTransaction txh) throws BackendException {
        log.trace("Deletion");
        Transaction tx = getTransaction(txh);
        try {
            log.trace("db={}, op=delete, tx={}", name, txh);
            OperationStatus status = db.delete(tx, key.as(ENTRY_FACTORY));
            if (status != OperationStatus.SUCCESS) {
                throw new PermanentBackendException("Could not remove: " + status);
            }
        } catch (DatabaseException e) {
            throw new PermanentBackendException(e);
        }
    }

    private static StaticBuffer getBuffer(DatabaseEntry entry) {
        return new StaticArrayBuffer(entry.getData(),entry.getOffset(),entry.getOffset()+entry.getSize());
    }

    private static LockMode getLockMode(StoreTransaction txh) {
        return ((BerkeleyJETx)txh).getLockMode();
    }
}

package com.thinkaurelius.titan.diskstorage.berkeleyje;

@PreInitializeConfigOptions
public class BerkeleyJEStoreManager extends LocalStoreManager implements OrderedKeyValueStoreManager {

    public static final ConfigNamespace BERKELEY_NS =
            new ConfigNamespace(GraphDatabaseConfiguration.STORAGE_NS, "berkeleyje", "BerkeleyDB JE configuration options");

    public static final ConfigOption<Integer> JVM_CACHE =
            new ConfigOption<Integer>(BERKELEY_NS,"cache-percentage",
            "Percentage of JVM heap reserved for BerkeleyJE's cache",
            ConfigOption.Type.MASKABLE, 65, ConfigOption.positiveInt());

    public static final ConfigOption<String> LOCK_MODE =
            new ConfigOption<>(BERKELEY_NS, "lock-mode",
            "The BDB record lock mode used for read operations",
            ConfigOption.Type.MASKABLE, String.class, LockMode.DEFAULT.toString(), disallowEmpty(String.class));

    public static final ConfigOption<String> ISOLATION_LEVEL =
            new ConfigOption<>(BERKELEY_NS, "isolation-level",
            "The isolation level used by transactions",
            ConfigOption.Type.MASKABLE,  String.class,
            IsolationLevel.REPEATABLE_READ.toString(), disallowEmpty(String.class));

    private final Map<String, BerkeleyJEKeyValueStore> stores;

    /// BerkeleyDB environment. Environments include support for some or all of caching, locking, logging and transactions.
    protected Environment environment;
    protected final StoreFeatures features;

    public BerkeleyJEStoreManager(Configuration configuration) throws BackendException {
        super(configuration);
        stores = new HashMap<String, BerkeleyJEKeyValueStore>();

        int cachePercentage = configuration.get(JVM_CACHE);
        initialize(cachePercentage);

        features = new StandardStoreFeatures.Builder()
                    .orderedScan(true)
                    .transactional(transactional)
                    .keyConsistent(GraphDatabaseConfiguration.buildGraphConfiguration())
                    .locking(true)
                    .keyOrdered(true)
                    .scanTxConfig(GraphDatabaseConfiguration.buildGraphConfiguration()
                            .set(ISOLATION_LEVEL, IsolationLevel.READ_UNCOMMITTED.toString()))
                    .supportsInterruption(false)
                    .build();

    }

    private void initialize(int cachePercent) throws BackendException {
        try {
            EnvironmentConfig envConfig = new EnvironmentConfig();
            envConfig.setAllowCreate(true);
            envConfig.setTransactional(transactional);
            envConfig.setCachePercent(cachePercent);

            if (batchLoading) {
                envConfig.setConfigParam(EnvironmentConfig.ENV_RUN_CHECKPOINTER, "false");
                envConfig.setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "false");
            }

            //Open the environment
            environment = new Environment(directory, envConfig);


        } catch (DatabaseException e) {
            throw new PermanentBackendException("Error during BerkeleyJE initialization: ", e);
        }

    }

    @Override
    public StoreFeatures getFeatures() {
        return features;
    }

    @Override
    public List<KeyRange> getLocalKeyPartition() throws BackendException {
        throw new UnsupportedOperationException();
    }

    @Override
    public BerkeleyJETx beginTransaction(final BaseTransactionConfig txCfg) throws BackendException {
        try {
            Transaction tx = null;

            Configuration effectiveCfg =
                    new MergedConfiguration(txCfg.getCustomOptions(), getStorageConfig());

            if (transactional) {
                TransactionConfig txnConfig = new TransactionConfig();
                ConfigOption.getEnumValue(effectiveCfg.get(ISOLATION_LEVEL),IsolationLevel.class).configure(txnConfig);
                tx = environment.beginTransaction(null, txnConfig);
            }
            BerkeleyJETx btx = new BerkeleyJETx(tx, ConfigOption.getEnumValue(effectiveCfg.get(LOCK_MODE),LockMode.class), txCfg);

            if (log.isTraceEnabled()) {
                log.trace("Berkeley tx created", new TransactionBegin(btx.toString()));
            }

            return btx;
        } catch (DatabaseException e) {
            throw new PermanentBackendException("Could not start BerkeleyJE transaction", e);
        }
    }

    @Override
    public BerkeleyJEKeyValueStore openDatabase(String name) throws BackendException {
        Preconditions.checkNotNull(name);
        if (stores.containsKey(name)) {
            BerkeleyJEKeyValueStore store = stores.get(name);
            return store;
        }
        try {
            DatabaseConfig dbConfig = new DatabaseConfig();
            dbConfig.setReadOnly(false);
            dbConfig.setAllowCreate(true);
            dbConfig.setTransactional(transactional);

            dbConfig.setKeyPrefixing(true);

            if (batchLoading) {
                dbConfig.setDeferredWrite(true);
            }

            Database db = environment.openDatabase(null, name, dbConfig);

            log.debug("Opened database {}", name, new Throwable());

            BerkeleyJEKeyValueStore store = new BerkeleyJEKeyValueStore(name, db, this);
            stores.put(name, store);
            return store;
        } catch (DatabaseException e) {
            throw new PermanentBackendException("Could not open BerkeleyJE data store", e);
        }
    }

    @Override
    public void mutateMany(Map<String, KVMutation> mutations, StoreTransaction txh) throws BackendException {
        for (Map.Entry<String,KVMutation> muts : mutations.entrySet()) {
            BerkeleyJEKeyValueStore store = openDatabase(muts.getKey());
            KVMutation mut = muts.getValue();

            if (!mut.hasAdditions() && !mut.hasDeletions()) {
                log.debug("Empty mutation set for {}, doing nothing", muts.getKey());
            } else {
                log.debug("Mutating {}", muts.getKey());
            }

            if (mut.hasAdditions()) {
                for (KeyValueEntry entry : mut.getAdditions()) {
                    store.insert(entry.getKey(),entry.getValue(),txh);
                    log.trace("Insertion on {}: {}", muts.getKey(), entry);
                }
            }
            if (mut.hasDeletions()) {
                for (StaticBuffer del : mut.getDeletions()) {
                    store.delete(del,txh);
                    log.trace("Deletion on {}: {}", muts.getKey(), del);
                }
            }
        }
    }

    void removeDatabase(BerkeleyJEKeyValueStore db) {
        if (!stores.containsKey(db.getName())) {
            throw new IllegalArgumentException("Tried to remove an unkown database from the storage manager");
        }
        String name = db.getName();
        stores.remove(name);
        log.debug("Removed database {}", name);
    }


    @Override
    public void close() throws BackendException {
        if (environment != null) {
            if (!stores.isEmpty())
                throw new IllegalStateException("Cannot shutdown manager since some databases are still open");
            try {
                // TODO this looks like a race condition
                //Wait just a little bit before closing so that independent transaction threads can clean up.
                Thread.sleep(30);
            } catch (InterruptedException e) {
                //Ignore
            }
            try {
                environment.close();
            } catch (DatabaseException e) {
                throw new PermanentBackendException("Could not close BerkeleyJE database", e);
            }
        }

    }

    @Override
    public void clearStorage() throws BackendException {
        if (!stores.isEmpty())
            throw new IllegalStateException("Cannot delete store, since database is open: " + stores.keySet().toString());

        Transaction tx = null;
        for (String db : environment.getDatabaseNames()) {
            environment.removeDatabase(tx, db);
            log.debug("Removed database {} (clearStorage)", db);
        }
        close();
        IOUtils.deleteFromDirectory(directory);
    }

    @Override
    public String getName() {
        return getClass().getSimpleName() + ":" + directory.toString();
    }


    public static enum IsolationLevel {
        READ_UNCOMMITTED {
            @Override
            void configure(TransactionConfig cfg) {
                cfg.setReadUncommitted(true);
            }
        }, READ_COMMITTED {
            @Override
            void configure(TransactionConfig cfg) {
                cfg.setReadCommitted(true);

            }
        }, REPEATABLE_READ {
            @Override
            void configure(TransactionConfig cfg) {
                // This is the default and has no setter
            }
        }, SERIALIZABLE {
            @Override
            void configure(TransactionConfig cfg) {
                cfg.setSerializableIsolation(true);
            }
        };

        abstract void configure(TransactionConfig cfg);
    };

    private static class TransactionBegin extends Exception {
        private static final long serialVersionUID = 1L;

        private TransactionBegin(String msg) {
            super(msg);
        }
    }
}

private final Map<String, BerkeleyJEKeyValueStore> stores;

/// BerkeleyDB environment. Environments include support for some or all of caching, locking, logging and transactions.
protected Environment environment;
protected final StoreFeatures features;

package com.thinkaurelius.titan.diskstorage.berkeleyje;

import com.google.common.base.Preconditions;
import com.sleepycat.je.Cursor;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.Transaction;
import com.thinkaurelius.titan.diskstorage.BackendException;
import com.thinkaurelius.titan.diskstorage.PermanentBackendException;
import com.thinkaurelius.titan.diskstorage.BaseTransactionConfig;
import com.thinkaurelius.titan.diskstorage.common.AbstractStoreTransaction;

public class BerkeleyJETx extends AbstractStoreTransaction {

    private static final Logger log = LoggerFactory.getLogger(BerkeleyJETx.class);

    private volatile Transaction tx;
    private final List<Cursor> openCursors = new ArrayList<Cursor>();
    private final LockMode lm;

    public BerkeleyJETx(Transaction t, LockMode lockMode, BaseTransactionConfig config) {
        super(config);
        tx = t;
        lm = lockMode;
        // tx may be null
        Preconditions.checkNotNull(lm);
    }

    public Transaction getTransaction() {
        return tx;
    }

    void registerCursor(Cursor cursor) {
        Preconditions.checkArgument(cursor != null);
        synchronized (openCursors) {
            //TODO: attempt to remove closed cursors if there are too many
            openCursors.add(cursor);
        }
    }

    private void closeOpenIterators() throws BackendException {
        for (Cursor cursor : openCursors) {
            cursor.close();
        }
    }

    LockMode getLockMode() {
        return lm;
    }

    @Override
    public synchronized void rollback() throws BackendException {
        super.rollback();
        if (tx == null) return;
        if (log.isTraceEnabled())
            log.trace("{} rolled back", this.toString(), new TransactionClose(this.toString()));
        try {
            closeOpenIterators();
            tx.abort();
            tx = null;
        } catch (DatabaseException e) {
            throw new PermanentBackendException(e);
        }
    }

    @Override
    public synchronized void commit() throws BackendException {
        super.commit();
        if (tx == null) return;
        if (log.isTraceEnabled())
            log.trace("{} committed", this.toString(), new TransactionClose(this.toString()));

        try {
            closeOpenIterators();
            tx.commit();
            tx = null;
        } catch (DatabaseException e) {
            throw new PermanentBackendException(e);
        }
    }

    @Override
    public String toString() {
        return getClass().getSimpleName() + (null == tx ? "nulltx" : tx.toString());
    }

    private static class TransactionClose extends Exception {
        private static final long serialVersionUID = 1L;

        private TransactionClose(String msg) {
            super(msg);
        }
    }
}

/**
    * Interface to a data store that has a BigTable like representation of its data. In other words, the data store is comprised of a set of rows
    * each of which is uniquely identified by a key. Each row is composed of a column-value pairs. For a given key, a subset of the column-value
    * pairs that fall within a column interval can be quickly retrieved.
    * <p/>
    * This interface provides methods for retrieving and mutating the data.
    * <p/>
    * In this generic representation keys, columns and values are represented as ByteBuffers.
    * <p/>
    */
public interface KeyColumnValueStore {

    public static final List<Entry> NO_ADDITIONS = ImmutableList.of();
    public static final List<StaticBuffer> NO_DELETIONS = ImmutableList.of();

    /**
        * Retrieves the list of entries (i.e. column-value pairs) for a specified query.
        *
        * @param query Query to get results for
        * @param txh   Transaction
        * @return List of entries up to a maximum of "limit" entries
        * @throws com.thinkaurelius.titan.diskstorage.BackendException when columnEnd < columnStart
        * @see KeySliceQuery
        */
    public EntryList getSlice(KeySliceQuery query, StoreTransaction txh) throws BackendException;

    /**
        * Retrieves the list of entries (i.e. column-value pairs) as specified by the given {@link SliceQuery} for all
        * of the given keys together.
        *
        * @param keys  List of keys
        * @param query Slicequery specifying matching entries
        * @param txh   Transaction
        * @return The result of the query for each of the given keys as a map from the key to the list of result entries.
        * @throws com.thinkaurelius.titan.diskstorage.BackendException
        */
    public Map<StaticBuffer,EntryList> getSlice(List<StaticBuffer> keys, SliceQuery query, StoreTransaction txh) throws BackendException;

    /**
        * Verifies acquisition of locks {@code txh} from previous calls to
        * {@link #acquireLock(StaticBuffer, StaticBuffer, StaticBuffer, StoreTransaction)}
        * , then writes supplied {@code additions} and/or {@code deletions} to
        * {@code key} in the underlying data store. Deletions are applied strictly
        * before additions. In other words, if both an addition and deletion are
        * supplied for the same column, then the column will first be deleted and
        * then the supplied Entry for the column will be added.
        * <p/>
        * Implementations which don't support locking should skip the initial lock
        * verification step but otherwise behave as described above.
        *
        * @param key       the key under which the columns in {@code additions} and
        *                  {@code deletions} will be written
        * @param additions the list of Entry instances representing column-value pairs to
        *                  create under {@code key}, or null to add no column-value pairs
        * @param deletions the list of columns to delete from {@code key}, or null to
        *                  delete no columns
        * @param txh       the transaction to use
        * @throws com.thinkaurelius.titan.diskstorage.locking.PermanentLockingException if locking is supported by the implementation and at least
        *                          one lock acquisition attempted by
        *                          {@link #acquireLock(StaticBuffer, StaticBuffer, StaticBuffer, StoreTransaction)}
        *                          has failed
        */
    public void mutate(StaticBuffer key, List<Entry> additions, List<StaticBuffer> deletions, StoreTransaction txh) throws BackendException;

    /**
        * Attempts to claim a lock on the value at the specified {@code key} and
        * {@code column} pair. These locks are discretionary.
        * <p/>
        * 这个接口是可选的。
        */
    public void acquireLock(StaticBuffer key, StaticBuffer column, StaticBuffer expectedValue, StoreTransaction txh) throws BackendException;

    /**
        * Returns a {@link KeyIterator} over all keys that fall within the key-range specified by the given query and have one or more columns matching the column-range.
        * Calling {@link KeyIterator#getEntries()} returns the list of all entries that match the column-range specified by the given query.
        * <p/>
        * This method is only supported by stores which keep keys in byte-order.
        */
    public KeyIterator getKeys(KeyRangeQuery query, StoreTransaction txh) throws BackendException;

    /**
        * Returns a {@link KeyIterator} over all keys in the store that have one or more columns matching the column-range. Calling {@link KeyIterator#getEntries()}
        * returns the list of all entries that match the column-range specified by the given query.
        * <p/>
        * This method is only supported by stores which do not keep keys in byte-order.
        */
    public KeyIterator getKeys(SliceQuery query, StoreTransaction txh) throws BackendException;
    // like current getKeys if column-slice is such that it queries for vertex state property

    public String getName();

    public void close() throws BackendException;

}

public class HBaseStoreManager extends DistributedStoreManager implements KeyColumnValueStoreManager {

    /// ...配置项相关

    private static final BiMap<String, String> SHORT_CF_NAME_MAP =
    ImmutableBiMap.<String, String>builder()
            .put(INDEXSTORE_NAME, "g")
            .put(INDEXSTORE_NAME + LOCK_STORE_SUFFIX, "h")
            .put(ID_STORE_NAME, "i")
            .put(EDGESTORE_NAME, "e")
            .put(EDGESTORE_NAME + LOCK_STORE_SUFFIX, "f")
            .put(SYSTEM_PROPERTIES_STORE_NAME, "s")
            .put(SYSTEM_PROPERTIES_STORE_NAME + LOCK_STORE_SUFFIX, "t")
            .put(SYSTEM_MGMT_LOG_NAME, "m")
            .put(SYSTEM_TX_LOG_NAME, "l")
            .build();

    // Immutable instance fields
    private final String tableName;
    private final String compression;
    private final int regionCount;
    private final int regionsPerServer;
    private final ConnectionMask cnx;
    private final org.apache.hadoop.conf.Configuration hconf;
    private final boolean shortCfNames;
    private final boolean skipSchemaCheck;
    private final String compatClass;
    private final HBaseCompat compat;

    private static final ConcurrentHashMap<HBaseStoreManager, Throwable> openManagers =
            new ConcurrentHashMap<HBaseStoreManager, Throwable>();

    // Mutable instance state
    private final ConcurrentMap<String, HBaseKeyColumnValueStore> openStores;

    public HBaseStoreManager(com.thinkaurelius.titan.diskstorage.configuration.Configuration config) throws BackendException {
        /// .. 初始化
    }

    @Override
    public Deployment getDeployment() {
        /// ...
    }

    @Override
    public String toString() {
        return "hbase[" + tableName + "@" + super.toString() + "]";
    }

    public void dumpOpenManagers() {
        /// ...
    }

    @Override
    public void close() {
        ///...
    }

    @Override
    public StoreFeatures getFeatures() {
        /// ...
    }

    @Override
    public void mutateMany(Map<String, Map<StaticBuffer, KCVMutation>> mutations, StoreTransaction txh) throws BackendException {
        final MaskedTimestamp commitTime = new MaskedTimestamp(txh);
        // In case of an addition and deletion with identical timestamps, the
        // deletion tombstone wins.
        // http://hbase.apache.org/book/versions.html#d244e4250
        Map<StaticBuffer, Pair<Put, Delete>> commandsPerKey =
                convertToCommands(
                        mutations,
                        commitTime.getAdditionTime(times),
                        commitTime.getDeletionTime(times));

        List<Row> batch = new ArrayList<Row>(commandsPerKey.size()); // actual batch operation

        // convert sorted commands into representation required for 'batch' operation
        for (Pair<Put, Delete> commands : commandsPerKey.values()) {
            if (commands.getFirst() != null)
                batch.add(commands.getFirst());

            if (commands.getSecond() != null)
                batch.add(commands.getSecond());
        }

        try {
            TableMask table = null;

            try {
                table = cnx.getTable(tableName);
                table.batch(batch, new Object[batch.size()]);
            } finally {
                IOUtils.closeQuietly(table);
            }
        } catch (IOException e) {
            throw new TemporaryBackendException(e);
        } catch (InterruptedException e) {
            throw new TemporaryBackendException(e);
        }

        sleepAfterWrite(txh, commitTime);
    }

    @Override
    public KeyColumnValueStore openDatabase(String longName, StoreMetaData.Container metaData) throws BackendException {

        HBaseKeyColumnValueStore store = openStores.get(longName);

        if (store == null) {
            final String cfName = shortCfNames ? shortenCfName(longName) : longName;

            HBaseKeyColumnValueStore newStore = new HBaseKeyColumnValueStore(this, cnx, tableName, cfName, longName);

            store = openStores.putIfAbsent(longName, newStore); // nothing bad happens if we loose to other thread

            if (store == null) {
                if (!skipSchemaCheck) {
                    int cfTTLInSeconds = -1;
                    if (metaData.contains(StoreMetaData.TTL)) {
                        cfTTLInSeconds = metaData.get(StoreMetaData.TTL);
                    }
                    ensureColumnFamilyExists(tableName, cfName, cfTTLInSeconds);
                }

                store = newStore;
            }
        }

        return store;
    }

    @Override
    public StoreTransaction beginTransaction(final BaseTransactionConfig config) throws BackendException {
        return new HBaseTransaction(config);
    }

    @Override
    public String getName() {
        return tableName;
    }

    /**
        * Deletes the specified table with all its columns.
        * ATTENTION: Invoking this method will delete the table if it exists and therefore causes data loss.
        */
    @Override
    public void clearStorage() throws BackendException {
        /// ...
    }

    @Override
    public List<KeyRange> getLocalKeyPartition() throws BackendException {

        /// ...
    }


    /**
        * This method generates the second argument to
        * {@link HBaseAdmin#createTable(HTableDescriptor, byte[], byte[], int)}.
        * <p/>
        * From the {@code createTable} javadoc:
        * "The start key specified will become the end key of the first region of
        * the table, and the end key specified will become the start key of the
        * last region of the table (the first region has a null start key and
        * the last region has a null end key)"
        * <p/>
        * To summarize, the {@code createTable} argument called "startKey" is
        * actually the end key of the first region.
        */
    private byte[] getStartKey(int regionCount) {
        ByteBuffer regionWidth = ByteBuffer.allocate(4);
        regionWidth.putInt((int)(((1L << 32) - 1L) / regionCount)).flip();
        return StaticArrayBuffer.of(regionWidth).getBytes(0, 4);
    }

    /**
        * Companion to {@link #getStartKey(int)}. See its javadoc for details.
        */
    private byte[] getEndKey(int regionCount) {
        ByteBuffer regionWidth = ByteBuffer.allocate(4);
        regionWidth.putInt((int)(((1L << 32) - 1L) / regionCount * (regionCount - 1))).flip();
        return StaticArrayBuffer.of(regionWidth).getBytes(0, 4);
    }

    /**
        * Convert Titan internal Mutation representation into HBase native commands.
        *
        * @param mutations    Mutations to convert into HBase commands.
        * @param putTimestamp The timestamp to use for Put commands.
        * @param delTimestamp The timestamp to use for Delete commands.
        * @return Commands sorted by key converted from Titan internal representation.
        * @throws com.thinkaurelius.titan.diskstorage.PermanentBackendException
        */
    private Map<StaticBuffer, Pair<Put, Delete>> convertToCommands(Map<String, Map<StaticBuffer, KCVMutation>> mutations,
                                                                    final long putTimestamp,
                                                                    final long delTimestamp) throws PermanentBackendException {
        Map<StaticBuffer, Pair<Put, Delete>> commandsPerKey = new HashMap<StaticBuffer, Pair<Put, Delete>>();

        for (Map.Entry<String, Map<StaticBuffer, KCVMutation>> entry : mutations.entrySet()) {

            String cfString = getCfNameForStoreName(entry.getKey());
            byte[] cfName = cfString.getBytes();

            for (Map.Entry<StaticBuffer, KCVMutation> m : entry.getValue().entrySet()) {
                byte[] key = m.getKey().as(StaticBuffer.ARRAY_FACTORY);
                KCVMutation mutation = m.getValue();

                Pair<Put, Delete> commands = commandsPerKey.get(m.getKey());

                if (commands == null) {
                    commands = new Pair<Put, Delete>();
                    commandsPerKey.put(m.getKey(), commands);
                }

                if (mutation.hasDeletions()) {
                    if (commands.getSecond() == null) {
                        Delete d = new Delete(key);
                        compat.setTimestamp(d, delTimestamp);
                        commands.setSecond(d);
                    }

                    for (StaticBuffer b : mutation.getDeletions()) {
                        commands.getSecond().deleteColumns(cfName, b.as(StaticBuffer.ARRAY_FACTORY), delTimestamp);
                    }
                }

                if (mutation.hasAdditions()) {
                    if (commands.getFirst() == null) {
                        Put p = new Put(key, putTimestamp);
                        commands.setFirst(p);
                    }

                    for (Entry e : mutation.getAdditions()) {
                        commands.getFirst().add(cfName,
                                e.getColumnAs(StaticBuffer.ARRAY_FACTORY),
                                putTimestamp,
                                e.getValueAs(StaticBuffer.ARRAY_FACTORY));
                    }
                }
            }
        }

        return commandsPerKey;
    }

}

private static final BiMap<String, String> SHORT_CF_NAME_MAP =
        ImmutableBiMap.<String, String>builder()
                .put(INDEXSTORE_NAME, "g")
                .put(INDEXSTORE_NAME + LOCK_STORE_SUFFIX, "h")
                .put(ID_STORE_NAME, "i")
                .put(EDGESTORE_NAME, "e")
                .put(EDGESTORE_NAME + LOCK_STORE_SUFFIX, "f")
                .put(SYSTEM_PROPERTIES_STORE_NAME, "s")
                .put(SYSTEM_PROPERTIES_STORE_NAME + LOCK_STORE_SUFFIX, "t")
                .put(SYSTEM_MGMT_LOG_NAME, "m")
                .put(SYSTEM_TX_LOG_NAME, "l")
                .build();

@Override
public KeyColumnValueStore openDatabase(String longName, StoreMetaData.Container metaData) throws BackendException {

    HBaseKeyColumnValueStore store = openStores.get(longName);

    if (store == null) {
        final String cfName = shortCfNames ? shortenCfName(longName) : longName;

        HBaseKeyColumnValueStore newStore = new HBaseKeyColumnValueStore(this, cnx, tableName, cfName, longName);

        store = openStores.putIfAbsent(longName, newStore); // nothing bad happens if we loose to other thread

        if (store == null) {
            if (!skipSchemaCheck) {
                int cfTTLInSeconds = -1;
                if (metaData.contains(StoreMetaData.TTL)) {
                    cfTTLInSeconds = metaData.get(StoreMetaData.TTL);
                }
                ensureColumnFamilyExists(tableName, cfName, cfTTLInSeconds);
            }

            store = newStore;
        }
    }

    return store;
}

private void ensureColumnFamilyExists(String tableName, String columnFamily, int ttlInSeconds) throws BackendException {
    AdminMask adm = null;
    try {
        adm = getAdminInterface();
        HTableDescriptor desc = ensureTableExists(tableName, columnFamily, ttlInSeconds);

        Preconditions.checkNotNull(desc);

        HColumnDescriptor cf = desc.getFamily(columnFamily.getBytes());

        // Create our column family, if necessary
        if (cf == null) {
            try {
                if (!adm.isTableDisabled(tableName)) {
                    adm.disableTable(tableName);
                }
            } catch (){
                ...
            }

            try {
                HColumnDescriptor cdesc = new HColumnDescriptor(columnFamily);

                setCFOptions(cdesc, ttlInSeconds);

                adm.addColumn(tableName, cdesc);

                try {
                    logger.debug("Added HBase ColumnFamily {}, waiting for 1 sec. to propogate.", columnFamily);
                    Thread.sleep(1000L);
                } catch (InterruptedException ie) {
                    throw new TemporaryBackendException(ie);
                }

                adm.enableTable(tableName);
            } catch (...){
                ...
            }
        }
    } finally {
        IOUtils.closeQuietly(adm);
    }
}

HBaseKeyColumnValueStore newStore = new HBaseKeyColumnValueStore(this, cnx, tableName, cfName, longName);

@Override
public StoreTransaction beginTransaction(final BaseTransactionConfig config) throws BackendException {
    return new HBaseTransaction(config);
}

public KeyColumnValueStore openDatabase(String longName, StoreMetaData.Container metaData) throws BackendException;

/**
    * Convert Titan internal Mutation representation into HBase native commands.
    *
    * @param mutations    Mutations to convert into HBase commands.
    * @param putTimestamp The timestamp to use for Put commands.
    * @param delTimestamp The timestamp to use for Delete commands.
    * @return Commands sorted by key converted from Titan internal representation.
    * @throws com.thinkaurelius.titan.diskstorage.PermanentBackendException
    */
private Map<StaticBuffer, Pair<Put, Delete>> convertToCommands(Map<String, Map<StaticBuffer, KCVMutation>> mutations,
                                                                final long putTimestamp,
                                                                final long delTimestamp) throws PermanentBackendException;

/**
    * This class overrides and adds nothing compared with ExpectedValueCheckingTransaction; 
    * however, it creates a transaction type specific to HBase, which lets us check for user errors 
    * like passing a Cassandra transaction into a HBase method.
    */
public class HBaseTransaction extends AbstractStoreTransaction {

    public HBaseTransaction(final BaseTransactionConfig config) {
        super(config);
    }
}

public class StandardTitanGraph extends TitanBlueprintsGraph {

    static {
        TraversalStrategies graphStrategies = TraversalStrategies.GlobalCache.getStrategies(Graph.class).clone()
                .addStrategies(AdjacentVertexFilterOptimizerStrategy.instance(), TitanLocalQueryOptimizerStrategy.instance(), TitanGraphStepStrategy.instance());

        //Register with cache
        TraversalStrategies.GlobalCache.registerStrategies(StandardTitanGraph.class, graphStrategies);
        TraversalStrategies.GlobalCache.registerStrategies(StandardTitanTx.class, graphStrategies);
    }

    private GraphDatabaseConfiguration config;
    private Backend backend;
    private IDManager idManager;
    private VertexIDAssigner idAssigner;
    private TimestampProvider times;

    //Serializers
    protected IndexSerializer indexSerializer;
    protected EdgeSerializer edgeSerializer;
    protected Serializer serializer;

    //Caches
    public SliceQuery vertexExistenceQuery;
    private RelationQueryCache queryCache;
    private SchemaCache schemaCache;

    //Log
    private ManagementLogger mgmtLogger;

    //Shutdown hook
    private volatile ShutdownThread shutdownHook;

    private volatile boolean isOpen = true;
    private AtomicLong txCounter;

    private Set<StandardTitanTx> openTransactions;

    ...

}

public abstract class TitanBlueprintsGraph implements TitanGraph {

    // ########## TRANSACTION HANDLING ###########################

    final GraphTransaction tinkerpopTxContainer = new GraphTransaction();

    private ThreadLocal<TitanBlueprintsTransaction> txs = new ThreadLocal<TitanBlueprintsTransaction>() {

    private TitanBlueprintsTransaction getAutoStartTx() {
        if (txs == null) throw new IllegalStateException("Graph has been closed");
        tinkerpopTxContainer.readWrite();

        TitanBlueprintsTransaction tx = txs.get();
        Preconditions.checkState(tx!=null,"Invalid read-write behavior configured: " +
                "Should either open transaction or throw exception.");
        return tx;
    }

    // ########## TRANSACTIONAL FORWARDING ###########################

    @Override
    public TitanVertex addVertex(Object... keyValues) {
        return getAutoStartTx().addVertex(keyValues);
    }

    @Override
    public Iterator<Edge> edges(Object... edgeIds) {
        return getAutoStartTx().edges(edgeIds);
    }

    @Override
    public TitanGraphQuery<? extends TitanGraphQuery> query() {
        return getAutoStartTx().query();
    }

    @Override
    public TitanIndexQuery indexQuery(String indexName, String query) {
        return getAutoStartTx().indexQuery(indexName,query);
    }

    ...

    //Schema

    @Override
    public PropertyKeyMaker makePropertyKey(String name) {
        return getAutoStartTx().makePropertyKey(name);
    }

    @Override
    public EdgeLabelMaker makeEdgeLabel(String name) {
        return getAutoStartTx().makeEdgeLabel(name);
    }

    @Override
    public VertexLabelMaker makeVertexLabel(String name) {
        return getAutoStartTx().makeVertexLabel(name);
    }
    
    ...

}

public TitanVertex addVertex(Object... keyValues) {
    ...

    final TitanVertex vertex = addVertex(null,label);
    com.thinkaurelius.titan.graphdb.util.ElementHelper.attachProperties(vertex, keyValues);
    return vertex;
}

@Override
public TitanVertex addVertex(Long vertexId, VertexLabel label) {
    
    ... // 各种检查

    StandardVertex vertex = new StandardVertex(this, IDManager.getTemporaryVertexID(IDManager.VertexIDType.NormalVertex, temporaryIds.nextID()), ElementLifeCycle.New);
    if (vertexId != null) {
        vertex.setId(vertexId);
    } else if (config.hasAssignIDsImmediately() || label.isPartitioned()) {
        graph.assignID(vertex,label);
    }
    addProperty(vertex, BaseKey.VertexExists, Boolean.TRUE);
    if (label!=BaseVertexLabel.DEFAULT_VERTEXLABEL) { //Add label
        Preconditions.checkArgument(label instanceof VertexLabelVertex);
        addEdge(vertex, (VertexLabelVertex) label, BaseLabel.VertexLabelEdge);
    }
    vertexCache.add(vertex, vertex.longId());
    return vertex;

}

private void connectRelation(InternalRelation r) {
    for (int i = 0; i < r.getLen(); i++) {
        boolean success = r.getVertex(i).addRelation(r);
        if (!success) throw new AssertionError("Could not connect relation: " + r);
    }
    addedRelations.add(r);
    for (int pos = 0; pos < r.getLen(); pos++) vertexCache.add(r.getVertex(pos), r.getVertex(pos).longId());
    if (TypeUtil.hasSimpleInternalVertexKeyIndex(r)) newVertexIndexEntries.add((TitanVertexProperty) r);
}

/**
    * This is essentially an adjusted copy&paste from TinkerPop's ElementHelper class.
    * The reason for copying it is so that we can determine the cardinality of a property key based on
    * Titan's schema which is tied to this particular transaction and not the graph.
    *
    * @param vertex
    * @param propertyKeyValues
    */
public static void attachProperties(final TitanVertex vertex, final Object... propertyKeyValues) {
    for (int i = 0; i < propertyKeyValues.length; i = i + 2) {
        if (!propertyKeyValues[i].equals(T.id) && !propertyKeyValues[i].equals(T.label))
            vertex.property((String) propertyKeyValues[i], propertyKeyValues[i + 1]);
    }
}

public<V> TitanVertexProperty<V> property(final String key, final V value, final Object... keyValues) {
    TitanVertexProperty<V> p = tx().addProperty(it(), tx().getOrCreatePropertyKey(key), value);
    ElementHelper.attachProperties(p,keyValues);
    return p;
}

public TitanVertexProperty addProperty(VertexProperty.Cardinality cardi, TitanVertex vertex, PropertyKey key, Object value) {
    if (key.cardinality().convert()!=cardi && cardi!=VertexProperty.Cardinality.single)
            throw new SchemaViolationException(String.format("Key is defined for %s cardinality which conflicts with specified: %s",key.cardinality(),cardi));
    verifyWriteAccess(vertex);
    Preconditions.checkArgument(!(key instanceof ImplicitKey),"Cannot create a property of implicit type: %s",key.name());
    vertex = ((InternalVertex) vertex).it();
    Preconditions.checkNotNull(key);
    final Object normalizedValue = verifyAttribute(key, value);
    Cardinality cardinality = key.cardinality();

    //Determine unique indexes
    List<IndexLockTuple> uniqueIndexTuples = new ArrayList<IndexLockTuple>();
    for (CompositeIndexType index : TypeUtil.getUniqueIndexes(key)) {
        IndexSerializer.IndexRecords matches = IndexSerializer.indexMatches(vertex, index, key, normalizedValue);
        for (Object[] match : matches.getRecordValues()) uniqueIndexTuples.add(new IndexLockTuple(index,match));
    }

    TransactionLock uniqueLock = getUniquenessLock(vertex, (InternalRelationType) key, normalizedValue);
    //Add locks for unique indexes
    for (IndexLockTuple lockTuple : uniqueIndexTuples) uniqueLock = new CombinerLock(uniqueLock,getLock(lockTuple),times);
    uniqueLock.lock(LOCK_TIMEOUT);
    try {
        //Delete properties if the cardinality is restricted
        if (cardi==VertexProperty.Cardinality.single || cardi== VertexProperty.Cardinality.set) {
            Consumer<TitanRelation> propertyRemover;
            if (cardi==VertexProperty.Cardinality.single)
                propertyRemover = p -> p.remove();
            else
                propertyRemover = p -> { if (((TitanVertexProperty)p).value().equals(normalizedValue)) p.remove(); };

            /* If we are simply overwriting a vertex property, then we don't have to explicitly remove it thereby saving a read operation
                However, this only applies if
                1) we don't lock on the property key or consistency checks are disabled and
                2) there are no indexes for this property key
                3) the cardinalities match (if we overwrite a set with single, we need to read all other values to delete)
            */
            ConsistencyModifier mod = ((InternalRelationType)key).getConsistencyModifier();
            boolean hasIndex = TypeUtil.hasAnyIndex(key);
            boolean cardiEqual = cardi==cardinality.convert();

            if ( (!config.hasVerifyUniqueness() || ((InternalRelationType)key).getConsistencyModifier()!=ConsistencyModifier.LOCK) &&
                    !TypeUtil.hasAnyIndex(key) && cardi==cardinality.convert()) {
                //Only delete in-memory so as to not trigger a read from the database which isn't necessary because we will overwrite blindly
                ((InternalVertex) vertex).getAddedRelations(p -> p.getType().equals(key)).forEach(propertyRemover);
            } else {
                ((InternalVertex) vertex).query().types(key).properties().forEach(propertyRemover);
            }
        }

        //Check index uniqueness
        if (config.hasVerifyUniqueness()) {
            //Check all unique indexes
            for (IndexLockTuple lockTuple : uniqueIndexTuples) {
                if (!Iterables.isEmpty(IndexHelper.getQueryResults(lockTuple.getIndex(), lockTuple.getAll(), this)))
                    throw new SchemaViolationException("Adding this property for key [%s] and value [%s] violates a uniqueness constraint [%s]", key.name(), normalizedValue, lockTuple.getIndex());
            }
        }
        StandardVertexProperty prop = new StandardVertexProperty(IDManager.getTemporaryRelationID(temporaryIds.nextID()), key, (InternalVertex) vertex, normalizedValue, ElementLifeCycle.New);
        if (config.hasAssignIDsImmediately()) graph.assignID(prop);
        connectRelation(prop);
        return prop;
    } finally {
        uniqueLock.unlock();
    }

}

StandardVertexProperty prop = new StandardVertexProperty(IDManager.getTemporaryRelationID(temporaryIds.nextID()), key, (InternalVertex) vertex, normalizedValue, ElementLifeCycle.New);
if (config.hasAssignIDsImmediately()) graph.assignID(prop);
connectRelation(prop);

@Override
public synchronized void commit() {
    
    ... // 一些检查

    try {
        if (hasModifications()) {
            graph.commit(addedRelations.getAll(), deletedRelations.values(), this);
        } else {
            txHandle.commit();
        }
        success = true;
    } catch (Exception e) {
        try {
            txHandle.rollback();
        } catch (BackendException e1) {
            throw new TitanException("Could not rollback after a failed commit", e);
        }
        throw new TitanException("Could not commit transaction due to exception during persistence", e);
    } finally {
        releaseTransaction();
        if (null != config.getGroupName() && !success) {
            MetricManager.INSTANCE.getCounter(config.getGroupName(), "tx", "commit.exceptions").inc();
        }
    }
}

public void commit(final Collection<InternalRelation> addedRelations,
                    final Collection<InternalRelation> deletedRelations, final StandardTitanTx tx) {
    if (addedRelations.isEmpty() && deletedRelations.isEmpty()) return;
    //1. Finalize transaction
    log.debug("Saving transaction. Added {}, removed {}", addedRelations.size(), deletedRelations.size());
    if (!tx.getConfiguration().hasCommitTime()) tx.getConfiguration().setCommitTime(times.getTime());
    final Instant txTimestamp = tx.getConfiguration().getCommitTime();
    final long transactionId = txCounter.incrementAndGet();

    //2. Assign TitanVertex IDs
    if (!tx.getConfiguration().hasAssignIDsImmediately())
        idAssigner.assignIDs(addedRelations);

    //3. Commit
    BackendTransaction mutator = tx.getTxHandle();
    final boolean acquireLocks = tx.getConfiguration().hasAcquireLocks();
    final boolean hasTxIsolation = backend.getStoreFeatures().hasTxIsolation();
    final boolean logTransaction = config.hasLogTransactions() && !tx.getConfiguration().hasEnabledBatchLoading();
    final KCVSLog txLog = logTransaction?backend.getSystemTxLog():null;
    final TransactionLogHeader txLogHeader = new TransactionLogHeader(transactionId,txTimestamp, times);
    ModificationSummary commitSummary;

    try {
        //3.1 Log transaction (write-ahead log) if enabled
        if (logTransaction) {
            //[FAILURE] Inability to log transaction fails the transaction by escalation since it's likely due to unavailability of primary
            //storage backend.
            txLog.add(txLogHeader.serializeModifications(serializer, LogTxStatus.PRECOMMIT, tx, addedRelations, deletedRelations),txLogHeader.getLogKey());
        }

        //3.2 Commit schema elements and their associated relations in a separate transaction if backend does not support
        //    transactional isolation
        boolean hasSchemaElements = !Iterables.isEmpty(Iterables.filter(deletedRelations,SCHEMA_FILTER))
                || !Iterables.isEmpty(Iterables.filter(addedRelations,SCHEMA_FILTER));
        Preconditions.checkArgument(!hasSchemaElements || (!tx.getConfiguration().hasEnabledBatchLoading() && acquireLocks),
                "Attempting to create schema elements in inconsistent state");

        if (hasSchemaElements && !hasTxIsolation) {
            /*
                * On storage without transactional isolation, create separate
                * backend transaction for schema aspects to make sure that
                * those are persisted prior to and independently of other
                * mutations in the tx. If the storage supports transactional
                * isolation, then don't create a separate tx.
                */
            final BackendTransaction schemaMutator = openBackendTransaction(tx);

            try {
                //[FAILURE] If the preparation throws an exception abort directly - nothing persisted since batch-loading cannot be enabled for schema elements
                commitSummary = prepareCommit(addedRelations,deletedRelations, SCHEMA_FILTER, schemaMutator, tx, acquireLocks);
                assert commitSummary.hasModifications && !commitSummary.has2iModifications;
            } catch (Throwable e) {
                //Roll back schema tx and escalate exception
                schemaMutator.rollback();
                throw e;
            }

            try {
                schemaMutator.commit();
            } catch (Throwable e) {
                //[FAILURE] Primary persistence failed => abort and escalate exception, nothing should have been persisted
                log.error("Could not commit transaction ["+transactionId+"] due to storage exception in system-commit",e);
                throw e;
            }
        }

        //[FAILURE] Exceptions during preparation here cause the entire transaction to fail on transactional systems
        //or just the non-system part on others. Nothing has been persisted unless batch-loading
        commitSummary = prepareCommit(addedRelations,deletedRelations, hasTxIsolation? NO_FILTER : NO_SCHEMA_FILTER, mutator, tx, acquireLocks);
        if (commitSummary.hasModifications) {
            String logTxIdentifier = tx.getConfiguration().getLogIdentifier();
            boolean hasSecondaryPersistence = logTxIdentifier!=null || commitSummary.has2iModifications;

            //1. Commit storage - failures lead to immediate abort

            //1a. Add success message to tx log which will be committed atomically with all transactional changes so that we can recover secondary failures
            //    This should not throw an exception since the mutations are just cached. If it does, it will be escalated since its critical
            if (logTransaction) {
                txLog.add(txLogHeader.serializePrimary(serializer,
                                    hasSecondaryPersistence?LogTxStatus.PRIMARY_SUCCESS:LogTxStatus.COMPLETE_SUCCESS),
                        txLogHeader.getLogKey(),mutator.getTxLogPersistor());
            }

            try {
                mutator.commitStorage();
            } catch (Throwable e) {
                //[FAILURE] If primary storage persistence fails abort directly (only schema could have been persisted)
                log.error("Could not commit transaction ["+transactionId+"] due to storage exception in commit",e);
                throw e;
            }

            if (hasSecondaryPersistence) {
                LogTxStatus status = LogTxStatus.SECONDARY_SUCCESS;
                Map<String,Throwable> indexFailures = ImmutableMap.of();
                boolean userlogSuccess = true;

                try {
                    //2. Commit indexes - [FAILURE] all exceptions are collected and logged but nothing is aborted
                    indexFailures = mutator.commitIndexes();
                    if (!indexFailures.isEmpty()) {
                        status = LogTxStatus.SECONDARY_FAILURE;
                        for (Map.Entry<String,Throwable> entry : indexFailures.entrySet()) {
                            log.error("Error while commiting index mutations for transaction ["+transactionId+"] on index: " +entry.getKey(),entry.getValue());
                        }
                    }
                    //3. Log transaction if configured - [FAILURE] is recorded but does not cause exception
                    if (logTxIdentifier!=null) {
                        try {
                            userlogSuccess = false;
                            final Log userLog = backend.getUserLog(logTxIdentifier);
                            Future<Message> env = userLog.add(txLogHeader.serializeModifications(serializer, LogTxStatus.USER_LOG, tx, addedRelations, deletedRelations));
                            if (env.isDone()) {
                                try {
                                    env.get();
                                } catch (ExecutionException ex) {
                                    throw ex.getCause();
                                }
                            }
                            userlogSuccess=true;
                        } catch (Throwable e) {
                            status = LogTxStatus.SECONDARY_FAILURE;
                            log.error("Could not user-log committed transaction ["+transactionId+"] to " + logTxIdentifier, e);
                        }
                    }
                } finally {
                    if (logTransaction) {
                        //[FAILURE] An exception here will be logged and not escalated; tx considered success and
                        // needs to be cleaned up later
                        try {
                            txLog.add(txLogHeader.serializeSecondary(serializer,status,indexFailures,userlogSuccess),txLogHeader.getLogKey());
                        } catch (Throwable e) {
                            log.error("Could not tx-log secondary persistence status on transaction ["+transactionId+"]",e);
                        }
                    }
                }
            } else {
                //This just closes the transaction since there are no modifications
                mutator.commitIndexes();
            }
        } else { //Just commit everything at once
            //[FAILURE] This case only happens when there are no non-system mutations in which case all changes
            //are already flushed. Hence, an exception here is unlikely and should abort
            mutator.commit();
        }
    } catch (Throwable e) {
        log.error("Could not commit transaction ["+transactionId+"] due to exception",e);
        try {
            //Clean up any left-over transaction handles
            mutator.rollback();
        } catch (Throwable e2) {
            log.error("Could not roll-back transaction ["+transactionId+"] after failure due to exception",e2);
        }
        if (e instanceof RuntimeException) throw (RuntimeException)e;
        else throw new TitanException("Unexpected exception",e);
    }
}

public ModificationSummary prepareCommit(final Collection<InternalRelation> addedRelations,
                                    final Collection<InternalRelation> deletedRelations,
                                    final Predicate<InternalRelation> filter,
                                    final BackendTransaction mutator, final StandardTitanTx tx,
                                    final boolean acquireLocks) throws BackendException {

    ListMultimap<Long, InternalRelation> mutations = ArrayListMultimap.create();
    ListMultimap<InternalVertex, InternalRelation> mutatedProperties = ArrayListMultimap.create();
    List<IndexSerializer.IndexUpdate> indexUpdates = Lists.newArrayList();
    //1) Collect deleted edges and their index updates and acquire edge locks
    for (InternalRelation del : Iterables.filter(deletedRelations,filter)) {
        Preconditions.checkArgument(del.isRemoved());
        for (int pos = 0; pos < del.getLen(); pos++) {
            InternalVertex vertex = del.getVertex(pos);
            if (pos == 0 || !del.isLoop()) {
                if (del.isProperty()) mutatedProperties.put(vertex,del);
                mutations.put(vertex.longId(), del);
            }
            if (acquireLock(del,pos,acquireLocks)) {
                Entry entry = edgeSerializer.writeRelation(del, pos, tx);
                mutator.acquireEdgeLock(idManager.getKey(vertex.longId()), entry);
            }
        }
        indexUpdates.addAll(indexSerializer.getIndexUpdates(del));
    }

    //2) Collect added edges and their index updates and acquire edge locks
    for (InternalRelation add : Iterables.filter(addedRelations,filter)) {
        Preconditions.checkArgument(add.isNew());

        for (int pos = 0; pos < add.getLen(); pos++) {
            InternalVertex vertex = add.getVertex(pos);
            if (pos == 0 || !add.isLoop()) {
                if (add.isProperty()) mutatedProperties.put(vertex,add);
                mutations.put(vertex.longId(), add);
            }
            if (!vertex.isNew() && acquireLock(add,pos,acquireLocks)) {
                Entry entry = edgeSerializer.writeRelation(add, pos, tx);
                mutator.acquireEdgeLock(idManager.getKey(vertex.longId()), entry.getColumn());
            }
        }
        indexUpdates.addAll(indexSerializer.getIndexUpdates(add));
    }

    //3) Collect all index update for vertices
    for (InternalVertex v : mutatedProperties.keySet()) {
        indexUpdates.addAll(indexSerializer.getIndexUpdates(v,mutatedProperties.get(v)));
    }
    //4) Acquire index locks (deletions first)
    for (IndexSerializer.IndexUpdate update : indexUpdates) {
        if (!update.isCompositeIndex() || !update.isDeletion()) continue;
        CompositeIndexType iIndex = (CompositeIndexType) update.getIndex();
        if (acquireLock(iIndex,acquireLocks)) {
            mutator.acquireIndexLock((StaticBuffer)update.getKey(), (Entry)update.getEntry());
        }
    }
    for (IndexSerializer.IndexUpdate update : indexUpdates) {
        if (!update.isCompositeIndex() || !update.isAddition()) continue;
        CompositeIndexType iIndex = (CompositeIndexType) update.getIndex();
        if (acquireLock(iIndex,acquireLocks)) {
            mutator.acquireIndexLock((StaticBuffer)update.getKey(), ((Entry)update.getEntry()).getColumn());
        }
    }

    //5) Add relation mutations
    for (Long vertexid : mutations.keySet()) {
        Preconditions.checkArgument(vertexid > 0, "Vertex has no id: %s", vertexid);
        List<InternalRelation> edges = mutations.get(vertexid);
        List<Entry> additions = new ArrayList<Entry>(edges.size());
        List<Entry> deletions = new ArrayList<Entry>(Math.max(10, edges.size() / 10));
        for (InternalRelation edge : edges) {
            InternalRelationType baseType = (InternalRelationType) edge.getType();
            assert baseType.getBaseType()==null;

            for (InternalRelationType type : baseType.getRelationIndexes()) {
                if (type.getStatus()== SchemaStatus.DISABLED) continue;
                for (int pos = 0; pos < edge.getArity(); pos++) {
                    if (!type.isUnidirected(Direction.BOTH) && !type.isUnidirected(EdgeDirection.fromPosition(pos)))
                        continue; //Directionality is not covered
                    if (edge.getVertex(pos).longId()==vertexid) {
                        StaticArrayEntry entry = edgeSerializer.writeRelation(edge, type, pos, tx);
                        if (edge.isRemoved()) {
                            deletions.add(entry);
                        } else {
                            Preconditions.checkArgument(edge.isNew());
                            int ttl = getTTL(edge);
                            if (ttl > 0) {
                                entry.setMetaData(EntryMetaData.TTL, ttl);
                            }
                            additions.add(entry);
                        }
                    }
                }
            }
        }

        StaticBuffer vertexKey = idManager.getKey(vertexid);
        mutator.mutateEdges(vertexKey, additions, deletions);
    }

    //6) Add index updates
    boolean has2iMods = false;
    for (IndexSerializer.IndexUpdate indexUpdate : indexUpdates) {
        assert indexUpdate.isAddition() || indexUpdate.isDeletion();
        if (indexUpdate.isCompositeIndex()) {
            IndexSerializer.IndexUpdate<StaticBuffer,Entry> update = indexUpdate;
            if (update.isAddition())
                mutator.mutateIndex(update.getKey(), Lists.newArrayList(update.getEntry()), KCVSCache.NO_DELETIONS);
            else
                mutator.mutateIndex(update.getKey(), KeyColumnValueStore.NO_ADDITIONS, Lists.newArrayList(update.getEntry()));
        } else {
            IndexSerializer.IndexUpdate<String,IndexEntry> update = indexUpdate;
            has2iMods = true;
            IndexTransaction itx = mutator.getIndexTransaction(update.getIndex().getBackingIndexName());
            String indexStore = ((MixedIndexType)update.getIndex()).getStoreName();
            if (update.isAddition())
                itx.add(indexStore, update.getKey(), update.getEntry(), update.getElement().isNew());
            else
                itx.delete(indexStore,update.getKey(),update.getEntry().field,update.getEntry().value,update.getElement().isRemoved());
        }
    }
    return new ModificationSummary(!mutations.isEmpty(),has2iMods);
}

public static void load(final TitanGraph graph, String mixedIndexName, boolean uniqueNameCompositeIndex) {

    //Create Schema
    TitanManagement mgmt = graph.openManagement();
    final PropertyKey name = mgmt.makePropertyKey("name").dataType(String.class).make();
    TitanManagement.IndexBuilder nameIndexBuilder = mgmt.buildIndex("name", Vertex.class).addKey(name);
    if (uniqueNameCompositeIndex)
        nameIndexBuilder.unique();
    TitanGraphIndex namei = nameIndexBuilder.buildCompositeIndex();
    mgmt.setConsistency(namei, ConsistencyModifier.LOCK);
    final PropertyKey age = mgmt.makePropertyKey("age").dataType(Integer.class).make();
    if (null != mixedIndexName)
        mgmt.buildIndex("vertices", Vertex.class).addKey(age).buildMixedIndex(mixedIndexName);

    final PropertyKey time = mgmt.makePropertyKey("time").dataType(Integer.class).make();
    final PropertyKey reason = mgmt.makePropertyKey("reason").dataType(String.class).make();
    final PropertyKey place = mgmt.makePropertyKey("place").dataType(Geoshape.class).make();
    if (null != mixedIndexName)
        mgmt.buildIndex("edges", Edge.class).addKey(reason).addKey(place).buildMixedIndex(mixedIndexName);

    mgmt.makeEdgeLabel("father").multiplicity(Multiplicity.MANY2ONE).make();
    mgmt.makeEdgeLabel("mother").multiplicity(Multiplicity.MANY2ONE).make();
    EdgeLabel battled = mgmt.makeEdgeLabel("battled").signature(time).make();
    mgmt.buildEdgeIndex(battled, "battlesByTime", Direction.BOTH, Order.decr, time);
    mgmt.makeEdgeLabel("lives").signature(reason).make();
    mgmt.makeEdgeLabel("pet").make();
    mgmt.makeEdgeLabel("brother").make();

    mgmt.makeVertexLabel("titan").make();
    mgmt.makeVertexLabel("location").make();
    mgmt.makeVertexLabel("god").make();
    mgmt.makeVertexLabel("demigod").make();
    mgmt.makeVertexLabel("human").make();
    mgmt.makeVertexLabel("monster").make();

    mgmt.commit();

    TitanTransaction tx = graph.newTransaction();
    // vertices

    Vertex saturn = tx.addVertex(T.label, "titan", "name", "saturn", "age", 10000);

    Vertex sky = tx.addVertex(T.label, "location", "name", "sky");

    Vertex sea = tx.addVertex(T.label, "location", "name", "sea");

    Vertex jupiter = tx.addVertex(T.label, "god", "name", "jupiter", "age", 5000);

    Vertex neptune = tx.addVertex(T.label, "god", "name", "neptune", "age", 4500);

    Vertex hercules = tx.addVertex(T.label, "demigod", "name", "hercules", "age", 30);

    Vertex alcmene = tx.addVertex(T.label, "human", "name", "alcmene", "age", 45);

    Vertex pluto = tx.addVertex(T.label, "god", "name", "pluto", "age", 4000);

    Vertex nemean = tx.addVertex(T.label, "monster", "name", "nemean");

    Vertex hydra = tx.addVertex(T.label, "monster", "name", "hydra");

    Vertex cerberus = tx.addVertex(T.label, "monster", "name", "cerberus");

    Vertex tartarus = tx.addVertex(T.label, "location", "name", "tartarus");

    // edges

    jupiter.addEdge("father", saturn);
    jupiter.addEdge("lives", sky, "reason", "loves fresh breezes");
    jupiter.addEdge("brother", neptune);
    jupiter.addEdge("brother", pluto);

    neptune.addEdge("lives", sea).property("reason", "loves waves");
    neptune.addEdge("brother", jupiter);
    neptune.addEdge("brother", pluto);

    hercules.addEdge("father", jupiter);
    hercules.addEdge("mother", alcmene);
    hercules.addEdge("battled", nemean, "time", 1, "place", Geoshape.point(38.1f, 23.7f));
    hercules.addEdge("battled", hydra, "time", 2, "place", Geoshape.point(37.7f, 23.9f));
    hercules.addEdge("battled", cerberus, "time", 12, "place", Geoshape.point(39f, 22f));

    pluto.addEdge("brother", jupiter);
    pluto.addEdge("brother", neptune);
    pluto.addEdge("lives", tartarus, "reason", "no fear of death");
    pluto.addEdge("pet", cerberus);

    cerberus.addEdge("lives", tartarus);

    // commit the transaction to disk
    tx.commit();
}