class SstSession extends Session {

        private final Map<String, Changes> batch;

        public SstSession() {
            this.batch = new HashMap<>();
        }

        @Override
        public void open() {
            this.opened = true;
        }

        @Override
        public void close() {
            assert this.closeable();
            this.opened = false;
        }

        /**
         * Any change in the session
         */
        @Override
        public boolean hasChanges() {
            return !this.batch.isEmpty();
        }

        /**
         * Commit all updates(put/delete) to DB
         */
        @Override
        public Integer commit() {<FILL_FUNCTION_BODY>}

        /**
         * Rollback updates not committed in the session
         */
        @Override
        public void rollback() {
            this.batch.clear();
        }

        @Override
        public String dataPath() {
            return RocksDBSstSessions.this.dataPath;
        }

        @Override
        public String walPath() {
            return RocksDBSstSessions.this.dataPath;
        }

        /**
         * Get property value by name from specified table
         */
        @Override
        public String property(String table, String property) {
            throw new NotSupportException("RocksDBSstStore property()");
        }

        @Override
        public Pair<byte[], byte[]> keyRange(String table) {
            return null;
        }

        @Override
        public void compactRange(String table) {
            throw new NotSupportException("RocksDBSstStore compactRange()");
        }

        /**
         * Add a KV record to a table
         */
        @Override
        public void put(String table, byte[] key, byte[] value) {
            Changes changes = this.batch.get(table);
            if (changes == null) {
                changes = new Changes();
                this.batch.put(table, changes);
            }
            changes.add(Pair.of(key, value));
        }

        /**
         * Merge a record to an existing key to a table
         * For more details about merge-operator:
         * <a href="https://github.com/facebook/rocksdb/wiki/merge-operator">...</a>
         */
        @Override
        public void merge(String table, byte[] key, byte[] value) {
            throw new NotSupportException("RocksDBSstStore merge()");
        }

        /**
         * Merge a record to an existing key to a table and commit immediately
         */
        @Override
        public void increase(String table, byte[] key, byte[] value) {
            throw new NotSupportException("RocksDBSstStore increase()");
        }

        /**
         * Delete a record by key from a table
         */
        @Override
        public void delete(String table, byte[] key) {
            throw new NotSupportException("RocksDBSstStore delete()");
        }

        /**
         * Delete the only one version of a record by key from a table
         * NOTE: requires that the key exists and was not overwritten.
         */
        @Override
        public void deleteSingle(String table, byte[] key) {
            throw new NotSupportException("RocksDBSstStore deleteSingle()");
        }

        /**
         * Delete a record by key(or prefix with key) from a table
         */
        @Override
        public void deletePrefix(String table, byte[] key) {
            throw new NotSupportException("RocksDBSstStore deletePrefix()");
        }

        /**
         * Delete a range of keys from a table
         */
        @Override
        public void deleteRange(String table, byte[] keyFrom, byte[] keyTo) {
            throw new NotSupportException("RocksDBSstStore deleteRange()");
        }

        /**
         * Get a record by key from a table
         */
        @Override
        public byte[] get(String table, byte[] key) {
            return null;
        }

        /**
         * Get records by a list of keys from a table
         */
        @Override
        public BackendColumnIterator get(String table, List<byte[]> keys) {
            assert !this.hasChanges();
            return BackendColumnIterator.empty();
        }

        /**
         * Scan all records from a table
         */
        @Override
        public BackendColumnIterator scan(String table) {
            assert !this.hasChanges();
            return BackendColumnIterator.empty();
        }

        /**
         * Scan records by key prefix from a table
         */
        @Override
        public BackendColumnIterator scan(String table, byte[] prefix) {
            assert !this.hasChanges();
            return BackendColumnIterator.empty();
        }

        /**
         * Scan records by key range from a table
         */
        @Override
        public BackendColumnIterator scan(String table, byte[] keyFrom,
                                          byte[] keyTo, int scanType) {
            assert !this.hasChanges();
            return BackendColumnIterator.empty();
        }
    }

            int count = this.batch.size();
            if (count == 0) {
                return 0;
            }

            try {
                for (Entry<String, Changes> table : this.batch.entrySet()) {
                    if (table.getValue().isEmpty() ||
                        table.getKey().endsWith("i")) {
                        // Skip empty value table or index table
                        continue;
                    }

                    // TODO: limit individual SST file size
                    SstFileWriter sst = table(table.getKey());
                    for (Pair<byte[], byte[]> change : table.getValue()) {
                        sst.put(change.getKey(), change.getValue());
                    }
                }
            } catch (RocksDBException e) {
                throw new BackendException("Failed to commit", e);
            }

            // Clear batch if write() successfully (retained if failed)
            this.batch.clear();

            return count;
        
<methods>public void <init>(HugeConfig, java.lang.String, java.lang.String) ,public abstract java.lang.String buildSnapshotPath(java.lang.String) ,public abstract void compactRange() ,public abstract org.apache.hugegraph.backend.store.rocksdb.RocksDBSessions copy(HugeConfig, java.lang.String, java.lang.String) ,public abstract void createSnapshot(java.lang.String) ,public transient abstract void createTable(java.lang.String[]) throws RocksDBException,public transient abstract void dropTable(java.lang.String[]) throws RocksDBException,public abstract boolean existsTable(java.lang.String) ,public abstract void forceCloseRocksDB() ,public abstract java.lang.String hardLinkSnapshot(java.lang.String) throws RocksDBException,public abstract Set<java.lang.String> openedTables() ,public abstract List<java.lang.String> property(java.lang.String) ,public abstract void reloadRocksDB() throws RocksDBException,public abstract void resumeSnapshot(java.lang.String) ,public abstract org.apache.hugegraph.backend.store.rocksdb.RocksDBSessions.Session session() <variables>