Repository: diennea/herddb
Pull request #2 (2016-04-07 15:02:26)

Transactions prototype


Patch:
diff --git a/herddb-core/pom.xml b/herddb-core/pom.xml
index c8cb51ac..01e187a9 100644
--- a/herddb-core/pom.xml
+++ b/herddb-core/pom.xml
@@ -11,8 +11,6 @@
     <artifactId>herddb-core</artifactId>
     <name>HerdDB Core</name>
     <packaging>jar</packaging>
-    <dependencies>       
-    </dependencies>
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
diff --git a/herddb-core/src/main/java/herddb/core/DBManager.java b/herddb-core/src/main/java/herddb/core/DBManager.java
index f7be60bf..1ee7b458 100644
--- a/herddb-core/src/main/java/herddb/core/DBManager.java
+++ b/herddb-core/src/main/java/herddb/core/DBManager.java
@@ -23,6 +23,8 @@ import herddb.log.CommitLog;
 import herddb.log.CommitLogManager;
 import herddb.log.LogNotAvailableException;
 import herddb.metadata.MetadataStorageManager;
+import herddb.metadata.MetadataStorageManagerException;
+import herddb.model.DDLException;
 import herddb.model.DDLStatementExecutionResult;
 import herddb.model.DMLStatement;
 import herddb.model.DMLStatementExecutionResult;
@@ -31,15 +33,20 @@ import herddb.model.TableSpace;
 import herddb.model.Statement;
 import herddb.model.StatementExecutionException;
 import herddb.model.StatementExecutionResult;
-import herddb.model.Transaction;
 import herddb.model.commands.CreateTableSpaceStatement;
 import herddb.model.commands.GetStatement;
 import herddb.storage.DataStorageManager;
 import herddb.storage.DataStorageManagerException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -49,7 +56,7 @@ import java.util.logging.Logger;
  *
  * @author enrico.olivelli
  */
-public class DBManager {
+public class DBManager implements AutoCloseable {
 
     private final Logger LOGGER = Logger.getLogger(DBManager.class.getName());
     private final Map<String, TableSpaceManager> tablesSpaces = new ConcurrentHashMap<>();
@@ -58,44 +65,59 @@ public class DBManager {
     private final CommitLogManager commitLogManager;
     private final String nodeId;
     private final ReentrantReadWriteLock generalLock = new ReentrantReadWriteLock();
+    private final Thread activator;
+    private final AtomicBoolean stopped = new AtomicBoolean();
+    private final BlockingQueue<Object> activatorQueue = new LinkedBlockingDeque<>();
 
     public DBManager(String nodeId, MetadataStorageManager metadataStorageManager, DataStorageManager dataStorageManager, CommitLogManager commitLogManager) {
         this.metadataStorageManager = metadataStorageManager;
         this.dataStorageManager = dataStorageManager;
         this.commitLogManager = commitLogManager;
         this.nodeId = nodeId;
+        this.activator = new Thread(new Activator(), "hdb-" + nodeId + "-activator");
+        this.activator.setDaemon(true);
     }
 
     /**
      * Initial boot of the system
      */
-    public void start() throws DataStorageManagerException {
+    public void start() throws DataStorageManagerException, LogNotAvailableException, MetadataStorageManagerException {
+
+        metadataStorageManager.start();
+
+        activator.start();
+
         generalLock.writeLock().lock();
         try {
             dataStorageManager.start();
-            for (String tableSpace : metadataStorageManager.listTableSpaces()) {
-                bootTableSpace(tableSpace);
-            }
         } finally {
             generalLock.writeLock().unlock();
         }
+        triggerActivator();
     }
 
-    private void bootTableSpace(String tableSpaceName) throws DataStorageManagerException {
+    public boolean waitForTablespace(String tableSpace, int millis) throws InterruptedException {
+        long now = System.currentTimeMillis();
+        while (System.currentTimeMillis() - now <= millis) {
+            TableSpaceManager manager = tablesSpaces.get(tableSpace);
+            if (manager != null && manager.isLeader()) {
+                return true;
+            }
+            Thread.sleep(100);
+        }
+        return false;
+    }
+
+    private void bootTableSpace(String tableSpaceName) throws DataStorageManagerException, LogNotAvailableException, MetadataStorageManagerException, DDLException {
         TableSpace tableSpace = metadataStorageManager.describeTableSpace(tableSpaceName);
         if (!tableSpace.replicas.contains(nodeId)) {
             return;
         }
-        LOGGER.log(Level.SEVERE, "Booting tablespace " + tableSpaceName);
+        LOGGER.log(Level.SEVERE, "Booting tablespace {0} on {1}", new Object[]{tableSpaceName, nodeId});
         CommitLog commitLog = commitLogManager.createCommitLog(tableSpaceName);
-        try {
-            commitLog.startWriting();
-        } catch (LogNotAvailableException err) {
-            throw new DataStorageManagerException(err);
-        }
         generalLock.writeLock().lock();
         try {
-            TableSpaceManager manager = new TableSpaceManager(tableSpaceName, metadataStorageManager, dataStorageManager, commitLog);
+            TableSpaceManager manager = new TableSpaceManager(nodeId, tableSpaceName, metadataStorageManager, dataStorageManager, commitLog);
             tablesSpaces.put(tableSpaceName, manager);
             manager.start();
         } finally {
@@ -163,17 +185,22 @@ public class DBManager {
 
         try {
             metadataStorageManager.registerTableSpace(tableSpace);
-            if (tableSpace.replicas.contains(nodeId)) {
-                bootTableSpace(tableSpace.name);
-            }
+            triggerActivator();
             return new DDLStatementExecutionResult();
-        } catch (DataStorageManagerException err) {
+        } catch (Exception err) {
             throw new StatementExecutionException(err);
         }
     }
 
-    void close() throws DataStorageManagerException {
-        dataStorageManager.close();
+    @Override
+    public void close() throws DataStorageManagerException {
+        stopped.set(true);
+        triggerActivator();
+        try {
+            activator.join();
+        } catch (InterruptedException ignore) {
+            ignore.printStackTrace();
+        }
     }
 
     public void flush() throws DataStorageManagerException {
@@ -190,4 +217,65 @@ public class DBManager {
         }
     }
 
+    private void triggerActivator() {
+        activatorQueue.offer("");
+    }
+
+    private class Activator implements Runnable {
+
+        @Override
+        public void run() {
+            try {
+                while (!stopped.get()) {
+                    activatorQueue.take();
+                    if (!stopped.get()) {
+                        generalLock.writeLock().lock();
+                        try {
+                            Collection<String> actualTablesSpaces = metadataStorageManager.listTableSpaces();
+                            for (String tableSpace : actualTablesSpaces) {
+                                if (!tablesSpaces.containsKey(tableSpace)) {
+                                    try {
+                                        bootTableSpace(tableSpace);
+                                    } catch (Exception err) {
+                                        err.printStackTrace();
+                                    }
+                                }
+                            }
+                        } catch (MetadataStorageManagerException error) {
+                            LOGGER.log(Level.SEVERE, "cannot access tablespace metadata", error);
+                        } finally {
+                            generalLock.writeLock().unlock();
+                        }
+                    }
+                }
+
+            } catch (InterruptedException ee) {
+            }
+
+            generalLock.writeLock().lock();
+            try {
+                for (Map.Entry<String, TableSpaceManager> manager : tablesSpaces.entrySet()) {
+                    try {
+                        manager.getValue().close();
+                    } catch (Exception err) {
+                        LOGGER.log(Level.SEVERE, "error during shutdown of manager of tablespace " + manager.getKey(), err);
+                    }
+                }
+            } finally {
+                generalLock.writeLock().unlock();
+            }
+            try {
+                dataStorageManager.close();
+            } catch (Exception err) {
+                LOGGER.log(Level.SEVERE, "error during shutdown", err);
+            }
+            try {
+                metadataStorageManager.close();
+            } catch (Exception err) {
+                LOGGER.log(Level.SEVERE, "error during shutdown", err);
+            }
+            LOGGER.log(Level.SEVERE, "{0} activator stopped", nodeId);
+
+        }
+    }
 }
diff --git a/herddb-core/src/main/java/herddb/core/TableManager.java b/herddb-core/src/main/java/herddb/core/TableManager.java
index 12b36380..a1187559 100644
--- a/herddb-core/src/main/java/herddb/core/TableManager.java
+++ b/herddb-core/src/main/java/herddb/core/TableManager.java
@@ -102,7 +102,7 @@ public class TableManager {
     /**
      * Access to Pages
      */
-    private ReentrantReadWriteLock pagesLock = new ReentrantReadWriteLock(true);
+    private final ReentrantReadWriteLock pagesLock = new ReentrantReadWriteLock(true);
 
     /**
      * Definition of the table
@@ -110,13 +110,19 @@ public class TableManager {
     private Table table;
     private final CommitLog log;
     private final DataStorageManager dataStorageManager;
+    private final TableSpaceManager tableSpaceManager;
 
-    TableManager(Table table, CommitLog log, DataStorageManager dataStorageManager) {
+    TableManager(Table table, CommitLog log, DataStorageManager dataStorageManager, TableSpaceManager tableSpaceManager) {
         this.table = table;
+        this.tableSpaceManager = tableSpaceManager;
         this.log = log;
         this.dataStorageManager = dataStorageManager;
     }
 
+    public Table getTable() {
+        return table;
+    }
+
     public void start() throws DataStorageManagerException {
         LOGGER.log(Level.SEVERE, "loading in memory all the keys for table {1}", new Object[]{keyToPage.size(), table.name});
         pagesLock.writeLock().lock();
@@ -198,7 +204,7 @@ public class TableManager {
         }
     }
 
-    private StatementExecutionResult executeInsert(InsertStatement insert, Transaction transaction) throws StatementExecutionException {
+    private StatementExecutionResult executeInsert(InsertStatement insert, Transaction transaction) throws StatementExecutionException, DataStorageManagerException {
         /*
             an insert can succeed only if the row is valid and the "keys" structure  does not contain the requested key
             the insert will add the row in the 'buffer' without assigning a page to it
@@ -215,11 +221,6 @@ public class TableManager {
             LogEntry entry = LogEntryFactory.insert(table, record.key.data, record.value.data, transaction);
             log.log(entry);
             apply(entry);
-
-            if (transaction != null) {
-                transaction.registerInsertOnTable(table.name, key);
-            }
-
             return new DMLStatementExecutionResult(1, key);
         } catch (LogNotAvailableException err) {
             throw new StatementExecutionException(err);
@@ -257,10 +258,6 @@ public class TableManager {
                 // record does not match predicate
                 return new DMLStatementExecutionResult(0, key);
             }
-            if (transaction != null) {
-                transaction.registerChangedOnTable(this.table.name, actual);
-            }
-
             byte[] newValue = function.computeNewValue(actual);
             if (newValue == null) {
                 throw new NullPointerException("new value cannot be null");
@@ -300,8 +297,6 @@ public class TableManager {
                 // page always need to be loaded because the other records on that page will be rewritten on a new page
                 ensurePageLoaded(pageId);
                 actual = buffer.get(key);
-            } else if (transaction != null) {
-                transaction.registerChangedOnTable(this.table.name, actual);
             }
             if (delete.getPredicate() != null && !delete.getPredicate().evaluate(actual)) {
                 // record does not match predicate
@@ -349,15 +344,26 @@ public class TableManager {
         transaction.releaseLocksOnTable(table.name, locksManager);
     }
 
-    private void apply(LogEntry entry) {
+    void apply(LogEntry entry) throws DataStorageManagerException {
         switch (entry.type) {
             case LogEntryType.DELETE: {
                 // remove the record from the set of existing records
                 Bytes key = new Bytes(entry.key);
                 Long pageId = keyToPage.remove(key);
                 deletedKeys.add(key);
-                buffer.remove(key);
                 dirtyPages.add(pageId);
+                Record actual = buffer.get(key);
+                if (actual == null) {
+                    // page always need to be loaded because the other records on that page will be rewritten on a new page
+                    ensurePageLoaded(pageId);
+                    actual = buffer.get(key);
+                }
+                buffer.remove(key);
+                if (entry.transactionId > 0) {
+                    Transaction transaction = tableSpaceManager.getTransaction(entry.transactionId);
+                    transaction.registerChangedOnTable(this.table.name, actual);
+                }
+
                 break;
             }
             case LogEntryType.UPDATE: {
@@ -365,16 +371,29 @@ public class TableManager {
                 Bytes key = new Bytes(entry.key);
                 Bytes value = new Bytes(entry.value);
                 Long pageId = keyToPage.put(key, NO_PAGE);
-                buffer.put(key, new Record(key, value));
+                Record actual = buffer.put(key, new Record(key, value));
+                if (actual == null) {
+                    ensurePageLoaded(pageId);
+                    actual = buffer.get(key);
+                }
                 dirtyPages.add(pageId);
+                if (entry.transactionId > 0) {
+                    Transaction transaction = tableSpaceManager.getTransaction(entry.transactionId);
+                    transaction.registerChangedOnTable(this.table.name, actual);
+                }
                 break;
             }
             case LogEntryType.INSERT: {
+
                 Bytes key = new Bytes(entry.key);
                 Bytes value = new Bytes(entry.value);
                 keyToPage.put(key, NO_PAGE);
                 buffer.put(key, new Record(key, value));
                 deletedKeys.remove(key);
+                if (entry.transactionId > 0) {
+                    Transaction transaction = tableSpaceManager.getTransaction(entry.transactionId);
+                    transaction.registerInsertOnTable(table.name, key);
+                }
                 break;
             }
             default:
diff --git a/herddb-core/src/main/java/herddb/core/TableSpaceManager.java b/herddb-core/src/main/java/herddb/core/TableSpaceManager.java
index 04c72434..ab1a916b 100644
--- a/herddb-core/src/main/java/herddb/core/TableSpaceManager.java
+++ b/herddb-core/src/main/java/herddb/core/TableSpaceManager.java
@@ -22,8 +22,12 @@ package herddb.core;
 import herddb.log.CommitLog;
 import herddb.log.LogEntry;
 import herddb.log.LogEntryFactory;
+import herddb.log.LogEntryType;
 import herddb.log.LogNotAvailableException;
+import herddb.log.LogSequenceNumber;
 import herddb.metadata.MetadataStorageManager;
+import herddb.metadata.MetadataStorageManagerException;
+import herddb.model.DDLException;
 import herddb.model.TransactionResult;
 import herddb.model.DDLStatementExecutionResult;
 import herddb.model.Statement;
@@ -31,6 +35,8 @@ import herddb.model.StatementExecutionException;
 import herddb.model.StatementExecutionResult;
 import herddb.model.Table;
 import herddb.model.TableAwareStatement;
+import herddb.model.TableDoesNotExistException;
+import herddb.model.TableSpace;
 import herddb.model.Transaction;
 import herddb.model.commands.BeginTransactionStatement;
 import herddb.model.commands.CommitTransactionStatement;
@@ -38,12 +44,17 @@ import herddb.model.commands.CreateTableStatement;
 import herddb.model.commands.RollbackTransactionStatement;
 import herddb.storage.DataStorageManager;
 import herddb.storage.DataStorageManagerException;
+import herddb.utils.Bytes;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BiConsumer;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.stream.Collectors;
 
 /**
  * Manages a TableSet in memory
@@ -52,31 +63,143 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
  */
 public class TableSpaceManager {
 
+    private static final Logger LOGGER = Logger.getLogger(TableSpaceManager.class.getName());
+
     private final MetadataStorageManager metadataStorageManager;
     private final DataStorageManager dataStorageManager;
     private final CommitLog log;
     private final String tableSpaceName;
-    private final Map<String, TableManager> tables = new ConcurrentHashMap<>();
+    private final String nodeId;
+    private final Map<Bytes, TableManager> tables = new ConcurrentHashMap<>();
     private final ReentrantReadWriteLock generalLock = new ReentrantReadWriteLock();
     private final AtomicLong newTransactionId = new AtomicLong();
+    private final Map<String, Table> tablesMetadata = new ConcurrentHashMap<>();
+    private boolean leader;
 
-    public TableSpaceManager(String tableSpaceName, MetadataStorageManager metadataStorageManager, DataStorageManager dataStorageManager, CommitLog log) {
+    public TableSpaceManager(String nodeId, String tableSpaceName, MetadataStorageManager metadataStorageManager, DataStorageManager dataStorageManager, CommitLog log) {
+        this.nodeId = nodeId;
         this.metadataStorageManager = metadataStorageManager;
         this.dataStorageManager = dataStorageManager;
         this.log = log;
         this.tableSpaceName = tableSpaceName;
     }
 
-    void start() throws DataStorageManagerException {
-        generalLock.writeLock().lock();
-        try {
-            for (String tableName : metadataStorageManager.listTablesByTableSpace(tableSpaceName)) {
-                Table table = metadataStorageManager.describeTable(tableName);
+    void start() throws DataStorageManagerException, LogNotAvailableException, MetadataStorageManagerException, DDLException {
+
+        TableSpace tableSpaceInfo = metadataStorageManager.describeTableSpace(tableSpaceName);
+        recover(tableSpaceInfo);
+
+        tableSpaceInfo = metadataStorageManager.describeTableSpace(tableSpaceName);
+        if (tableSpaceInfo.leaderId.equals(nodeId)) {
+            startAsLeader();
+        }
+    }
+
+    void recover(TableSpace tableSpaceInfo) throws DataStorageManagerException, LogNotAvailableException {
+
+        List<Table> tablesAtBoot = dataStorageManager.loadTables(tableSpaceInfo.lastCheckpointLogPosition, tableSpaceName);
+        LOGGER.log(Level.SEVERE, "tablesAtBoot", tablesAtBoot.stream().map(t -> {
+            return t.name;
+        }).collect(Collectors.joining()));
+        for (Table table : tablesAtBoot) {
+            bootTable(table);
+        }
+
+        LOGGER.log(Level.SEVERE, "recovering tablespace " + tableSpaceName + " log from sequence number " + tableSpaceInfo.lastCheckpointLogPosition);
+
+        log.recovery(tableSpaceInfo.lastCheckpointLogPosition, new BiConsumer<LogSequenceNumber, LogEntry>() {
+            @Override
+            public void accept(LogSequenceNumber t, LogEntry u) {
+                try {
+                    apply(u);
+                } catch (Exception err) {
+                    throw new RuntimeException(err);
+                }
+            }
+        }, false);
+
+    }
+
+    void apply(LogEntry entry) throws DataStorageManagerException, DDLException {
+        LOGGER.log(Level.SEVERE, "apply entry {0}", new Object[]{entry});
+        switch (entry.type) {
+            case LogEntryType.BEGINTRANSACTION: {
+                long id = entry.transactionId;
+                Transaction transaction = new Transaction(id, tableSpaceName);
+                transactions.put(id, transaction);
+            }
+            break;
+            case LogEntryType.ROLLBACKTRANSACTION: {
+                long id = entry.transactionId;
+                Transaction transaction = transactions.get(id);
+                List<TableManager> managers;
+                try {
+                    generalLock.writeLock().lock();
+                    managers = new ArrayList<>(tables.values());
+                } finally {
+                    generalLock.writeLock().unlock();
+                }
+                for (TableManager manager : managers) {
+                    if (transaction.getNewTables().containsKey(manager.getTable().name)) {
+                        manager.close();
+                        tables.remove(Bytes.from_string(manager.getTable().name));
+                    } else {
+                        manager.onTransactionRollback(transaction);
+                    }
+                }
+                transactions.remove(transaction.transactionId);
+            }
+            break;
+            case LogEntryType.COMMITTRANSACTION: {
+                long id = entry.transactionId;
+                Transaction transaction = transactions.get(id);
+                List<TableManager> managers;
+                try {
+                    generalLock.writeLock().lock();
+                    managers = new ArrayList<>(tables.values());
+                } finally {
+                    generalLock.writeLock().unlock();
+                }
+                for (TableManager manager : managers) {
+                    manager.onTransactionCommit(transaction);
+                }
+                transactions.remove(transaction.transactionId);
+            }
+            break;
+            case LogEntryType.CREATE_TABLE: {
+                Table table = Table.deserialize(entry.value);
+                if (entry.transactionId > 0) {
+                    long id = entry.transactionId;
+                    Transaction transaction = transactions.get(id);
+                    transaction.registerNewTable(table);
+                }
                 bootTable(table);
             }
-        } finally {
-            generalLock.writeLock().unlock();
+            ;
+            break;
+        }
+
+        if (entry.tableName != null) {
+            TableManager tableManager = tables.get(new Bytes(entry.tableName));
+            tableManager.apply(entry);
+        }
+
+    }
+
+    void startAsLeader() throws DataStorageManagerException, DDLException, LogNotAvailableException {
+
+        // every pending transaction MUST be rollback back
+        List<Long> pending_transactions = new ArrayList<>(this.transactions.keySet());
+        log.startWriting();
+        LOGGER.log(Level.SEVERE, "startAsLeader tablespace " + tableSpaceName + " log, there were " + transactions.size() + " pending transactions to be rolledback");
+        for (long tx : pending_transactions) {
+            LOGGER.log(Level.SEVERE, "rolling back transaction " + tx);
+            LogEntry rollback = LogEntryFactory.rollbackTransaction(tableSpaceName, tx);
+            // let followers see the rollback on the log
+            log.log(rollback);
+            apply(rollback);
         }
+        leader = true;
     }
 
     private ConcurrentHashMap<Long, Transaction> transactions = new ConcurrentHashMap<>();
@@ -108,12 +231,12 @@ public class TableSpaceManager {
             TableManager manager;
             generalLock.readLock().lock();
             try {
-                manager = tables.get(table);
+                manager = tables.get(Bytes.from_string(table));
             } finally {
                 generalLock.readLock().unlock();
             }
             if (manager == null) {
-                throw new StatementExecutionException("no table " + table + " in tablespace " + tableSpaceName);
+                throw new TableDoesNotExistException("no table " + table + " in tablespace " + tableSpaceName);
             }
             return manager.executeStatement(statement, transaction);
         }
@@ -125,16 +248,15 @@ public class TableSpaceManager {
         try {
             generalLock.writeLock().lock();
 
+            LogEntry entry = LogEntryFactory.createTable(statement.getTableDefinition(), transaction);
             try {
-                LogEntry entry = LogEntryFactory.createTable(statement.getTableDefinition(), transaction);
                 log.log(entry);
             } catch (LogNotAvailableException ex) {
                 throw new StatementExecutionException(ex);
             }
 
-            Table table = statement.getTableDefinition();
-            metadataStorageManager.registerTable(table);
-            bootTable(table);
+            apply(entry);
+
             return new DDLStatementExecutionResult();
         } catch (DataStorageManagerException err) {
             throw new StatementExecutionException(err);
@@ -144,17 +266,19 @@ public class TableSpaceManager {
     }
 
     private void bootTable(Table table) throws DataStorageManagerException {
-        TableManager tableManager = new TableManager(table, log, dataStorageManager);
-        tables.put(table.name, tableManager);
+        LOGGER.log(Level.SEVERE, "bootTable", table.name);
+        TableManager tableManager = new TableManager(table, log, dataStorageManager, this);
+        tables.put(Bytes.from_string(table.name), tableManager);
         tableManager.start();
     }
 
-    public void close() {
+    public void close() throws LogNotAvailableException {
         try {
             generalLock.writeLock().lock();
             for (TableManager table : tables.values()) {
                 table.close();
             }
+            log.close();
         } finally {
             generalLock.writeLock().unlock();
         }
@@ -175,14 +299,15 @@ public class TableSpaceManager {
 
     private StatementExecutionResult beginTransaction() throws StatementExecutionException {
         long id = newTransactionId.incrementAndGet();
-        Transaction transaction = new Transaction(id, tableSpaceName);
-        LogEntry entry = LogEntryFactory.beginTransaction(tableSpaceName, transaction);
+
+        LogEntry entry = LogEntryFactory.beginTransaction(tableSpaceName, id);
         try {
             log.log(entry);
-        } catch (LogNotAvailableException err) {
+            apply(entry);
+        } catch (Exception err) {
             throw new StatementExecutionException(err);
         }
-        transactions.put(id, transaction);
+
         return new TransactionResult(id);
     }
 
@@ -191,23 +316,14 @@ public class TableSpaceManager {
         if (tx == null) {
             throw new StatementExecutionException("no such transaction " + rollbackTransactionStatement.getTransactionId());
         }
-        LogEntry entry = LogEntryFactory.rollbackTransaction(tableSpaceName, tx);
+        LogEntry entry = LogEntryFactory.rollbackTransaction(tableSpaceName, tx.transactionId);
         try {
             log.log(entry);
-        } catch (LogNotAvailableException err) {
+            apply(entry);
+        } catch (Exception err) {
             throw new StatementExecutionException(err);
         }
-        List<TableManager> managers;
-        try {
-            generalLock.writeLock().lock();
-            managers = new ArrayList<>(tables.values());
-        } finally {
-            generalLock.writeLock().unlock();
-        }
-        for (TableManager manager : managers) {
-            manager.onTransactionRollback(tx);
-        }
-        transactions.remove(tx.transactionId);
+
         return new TransactionResult(tx.transactionId);
     }
 
@@ -216,24 +332,23 @@ public class TableSpaceManager {
         if (tx == null) {
             throw new StatementExecutionException("no such transaction " + commitTransactionStatement.getTransactionId());
         }
-        LogEntry entry = LogEntryFactory.commitTransaction(tableSpaceName, tx);
+        LogEntry entry = LogEntryFactory.commitTransaction(tableSpaceName, tx.transactionId);
         try {
             log.log(entry);
-        } catch (LogNotAvailableException err) {
+            apply(entry);
+        } catch (Exception err) {
             throw new StatementExecutionException(err);
         }
-        List<TableManager> managers;
-        try {
-            generalLock.writeLock().lock();
-            managers = new ArrayList<>(tables.values());
-        } finally {
-            generalLock.writeLock().unlock();
-        }
-        for (TableManager manager : managers) {
-            manager.onTransactionCommit(tx);
-        }
-        transactions.remove(tx.transactionId);
+
         return new TransactionResult(tx.transactionId);
     }
 
+    public boolean isLeader() {
+        return leader;
+    }
+
+    Transaction getTransaction(long transactionId) {
+        return transactions.get(transactionId);
+    }
+
 }
diff --git a/herddb-core/src/main/java/herddb/file/FileCommitLog.java b/herddb-core/src/main/java/herddb/file/FileCommitLog.java
index 97bd62d5..35bdfb7f 100644
--- a/herddb-core/src/main/java/herddb/file/FileCommitLog.java
+++ b/herddb-core/src/main/java/herddb/file/FileCommitLog.java
@@ -23,19 +23,14 @@ import herddb.log.CommitLog;
 import herddb.log.LogEntry;
 import herddb.log.LogNotAvailableException;
 import herddb.log.LogSequenceNumber;
-import herddb.utils.FileUtils;
-import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.Comparator;
@@ -44,8 +39,6 @@ import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import java.util.zip.GZIPInputStream;
-import java.util.zip.GZIPOutputStream;
 
 /**
  * Commit log on file
@@ -65,8 +58,7 @@ public class FileCommitLog extends CommitLog {
 
     private volatile CommitFileWriter writer;
 
-    private final ReentrantLock writeLock = new ReentrantLock();
-    private final ReentrantLock snapshotLock = new ReentrantLock();
+    private final ReentrantLock writeLock = new ReentrantLock();    
 
     private final static byte ENTRY_START = 13;
     private final static byte ENTRY_END = 25;
diff --git a/herddb-core/src/main/java/herddb/file/FileCommitLogManager.java b/herddb-core/src/main/java/herddb/file/FileCommitLogManager.java
new file mode 100644
index 00000000..69953977
--- /dev/null
+++ b/herddb-core/src/main/java/herddb/file/FileCommitLogManager.java
@@ -0,0 +1,50 @@
+/*
+ Licensed to Diennea S.r.l. under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. Diennea S.r.l. licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ */
+package herddb.file;
+
+import herddb.log.CommitLog;
+import herddb.log.CommitLogManager;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+/**
+ * Commit logs on local files
+ * @author enrico.olivelli
+ */
+public class FileCommitLogManager extends CommitLogManager {
+
+    private final Path baseDirectory;
+
+    public FileCommitLogManager(Path baseDirectory) {
+        this.baseDirectory = baseDirectory;
+    }
+
+    @Override
+    public CommitLog createCommitLog(String tableSpace) {
+        try {
+            Path folder = baseDirectory.resolve(tableSpace + ".txlog");
+            Files.createDirectories(folder);
+            return new FileCommitLog(folder, 1024 * 1024);
+        } catch (IOException err) {
+            throw new RuntimeException(err);
+        }
+    }
+}
diff --git a/herddb-core/src/main/java/herddb/file/SimpleFileDataStorageManager.java b/herddb-core/src/main/java/herddb/file/FileDataStorageManager.java
similarity index 62%
rename from herddb-core/src/main/java/herddb/file/SimpleFileDataStorageManager.java
rename to herddb-core/src/main/java/herddb/file/FileDataStorageManager.java
index 2c0ff5ea..c2330a21 100644
--- a/herddb-core/src/main/java/herddb/file/SimpleFileDataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/file/FileDataStorageManager.java
@@ -21,6 +21,7 @@ package herddb.file;
 
 import herddb.log.LogSequenceNumber;
 import herddb.model.Record;
+import herddb.model.Table;
 import herddb.storage.DataStorageManager;
 import herddb.storage.DataStorageManagerException;
 import herddb.utils.Bytes;
@@ -35,22 +36,26 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiConsumer;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
  * Data Storage on local filesystem
  *
  * @author enrico.olivelli
  */
-public class SimpleFileDataStorageManager extends DataStorageManager {
+public class FileDataStorageManager extends DataStorageManager {
 
+    private static final Logger LOGGER = Logger.getLogger(FileDataStorageManager.class.getName());
     private final Path baseDirectory;
     private final AtomicLong newPageId = new AtomicLong();
 
-    public SimpleFileDataStorageManager(Path baseDirectory) {
+    public FileDataStorageManager(Path baseDirectory) {
         this.baseDirectory = baseDirectory;
     }
 
@@ -68,8 +73,16 @@ public class SimpleFileDataStorageManager extends DataStorageManager {
 
     }
 
+    private Path getTablespaceDirectory(String tablespace) {
+        return baseDirectory.resolve(tablespace + ".tablespace");
+    }
+
+    private Path getTablespaceTablesMetadataFile(String tablespace, LogSequenceNumber sequenceNumber) {
+        return getTableDirectory(tablespace).resolve("tables." + sequenceNumber.ledgerId + "." + sequenceNumber.offset + ".tablesmetadata");
+    }
+
     private Path getTableDirectory(String tablename) {
-        return baseDirectory.resolve(tablename);
+        return baseDirectory.resolve(tablename + ".table");
     }
 
     private Path getPageFile(Path tableDirectory, Long pageId) {
@@ -156,7 +169,7 @@ public class SimpleFileDataStorageManager extends DataStorageManager {
         } catch (IOException err) {
             throw new DataStorageManagerException(err);
         }
-        try (OutputStream outputKeys = Files.newOutputStream(keys, StandardOpenOption.APPEND,StandardOpenOption.CREATE);
+        try (OutputStream outputKeys = Files.newOutputStream(keys, StandardOpenOption.APPEND, StandardOpenOption.CREATE);
                 DataOutputStream dataOutputKeys = new DataOutputStream(outputKeys)) {
             dataOutputKeys.writeInt(newPage.size());
             for (Record record : newPage) {
@@ -181,7 +194,6 @@ public class SimpleFileDataStorageManager extends DataStorageManager {
                 return path.toString().endsWith(".page");
             });) {
                 stream.forEach(p -> {
-                    System.out.println("GOT PAGE " + p.toString());
                     count.incrementAndGet();
                 });
             }
@@ -191,4 +203,67 @@ public class SimpleFileDataStorageManager extends DataStorageManager {
         }
     }
 
+    @Override
+    public List<Table> loadTables(LogSequenceNumber sequenceNumber, String tableSpace) throws DataStorageManagerException {
+        try {
+            Path tableSpaceDirectory = getTablespaceDirectory(tableSpace);
+            Files.createDirectories(tableSpaceDirectory);
+            Path file = getTablespaceTablesMetadataFile(tableSpace, sequenceNumber);
+            LOGGER.log(Level.SEVERE, "loadTables for tableSpace " + tableSpace + " from " + file.toAbsolutePath().toString());
+            if (!Files.isRegularFile(file)) {
+                LOGGER.log(Level.SEVERE, "file " + file.toAbsolutePath().toString()+" not found");
+                return Collections.emptyList();
+            }
+            try (InputStream in = Files.newInputStream(file);
+                    DataInputStream din = new DataInputStream(in);) {
+                String readname = din.readUTF();
+                if (!readname.equals(tableSpace)) {
+                    throw new DataStorageManagerException("file " + file.toAbsolutePath() + " is not for spablespace " + tableSpace);
+                }
+                long ledgerId = din.readLong();
+                long offset = din.readLong();
+                if (ledgerId != sequenceNumber.ledgerId || offset != sequenceNumber.offset) {
+                    throw new DataStorageManagerException("file " + file.toAbsolutePath() + " is not for sequence number " + sequenceNumber);
+                }
+                int numTables = din.readInt();
+                List<Table> res = new ArrayList<>();
+                for (int i = 0; i < numTables; i++) {
+                    int size = din.readInt();
+                    byte[] tableData = new byte[size];
+                    din.readFully(tableData);
+                    Table table = Table.deserialize(tableData);
+                    res.add(table);
+                }
+                return Collections.unmodifiableList(res);
+            }
+        } catch (IOException err) {
+            throw new DataStorageManagerException(err);
+        }
+    }
+
+    @Override
+    public void writeTables(String tableSpace, LogSequenceNumber sequenceNumber, List<Table> tables) throws DataStorageManagerException {        
+        try {
+            Path tableSpaceDirectory = getTablespaceDirectory(tableSpace);
+            Files.createDirectories(tableSpaceDirectory);
+            Path file = getTablespaceTablesMetadataFile(tableSpace, sequenceNumber);
+            LOGGER.log(Level.SEVERE, "writeTables for tableSpace " + tableSpace + " sequenceNumber "+sequenceNumber+" to " + file.toAbsolutePath().toString());
+            try (OutputStream out = Files.newOutputStream(file, StandardOpenOption.CREATE_NEW);
+                    DataOutputStream dout = new DataOutputStream(out)) {
+                dout.writeUTF(tableSpace);
+                dout.writeLong(sequenceNumber.ledgerId);
+                dout.writeLong(sequenceNumber.offset);
+                dout.writeInt(tables.size());
+                for (Table t : tables) {
+                    byte[] tableSerialized = t.serialize();
+                    dout.writeInt(tableSerialized.length);
+                    dout.write(tableSerialized);
+                }
+            }
+        } catch (IOException err) {
+            throw new DataStorageManagerException(err);
+        }
+
+    }
+
 }
diff --git a/herddb-core/src/main/java/herddb/file/FileMetadataStorageManager.java b/herddb-core/src/main/java/herddb/file/FileMetadataStorageManager.java
new file mode 100644
index 00000000..5bff83c1
--- /dev/null
+++ b/herddb-core/src/main/java/herddb/file/FileMetadataStorageManager.java
@@ -0,0 +1,165 @@
+/*
+ Licensed to Diennea S.r.l. under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. Diennea S.r.l. licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ */
+package herddb.file;
+
+import herddb.metadata.MetadataStorageManager;
+import herddb.metadata.MetadataStorageManagerException;
+import herddb.model.DDLException;
+import herddb.model.TableSpace;
+import herddb.model.TableSpaceAlreadyExistsException;
+import herddb.model.TableSpaceDoesNotExistException;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Metadata on local files. Useful for single instance deplyments
+ *
+ * @author enrico.olivelli
+ */
+public class FileMetadataStorageManager extends MetadataStorageManager {
+
+    private final Path baseDirectory;
+    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
+    private final Map<String, TableSpace> tableSpaces = new HashMap<>();
+    private final static Logger LOGGER = Logger.getLogger(FileMetadataStorageManager.class.getName());
+
+    public FileMetadataStorageManager(Path baseDirectory) {
+        this.baseDirectory = baseDirectory;
+    }
+
+    @Override
+    public void start() throws MetadataStorageManagerException {
+        try {
+            Files.createDirectories(baseDirectory);
+            reloadFromDisk();
+        } catch (IOException err) {
+            throw new MetadataStorageManagerException(err);
+        }
+    }
+
+    @Override
+    public void close() {
+
+    }
+
+    @Override
+    public Collection<String> listTableSpaces() {
+        lock.readLock().lock();
+        try {
+            return new ArrayList<>(tableSpaces.keySet());
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    @Override
+    public TableSpace describeTableSpace(String name) {
+        lock.readLock().lock();
+        try {
+            return tableSpaces.get(name);
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    @Override
+    public void registerTableSpace(TableSpace tableSpace) throws DDLException, MetadataStorageManagerException {
+        validateTableSpace(tableSpace);
+        lock.writeLock().lock();
+        try {
+            if (tableSpaces.containsKey(tableSpace.name)) {
+                throw new TableSpaceAlreadyExistsException("a tablespace named " + tableSpace.name + " already exists");
+            }
+            persistTableSpaceOnDisk(tableSpace);
+            tableSpaces.put(tableSpace.name, tableSpace);
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    @Override
+    public void updateTableSpace(TableSpace tableSpace) throws DDLException, MetadataStorageManagerException {
+        validateTableSpace(tableSpace);
+        lock.writeLock().lock();
+        try {
+            if (!tableSpaces.containsKey(tableSpace.name)) {
+                throw new TableSpaceDoesNotExistException("a tablespace named " + tableSpace.name + " does not exist");
+            }
+            persistTableSpaceOnDisk(tableSpace);
+            tableSpaces.put(tableSpace.name, tableSpace);
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+    private void reloadFromDisk() throws MetadataStorageManagerException {
+        tableSpaces.clear();
+        try (DirectoryStream<Path> stream = Files.newDirectoryStream(baseDirectory)) {
+            for (Path p : stream) {
+                String filename = p.getFileName().toString();
+                LOGGER.log(Level.SEVERE, "reading metadata file " + p.toAbsolutePath().toString());
+                if (filename.endsWith(".metadata")) {
+                    try (InputStream in = Files.newInputStream(p);
+                            DataInputStream iin = new DataInputStream(in);) {
+                        TableSpace ts = TableSpace.deserialize(iin);
+                        if (filename.equals(ts.name + ".metadata")) {
+                            tableSpaces.put(ts.name, ts);
+                        }
+                    }
+                }
+
+            }
+        } catch (IOException err) {
+            throw new MetadataStorageManagerException(err);
+        }
+    }
+
+    private void persistTableSpaceOnDisk(TableSpace tableSpace) throws MetadataStorageManagerException {
+        Path file_tmp = baseDirectory.resolve(tableSpace.name + "." + System.nanoTime() + ".tmpmetadata");
+        Path file = baseDirectory.resolve(tableSpace.name + ".metadata");
+        try (OutputStream out = Files.newOutputStream(file_tmp, StandardOpenOption.CREATE_NEW);
+                DataOutputStream dout = new DataOutputStream(out)) {
+            tableSpace.serialize(dout);
+        } catch (IOException err) {
+            throw new MetadataStorageManagerException(err);
+        }
+        try {
+            Files.move(file_tmp, file, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
+        } catch (IOException err) {
+            throw new MetadataStorageManagerException(err);
+        }
+    }
+
+}
diff --git a/herddb-core/src/main/java/herddb/log/LogEntry.java b/herddb-core/src/main/java/herddb/log/LogEntry.java
index d36cd721..5e98a24f 100644
--- a/herddb-core/src/main/java/herddb/log/LogEntry.java
+++ b/herddb-core/src/main/java/herddb/log/LogEntry.java
@@ -24,6 +24,7 @@ import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 
 /**
  * An entry on the log
@@ -87,6 +88,10 @@ public class LogEntry {
                         doo.writeInt(value.length);
                         doo.write(value);
                         break;
+                    case LogEntryType.BEGINTRANSACTION:
+                    case LogEntryType.COMMITTRANSACTION:
+                    case LogEntryType.ROLLBACKTRANSACTION:
+                        break;
                     default:
                         throw new IllegalArgumentException("unsupported type " + type);
                 }
@@ -129,6 +134,10 @@ public class LogEntry {
                     // value contains the table definition                                        
                     value = readArray(dis);
                     break;
+                case LogEntryType.BEGINTRANSACTION:
+                case LogEntryType.COMMITTRANSACTION:
+                case LogEntryType.ROLLBACKTRANSACTION:
+                    break;
                 default:
                     throw new IllegalArgumentException("unsupported type " + type);
             }
@@ -145,4 +154,9 @@ public class LogEntry {
         return res;
     }
 
+    @Override
+    public String toString() {
+        return "LogEntry{" + "type=" + type + ", tableSpace=" + new String(tableSpace, StandardCharsets.UTF_8) + ", transactionId=" + transactionId + ", tableName=" + tableName + '}';
+    }
+
 }
diff --git a/herddb-core/src/main/java/herddb/log/LogEntryFactory.java b/herddb-core/src/main/java/herddb/log/LogEntryFactory.java
index 395f1e24..c7797ec0 100644
--- a/herddb-core/src/main/java/herddb/log/LogEntryFactory.java
+++ b/herddb-core/src/main/java/herddb/log/LogEntryFactory.java
@@ -39,16 +39,16 @@ public class LogEntryFactory {
         return new LogEntry(System.currentTimeMillis(), LogEntryType.CREATE_TABLE, bytes(table.tablespace), transaction != null ? transaction.transactionId : 0, null, null, payload);
     }
 
-    public static LogEntry beginTransaction(String tablespace, Transaction transaction) {
-        return new LogEntry(System.currentTimeMillis(), LogEntryType.BEGINTRANSACTION, bytes(tablespace), transaction.transactionId, null, null, null);
+    public static LogEntry beginTransaction(String tablespace, long transactionId) {
+        return new LogEntry(System.currentTimeMillis(), LogEntryType.BEGINTRANSACTION, bytes(tablespace), transactionId, null, null, null);
     }
 
-    public static LogEntry commitTransaction(String tablespace, Transaction transaction) {
-        return new LogEntry(System.currentTimeMillis(), LogEntryType.COMMITTRANSACTION, bytes(tablespace), transaction.transactionId, null, null, null);
+    public static LogEntry commitTransaction(String tablespace, long transactionId) {
+        return new LogEntry(System.currentTimeMillis(), LogEntryType.COMMITTRANSACTION, bytes(tablespace), transactionId, null, null, null);
     }
 
-    public static LogEntry rollbackTransaction(String tablespace, Transaction transaction) {
-        return new LogEntry(System.currentTimeMillis(), LogEntryType.ROLLBACKTRANSACTION, bytes(tablespace), transaction.transactionId, null, null, null);
+    public static LogEntry rollbackTransaction(String tablespace, long transactionId) {
+        return new LogEntry(System.currentTimeMillis(), LogEntryType.ROLLBACKTRANSACTION, bytes(tablespace), transactionId, null, null, null);
     }
 
     public static LogEntry insert(Table table, byte[] key, byte[] value, Transaction transaction) {
diff --git a/herddb-core/src/main/java/herddb/mem/MemoryDataStorageManager.java b/herddb-core/src/main/java/herddb/mem/MemoryDataStorageManager.java
index b9442385..737653af 100644
--- a/herddb-core/src/main/java/herddb/mem/MemoryDataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/mem/MemoryDataStorageManager.java
@@ -21,14 +21,18 @@ package herddb.mem;
 
 import herddb.log.LogSequenceNumber;
 import herddb.model.Record;
+import herddb.model.Table;
 import herddb.storage.DataStorageManager;
 import herddb.storage.DataStorageManagerException;
 import herddb.utils.Bytes;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
 
 /**
  * In memory StorageManager, for tests
@@ -59,6 +63,7 @@ public class MemoryDataStorageManager extends DataStorageManager {
     private final ConcurrentHashMap<String, Page> pages = new ConcurrentHashMap<>();
     private final ConcurrentHashMap<String, Set<Bytes>> keysByPage = new ConcurrentHashMap<>();
     private final AtomicLong newPageId = new AtomicLong();
+    private final ConcurrentHashMap<String, List<Table>> tablesByTablespace = new ConcurrentHashMap<>();
 
     @Override
     public int getActualNumberOfPages(String tableName) throws DataStorageManagerException {
@@ -105,4 +110,30 @@ public class MemoryDataStorageManager extends DataStorageManager {
         this.keysByPage.clear();
     }
 
+    @Override
+    public List<Table> loadTables(LogSequenceNumber sequenceNumber, String tableSpace) throws DataStorageManagerException {
+        List<Table> res = tablesByTablespace.get(tableSpace);
+        if (res != null) {
+            return Collections.unmodifiableList(res);
+        } else {
+            return Collections.emptyList();
+        }
+    }
+
+    @Override
+    public void writeTables(String tableSpace, LogSequenceNumber sequenceNumber, List<Table> tables) throws DataStorageManagerException {
+
+        tables.forEach((t) -> {
+            if (!t.tablespace.equals(tableSpace)) {
+                throw new IllegalArgumentException("illegal tablespace");
+            }
+        });
+        List<Table> res = tablesByTablespace.get(tableSpace);
+        if (res == null) {
+            this.tablesByTablespace.put(tableSpace, new ArrayList<>(tables));
+        } else {
+            res.addAll(tables);
+        }
+    }
+
 }
diff --git a/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java b/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java
index 5ac2281c..a4845b01 100644
--- a/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java
@@ -21,10 +21,6 @@ package herddb.mem;
 
 import herddb.metadata.MetadataStorageManager;
 import herddb.model.DDLException;
-import herddb.model.InvalidTableException;
-import herddb.model.Table;
-import herddb.model.TableAlreadyExistsException;
-import herddb.model.TableDoesNotExistException;
 import herddb.model.TableSpace;
 import herddb.model.TableSpaceAlreadyExistsException;
 import herddb.model.TableSpaceDoesNotExistException;
@@ -33,7 +29,6 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.stream.Collectors;
 
 /**
  * In-memory MetadataStorageManager implementation. For tests
@@ -44,7 +39,6 @@ public class MemoryMetadataStorageManager extends MetadataStorageManager {
 
     private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
     private final Map<String, TableSpace> tableSpaces = new HashMap<>();
-    private final Map<String, Table> tables = new HashMap<>();
 
     @Override
     public Collection<String> listTableSpaces() {
@@ -56,20 +50,6 @@ public class MemoryMetadataStorageManager extends MetadataStorageManager {
         }
     }
 
-    @Override
-    public Collection<String> listTablesByTableSpace(String tableSpace) {
-        lock.readLock().lock();
-        try {
-            return tables.values().stream().filter(s -> {
-                return s.tablespace.equals(tableSpace);
-            }).map(t -> {
-                return t.name;
-            }).collect(Collectors.toList());
-        } finally {
-            lock.readLock().unlock();
-        }
-    }
-
     @Override
     public TableSpace describeTableSpace(String name) {
         lock.readLock().lock();
@@ -80,32 +60,6 @@ public class MemoryMetadataStorageManager extends MetadataStorageManager {
         }
     }
 
-    @Override
-    public Table describeTable(String name) {
-        lock.readLock().lock();
-        try {
-            return tables.get(name);
-        } finally {
-            lock.readLock().unlock();
-        }
-    }
-
-    @Override
-    public void registerTable(Table table) throws DDLException {
-        validateTable(table);
-        lock.writeLock().lock();
-        try {
-            if (!tableSpaces.containsKey(table.tablespace)) {
-                throw new TableSpaceDoesNotExistException(table.tablespace);
-            }
-            if (tables.putIfAbsent(table.name, table) != null) {
-                throw new TableAlreadyExistsException(table.name);
-            }
-        } finally {
-            lock.writeLock().unlock();
-        }
-    }
-
     @Override
     public void registerTableSpace(TableSpace tableSpace) throws DDLException {
         validateTableSpace(tableSpace);
@@ -119,27 +73,6 @@ public class MemoryMetadataStorageManager extends MetadataStorageManager {
         }
     }
 
-    @Override
-    public void updateTable(Table table) throws DDLException {
-        validateTable(table);
-        lock.writeLock().lock();
-        try {
-            Table prev = tables.get(table.name);
-            if (prev == null) {
-                throw new TableDoesNotExistException(table.name);
-            }
-            if (!prev.tablespace.equals(table.tablespace)) {
-                throw new InvalidTableException("cannot change tablespace");
-            }
-            if (!tableSpaces.containsKey(table.tablespace)) {
-                throw new TableSpaceDoesNotExistException(table.tablespace);
-            }
-            tables.put(table.name, table);
-        } finally {
-            lock.writeLock().unlock();
-        }
-    }
-
     @Override
     public void updateTableSpace(TableSpace tableSpace) throws DDLException {
         validateTableSpace(tableSpace);
@@ -155,18 +88,14 @@ public class MemoryMetadataStorageManager extends MetadataStorageManager {
         }
     }
 
-    private void validateTableSpace(TableSpace tableSpace) throws DDLException {
-        // TODO: implement sensible validations
-        if (tableSpace.name == null) {
-            throw new InvalidTableException("null tablespace name");
-        }
+    @Override
+    public void start() {
+
     }
 
-    private void validateTable(Table table) throws DDLException {
-        // TODO: implement sensible validations
-        if (table.name == null) {
-            throw new InvalidTableException("null table name");
-        }
+    @Override
+    public void close() {
+
     }
 
 }
diff --git a/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java
index 9eca1aef..e8fbec5f 100644
--- a/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java
@@ -20,7 +20,7 @@
 package herddb.metadata;
 
 import herddb.model.DDLException;
-import herddb.model.Table;
+import herddb.model.InvalidTableException;
 import herddb.model.TableSpace;
 import java.util.Collection;
 
@@ -30,21 +30,18 @@ import java.util.Collection;
  *
  * @author enrico.olivelli
  */
-public abstract class MetadataStorageManager {
+public abstract class MetadataStorageManager implements AutoCloseable {
 
-    /**
-     * Enumerates all the available TableSpaces in the system
-     *
-     * @return
-     */
-    public abstract Collection<String> listTableSpaces();
+    public abstract void start() throws MetadataStorageManagerException;
+
+    public abstract void close() throws MetadataStorageManagerException;
 
     /**
      * Enumerates all the available TableSpaces in the system
      *
      * @return
      */
-    public abstract Collection<String> listTablesByTableSpace(String tableSpace);
+    public abstract Collection<String> listTableSpaces() throws MetadataStorageManagerException;
 
     /**
      * Describe a single TableSpace
@@ -52,40 +49,27 @@ public abstract class MetadataStorageManager {
      * @param name
      * @return
      */
-    public abstract TableSpace describeTableSpace(String name);
-
-    /**
-     * Describe a single table
-     *
-     * @return
-     */
-    public abstract Table describeTable(String name);
-
-    /**
-     * Registers a new table on the metadata storage
-     *
-     * @param table
-     */
-    public abstract void registerTable(Table table) throws DDLException;
+    public abstract TableSpace describeTableSpace(String name) throws MetadataStorageManagerException;
 
     /**
      * Registers a new table space on the metadata storage
      *
      * @param tableSpace
      */
-    public abstract void registerTableSpace(TableSpace tableSpace) throws DDLException;
-
-    /**
-     * Updates table metadata on the metadata storage
-     *
-     * @param table
-     */
-    public abstract void updateTable(Table table) throws DDLException;
+    public abstract void registerTableSpace(TableSpace tableSpace) throws DDLException, MetadataStorageManagerException;
 
     /**
      * Updates table space metadata on the metadata storage
      *
      * @param tableSpace
      */
-    public abstract void updateTableSpace(TableSpace tableSpace) throws DDLException;
+    public abstract void updateTableSpace(TableSpace tableSpace) throws DDLException, MetadataStorageManagerException;
+
+    protected void validateTableSpace(TableSpace tableSpace) throws DDLException {
+        // TODO: implement sensible validations
+        if (tableSpace.name == null || tableSpace.name.trim().isEmpty()) {
+            throw new InvalidTableException("null tablespace name");
+        }
+    }
+
 }
diff --git a/herddb-core/src/main/java/herddb/metadata/MetadataStorageManagerException.java b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManagerException.java
new file mode 100644
index 00000000..072768be
--- /dev/null
+++ b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManagerException.java
@@ -0,0 +1,33 @@
+/*
+ Licensed to Diennea S.r.l. under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. Diennea S.r.l. licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ */
+package herddb.metadata;
+
+/**
+ * Error on Metadata
+ *
+ * @author enrico.olivelli
+ */
+public class MetadataStorageManagerException extends Exception {
+
+    public MetadataStorageManagerException(Throwable cause) {
+        super(cause);
+    }
+
+}
diff --git a/herddb-core/src/main/java/herddb/model/TableSpace.java b/herddb-core/src/main/java/herddb/model/TableSpace.java
index 304b421d..81cd71b7 100644
--- a/herddb-core/src/main/java/herddb/model/TableSpace.java
+++ b/herddb-core/src/main/java/herddb/model/TableSpace.java
@@ -19,6 +19,10 @@
  */
 package herddb.model;
 
+import herddb.log.LogSequenceNumber;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
@@ -45,21 +49,53 @@ public class TableSpace {
      */
     public final Set<String> replicas;
 
-    private TableSpace(String name, String leaderId, Set<String> replicas) {
+    public final LogSequenceNumber lastCheckpointLogPosition;
+
+    private TableSpace(String name, String leaderId, Set<String> replicas, LogSequenceNumber lastCheckpointLogPosition) {
         this.name = name;
         this.leaderId = leaderId;
         this.replicas = replicas;
+        this.lastCheckpointLogPosition = lastCheckpointLogPosition;
     }
 
     public static Builder builder() {
         return new Builder();
     }
 
+    public static TableSpace deserialize(DataInputStream in) throws IOException {
+        String name = in.readUTF();
+        String leaderId = in.readUTF();
+        int numreplicas = in.readInt();
+        Set<String> replicas = new HashSet<>();
+        for (int i = 0; i < numreplicas; i++) {
+            replicas.add(in.readUTF());
+        }
+        LogSequenceNumber number = new LogSequenceNumber(in.readLong(), in.readLong());
+        return new TableSpace(name, leaderId, replicas, number);
+    }
+
+    public void serialize(DataOutputStream out) throws IOException {
+        out.writeUTF(name);
+        out.writeUTF(leaderId);
+        out.writeInt(replicas.size());
+        for (String replica : replicas) {
+            out.writeUTF(replica);
+        }
+        if (lastCheckpointLogPosition != null) {
+            out.writeLong(-1);
+            out.writeLong(-1);
+        } else {
+            out.writeLong(lastCheckpointLogPosition.ledgerId);
+            out.writeLong(lastCheckpointLogPosition.offset);
+        }
+    }
+
     public static class Builder {
 
         private final Set<String> replicas = new HashSet<>();
         private String name;
         private String leaderId;
+        private LogSequenceNumber lastCheckpointLogPosition = new LogSequenceNumber(-1, -1);
 
         private Builder() {
         }
@@ -69,10 +105,16 @@ public class TableSpace {
             return this;
         }
 
+        public Builder lastCheckpointLogPosition(LogSequenceNumber lastCheckpointLogPosition) {
+            this.lastCheckpointLogPosition = lastCheckpointLogPosition;
+            return this;
+        }
+
         public Builder replica(String id) {
             this.replicas.add(id);
             return this;
         }
+
         public Builder replicas(Set<String> replicas) {
             this.replicas.addAll(replicas);
             return this;
@@ -96,7 +138,7 @@ public class TableSpace {
             if (!replicas.contains(leaderId)) {
                 throw new IllegalArgumentException("leader " + leaderId + " must be in replica list " + replicas);
             }
-            return new TableSpace(name, leaderId, Collections.unmodifiableSet(replicas));
+            return new TableSpace(name, leaderId, Collections.unmodifiableSet(replicas), lastCheckpointLogPosition);
         }
 
     }
diff --git a/herddb-core/src/main/java/herddb/model/Transaction.java b/herddb-core/src/main/java/herddb/model/Transaction.java
index ba6666cb..9cfbdd97 100644
--- a/herddb-core/src/main/java/herddb/model/Transaction.java
+++ b/herddb-core/src/main/java/herddb/model/Transaction.java
@@ -41,6 +41,7 @@ public class Transaction {
     public final Map<String, List<LockHandle>> locks;
     public final Map<String, List<Record>> changedRecords;
     public final Map<String, List<Bytes>> newRecords;
+    public final Map<String, Table> newTables;
 
     public Transaction(long transactionId, String tableSpace) {
         this.transactionId = transactionId;
@@ -48,6 +49,7 @@ public class Transaction {
         this.locks = new HashMap<>();
         this.changedRecords = new HashMap<>();
         this.newRecords = new HashMap<>();
+        this.newTables = new HashMap<>();
     }
 
     public List<Record> getChangedRecordsForTable(String tableName) {
@@ -58,6 +60,10 @@ public class Transaction {
         return newRecords.get(tableName);
     }
 
+    public Map<String, Table> getNewTables() {
+        return newTables;
+    }
+
     public LockHandle lookupLock(String tableName, Bytes key) {
         List<LockHandle> ll = locks.get(tableName);
         if (ll == null) {
@@ -80,6 +86,10 @@ public class Transaction {
         ll.add(handle);
     }
 
+    public void registerNewTable(Table table) {
+        newTables.put(table.name, table);
+    }
+
     public void registerChangedOnTable(String tableName, Record record) {
         List<Record> ll = changedRecords.get(tableName);
         if (ll == null) {
diff --git a/herddb-core/src/main/java/herddb/storage/DataStorageManager.java b/herddb-core/src/main/java/herddb/storage/DataStorageManager.java
index 48c23103..24dd66b0 100644
--- a/herddb-core/src/main/java/herddb/storage/DataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/storage/DataStorageManager.java
@@ -21,6 +21,7 @@ package herddb.storage;
 
 import herddb.log.LogSequenceNumber;
 import herddb.model.Record;
+import herddb.model.Table;
 import herddb.utils.Bytes;
 import java.util.List;
 import java.util.function.BiConsumer;
@@ -83,4 +84,22 @@ public abstract class DataStorageManager {
      */
     public abstract void close() throws DataStorageManagerException;
 
+    /**
+     * Load tables metadata
+     * @param sequenceNumber 
+     * @param tableSpace 
+     * @return
+     * @throws DataStorageManagerException
+     */
+    public abstract List<Table> loadTables(LogSequenceNumber sequenceNumber,String tableSpace) throws DataStorageManagerException;
+
+    /**
+     * Writes tables metadata
+     * @param sequenceNumber
+     * @param tableSpace 
+     * @param tables
+     * @throws DataStorageManagerException 
+     */
+    public abstract void writeTables(String tableSpace, LogSequenceNumber sequenceNumber, List<Table> tables) throws DataStorageManagerException;
+
 }


Test patch:
diff --git a/herddb-core/src/test/java/herddb/core/BaseTestcase.java b/herddb-core/src/test/java/herddb/core/BaseTestcase.java
index 8b332fe1..823e015e 100644
--- a/herddb-core/src/test/java/herddb/core/BaseTestcase.java
+++ b/herddb-core/src/test/java/herddb/core/BaseTestcase.java
@@ -32,6 +32,7 @@ import herddb.model.commands.CreateTableStatement;
 import herddb.storage.DataStorageManager;
 import java.util.Collections;
 import org.junit.After;
+import static org.junit.Assert.assertTrue;
 import org.junit.Before;
 
 /**
@@ -49,15 +50,15 @@ public class BaseTestcase {
     protected MetadataStorageManager metadataStorageManager;
     protected CommitLogManager commitLogManager;
 
-    protected CommitLogManager makeCommitLogManager() {
+    protected CommitLogManager makeCommitLogManager() throws Exception {
         return new MemoryCommitLogManager();
     }
 
-    protected DataStorageManager makeDataStorageManager() {
+    protected DataStorageManager makeDataStorageManager() throws Exception {
         return new MemoryDataStorageManager();
     }
 
-    protected MetadataStorageManager makeMetadataStorageManager() {
+    protected MetadataStorageManager makeMetadataStorageManager() throws Exception {
         return new MemoryMetadataStorageManager();
     }
 
@@ -71,6 +72,7 @@ public class BaseTestcase {
         manager.start();
         CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
         manager.executeStatement(st1);
+        assertTrue(manager.waitForTablespace(TableSpace.DEFAULT, 10000));
         tableName = "t1";
         table = Table
                 .builder()
@@ -88,5 +90,13 @@ public class BaseTestcase {
     public void teardown() throws Exception {
         manager.close();
         manager = null;
+
+        table = null;
+        tableName = null;
+
+        dataStorageManager = null;
+        metadataStorageManager = null;
+        commitLogManager = null;
+
     }
 }
diff --git a/herddb-core/src/test/java/herddb/core/CreateTableTest.java b/herddb-core/src/test/java/herddb/core/CreateTableTest.java
index 4470b9e7..7479bdfa 100644
--- a/herddb-core/src/test/java/herddb/core/CreateTableTest.java
+++ b/herddb-core/src/test/java/herddb/core/CreateTableTest.java
@@ -40,21 +40,23 @@ public class CreateTableTest {
     @Test
     public void createTable1() throws Exception {
         String nodeId = "localhost";
-        DBManager manager = new DBManager("localhost", new MemoryMetadataStorageManager(), new MemoryDataStorageManager(), new MemoryCommitLogManager());
-        manager.start();
-        CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
-        manager.executeStatement(st1);
-
-        Table table = Table
-                .builder()
-                .name("t1")
-                .column("id", ColumnTypes.STRING)
-                .column("name", ColumnTypes.STRING)
-                .primaryKey("id")
-                .build();
-
-        CreateTableStatement st2 = new CreateTableStatement(table);
-        manager.executeStatement(st2);
+        try (DBManager manager = new DBManager("localhost", new MemoryMetadataStorageManager(), new MemoryDataStorageManager(), new MemoryCommitLogManager());) {
+            manager.start();
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+        }
 
     }
 }
diff --git a/herddb-core/src/test/java/herddb/core/FlushFileTest.java b/herddb-core/src/test/java/herddb/core/FlushFileTest.java
index 97b2985e..5dadeb2b 100644
--- a/herddb-core/src/test/java/herddb/core/FlushFileTest.java
+++ b/herddb-core/src/test/java/herddb/core/FlushFileTest.java
@@ -20,7 +20,7 @@
 package herddb.core;
 
 import herddb.file.FileCommitLog;
-import herddb.file.SimpleFileDataStorageManager;
+import herddb.file.FileDataStorageManager;
 import herddb.log.CommitLog;
 import herddb.log.CommitLogManager;
 import herddb.mem.MemoryDataStorageManager;
@@ -71,7 +71,7 @@ public class FlushFileTest extends BaseTestcase {
     @Override
     protected DataStorageManager makeDataStorageManager() {
         try {
-            return new SimpleFileDataStorageManager(folder.newFolder("data").toPath());
+            return new FileDataStorageManager(folder.newFolder("data").toPath());
         } catch (IOException err) {
             throw new RuntimeException(err);
         }
diff --git a/herddb-core/src/test/java/herddb/core/SimpleDMLFileTest.java b/herddb-core/src/test/java/herddb/core/SimpleDMLFileTest.java
index 707dd30f..20a93da3 100644
--- a/herddb-core/src/test/java/herddb/core/SimpleDMLFileTest.java
+++ b/herddb-core/src/test/java/herddb/core/SimpleDMLFileTest.java
@@ -19,8 +19,7 @@
  */
 package herddb.core;
 
-import herddb.file.FileCommitLog;
-import herddb.log.CommitLog;
+import herddb.file.FileCommitLogManager;
 import herddb.log.CommitLogManager;
 import herddb.mem.MemoryDataStorageManager;
 import herddb.model.GetResult;
@@ -29,9 +28,7 @@ import herddb.model.Record;
 import herddb.model.commands.DeleteStatement;
 import herddb.model.commands.GetStatement;
 import herddb.model.commands.UpdateStatement;
-import herddb.model.predicates.RawValueEquals;
 import herddb.utils.Bytes;
-import java.io.IOException;
 import java.util.List;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -53,17 +50,8 @@ public class SimpleDMLFileTest extends BaseTestcase {
     public TemporaryFolder folder = new TemporaryFolder();
 
     @Override
-    protected CommitLogManager makeCommitLogManager() {
-        return new CommitLogManager() {
-            @Override
-            public CommitLog createCommitLog(String tableSpace) {
-                try {
-                    return new FileCommitLog(folder.newFolder(tableSpace).toPath(), 1024 * 1024);
-                } catch (IOException err) {
-                    throw new RuntimeException(err);
-                }
-            }
-        };
+    protected CommitLogManager makeCommitLogManager() throws Exception {
+        return new FileCommitLogManager(folder.newFolder().toPath());
     }
 
     @Test
diff --git a/herddb-core/src/test/java/herddb/core/SimpleRecoveryTest.java b/herddb-core/src/test/java/herddb/core/SimpleRecoveryTest.java
new file mode 100644
index 00000000..8e5b1666
--- /dev/null
+++ b/herddb-core/src/test/java/herddb/core/SimpleRecoveryTest.java
@@ -0,0 +1,495 @@
+/*
+ Licensed to Diennea S.r.l. under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. Diennea S.r.l. licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ */
+package herddb.core;
+
+import herddb.file.FileCommitLogManager;
+import herddb.file.FileDataStorageManager;
+import herddb.file.FileMetadataStorageManager;
+import herddb.model.ColumnTypes;
+import herddb.model.GetResult;
+import herddb.model.Record;
+import herddb.model.Table;
+import herddb.model.TableDoesNotExistException;
+import herddb.model.TableSpace;
+import herddb.model.TransactionResult;
+import herddb.model.commands.BeginTransactionStatement;
+import herddb.model.commands.CommitTransactionStatement;
+import herddb.model.commands.CreateTableSpaceStatement;
+import herddb.model.commands.CreateTableStatement;
+import herddb.model.commands.DeleteStatement;
+import herddb.model.commands.GetStatement;
+import herddb.model.commands.InsertStatement;
+import herddb.model.commands.RollbackTransactionStatement;
+import herddb.model.commands.UpdateStatement;
+import herddb.utils.Bytes;
+import java.nio.file.Path;
+import java.util.Collections;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Recovery from file
+ *
+ * @author enrico.olivelli
+ */
+public class SimpleRecoveryTest {
+
+    @Rule
+    public TemporaryFolder folder = new TemporaryFolder();
+
+    @Test
+    public void createTable1() throws Exception {
+
+        Path dataPath = folder.newFolder("data").toPath();
+        Path logsPath = folder.newFolder("logs").toPath();
+        Path metadataPath = folder.newFolder("metadata").toPath();
+
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+        }
+
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+        }
+
+    }
+
+    @Test
+    public void createInsertAndRestart() throws Exception {
+
+        Path dataPath = folder.newFolder("data").toPath();
+        Path logsPath = folder.newFolder("logs").toPath();
+        Path metadataPath = folder.newFolder("metadata").toPath();
+        Bytes key = Bytes.from_int(1234);
+        Bytes value = Bytes.from_long(8888);
+
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+
+            InsertStatement insert = new InsertStatement(TableSpace.DEFAULT, "t1", new Record(key, value));
+            assertEquals(1, manager.executeUpdate(insert).getUpdateCount());
+        }
+
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+            GetStatement get = new GetStatement(TableSpace.DEFAULT, "t1", key, null);
+            GetResult result = manager.get(get);
+            assertTrue(result.found());
+            assertEquals(key, result.getRecord().key);
+            assertEquals(value, result.getRecord().value);
+        }
+
+    }
+
+    @Test
+    public void createInsertInTransactionAndRestart() throws Exception {
+
+        Path dataPath = folder.newFolder("data").toPath();
+        Path logsPath = folder.newFolder("logs").toPath();
+        Path metadataPath = folder.newFolder("metadata").toPath();
+        Bytes key = Bytes.from_int(1234);
+        Bytes value = Bytes.from_long(8888);
+
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+
+            long tx = ((TransactionResult) manager.executeStatement(new BeginTransactionStatement(TableSpace.DEFAULT))).getTransactionId();
+
+            InsertStatement insert = new InsertStatement(TableSpace.DEFAULT, "t1", new Record(key, value)).setTransactionId(tx);
+            assertEquals(1, manager.executeUpdate(insert).getUpdateCount());
+
+            manager.executeStatement(new CommitTransactionStatement(TableSpace.DEFAULT, tx));
+        }
+
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+            GetStatement get = new GetStatement(TableSpace.DEFAULT, "t1", key, null);
+            GetResult result = manager.get(get);
+            assertTrue(result.found());
+            assertEquals(key, result.getRecord().key);
+            assertEquals(value, result.getRecord().value);
+        }
+
+    }
+
+    @Test
+    public void rollbackInsertTransactionOnRestart() throws Exception {
+
+        Path dataPath = folder.newFolder("data").toPath();
+        Path logsPath = folder.newFolder("logs").toPath();
+        Path metadataPath = folder.newFolder("metadata").toPath();
+        Bytes key = Bytes.from_int(1234);
+        Bytes key2 = Bytes.from_int(1235);
+        Bytes value = Bytes.from_long(8888);
+
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+
+            long tx = ((TransactionResult) manager.executeStatement(new BeginTransactionStatement(TableSpace.DEFAULT))).getTransactionId();
+            InsertStatement insert = new InsertStatement(TableSpace.DEFAULT, "t1", new Record(key, value)).setTransactionId(tx);
+            assertEquals(1, manager.executeUpdate(insert).getUpdateCount());
+
+            // transaction is not committed, and the leader dies, key MUST not appear anymore
+            InsertStatement insert2 = new InsertStatement(TableSpace.DEFAULT, "t1", new Record(key2, value));
+            assertEquals(1, manager.executeUpdate(insert2).getUpdateCount());
+
+        }
+
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            {
+                GetStatement get = new GetStatement(TableSpace.DEFAULT, "t1", key2, null);
+                GetResult result = manager.get(get);
+                assertTrue(result.found());
+                assertEquals(key2, result.getRecord().key);
+                assertEquals(value, result.getRecord().value);
+            }
+
+            {
+                // transaction rollback occurred
+                GetStatement get = new GetStatement(TableSpace.DEFAULT, "t1", key, null);
+                GetResult result = manager.get(get);
+                assertFalse(result.found());
+            }
+
+        }
+
+    }
+
+    @Test
+    public void rollbackDeleteTransactionOnRestart() throws Exception {
+
+        Path dataPath = folder.newFolder("data").toPath();
+        Path logsPath = folder.newFolder("logs").toPath();
+        Path metadataPath = folder.newFolder("metadata").toPath();
+        Bytes key = Bytes.from_int(1234);
+        Bytes value = Bytes.from_long(8888);
+
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+
+            InsertStatement insert = new InsertStatement(TableSpace.DEFAULT, "t1", new Record(key, value));
+            assertEquals(1, manager.executeUpdate(insert).getUpdateCount());
+
+            // transaction is not committed, and the leader dies, key appear again
+            long tx = ((TransactionResult) manager.executeStatement(new BeginTransactionStatement(TableSpace.DEFAULT))).getTransactionId();
+            DeleteStatement delete = new DeleteStatement(TableSpace.DEFAULT, "t1", key, null).setTransactionId(tx);
+            assertEquals(1, manager.executeUpdate(delete).getUpdateCount());
+
+        }
+
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            {
+                GetStatement get = new GetStatement(TableSpace.DEFAULT, "t1", key, null);
+                GetResult result = manager.get(get);
+                assertTrue(result.found());
+                assertEquals(key, result.getRecord().key);
+                assertEquals(value, result.getRecord().value);
+            }
+
+        }
+
+    }
+
+    @Test
+    public void rollbackUpdateTransactionOnRestart() throws Exception {
+
+        Path dataPath = folder.newFolder("data").toPath();
+        Path logsPath = folder.newFolder("logs").toPath();
+        Path metadataPath = folder.newFolder("metadata").toPath();
+        Bytes key = Bytes.from_int(1234);
+        Bytes value = Bytes.from_long(8888);
+        Bytes value2 = Bytes.from_long(8889);
+
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+
+            InsertStatement insert = new InsertStatement(TableSpace.DEFAULT, "t1", new Record(key, value));
+            assertEquals(1, manager.executeUpdate(insert).getUpdateCount());
+
+            // transaction is not committed, and the leader dies, key appear again with the old value
+            long tx = ((TransactionResult) manager.executeStatement(new BeginTransactionStatement(TableSpace.DEFAULT))).getTransactionId();
+            UpdateStatement update = new UpdateStatement(TableSpace.DEFAULT, "t1", new Record(key, value2), null).setTransactionId(tx);
+            assertEquals(1, manager.executeUpdate(update).getUpdateCount());
+
+        }
+
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            {
+                GetStatement get = new GetStatement(TableSpace.DEFAULT, "t1", key, null);
+                GetResult result = manager.get(get);
+                assertTrue(result.found());
+                assertEquals(key, result.getRecord().key);
+                assertEquals(value, result.getRecord().value);
+            }
+
+        }
+
+    }
+
+   
+    @Test
+    public void rollbackCreateTableOnRestart() throws Exception {
+
+        Path dataPath = folder.newFolder("data").toPath();
+        Path logsPath = folder.newFolder("logs").toPath();
+        Path metadataPath = folder.newFolder("metadata").toPath();
+        Bytes key = Bytes.from_int(1234);
+        Bytes value = Bytes.from_long(8888);        
+
+        String nodeId = "localhost";
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            CreateTableSpaceStatement st1 = new CreateTableSpaceStatement(TableSpace.DEFAULT, Collections.singleton(nodeId), nodeId);
+            manager.executeStatement(st1);
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            Table table = Table
+                    .builder()
+                    .name("t1")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            CreateTableStatement st2 = new CreateTableStatement(table);
+            manager.executeStatement(st2);
+            manager.flush();
+
+            Table transacted_table = Table
+                    .builder()
+                    .name("t2")
+                    .column("id", ColumnTypes.STRING)
+                    .column("name", ColumnTypes.STRING)
+                    .primaryKey("id")
+                    .build();
+
+            long tx = ((TransactionResult) manager.executeStatement(new BeginTransactionStatement(TableSpace.DEFAULT))).getTransactionId();
+            CreateTableStatement st_create = new CreateTableStatement(transacted_table).setTransactionId(tx);
+            manager.executeStatement(st_create);
+
+            InsertStatement insert = new InsertStatement(TableSpace.DEFAULT, "t2", new Record(key, value)).setTransactionId(tx);
+            assertEquals(1, manager.executeUpdate(insert).getUpdateCount());
+
+            GetStatement get = new GetStatement(TableSpace.DEFAULT, "t2", key, null).setTransactionId(tx);
+            GetResult result = manager.get(get);
+            assertTrue(result.found());
+            assertEquals(key, result.getRecord().key);
+            assertEquals(value, result.getRecord().value);
+
+            // transaction is not committed, and the leader dies, the table will disappear
+        }
+
+        try (DBManager manager = new DBManager("localhost",
+                new FileMetadataStorageManager(metadataPath),
+                new FileDataStorageManager(dataPath),
+                new FileCommitLogManager(logsPath))) {
+            manager.start();
+
+            manager.waitForTablespace(TableSpace.DEFAULT, 10000);
+
+            GetStatement get = new GetStatement(TableSpace.DEFAULT, "t2", key, null);
+            try {
+                manager.get(get);
+                fail();
+            } catch (TableDoesNotExistException error) {
+            }
+
+        }
+
+    }
+
+}
diff --git a/herddb-core/src/test/java/herddb/core/SimpleTransactionTest.java b/herddb-core/src/test/java/herddb/core/SimpleTransactionTest.java
index a7cf5d94..54059cfa 100644
--- a/herddb-core/src/test/java/herddb/core/SimpleTransactionTest.java
+++ b/herddb-core/src/test/java/herddb/core/SimpleTransactionTest.java
@@ -19,13 +19,19 @@
  */
 package herddb.core;
 
+import herddb.file.FileCommitLogManager;
+import herddb.file.FileDataStorageManager;
+import herddb.file.FileMetadataStorageManager;
 import herddb.model.ColumnTypes;
 import herddb.model.GetResult;
 import herddb.model.Record;
 import herddb.model.Table;
+import herddb.model.TableDoesNotExistException;
+import herddb.model.TableSpace;
 import herddb.model.TransactionResult;
 import herddb.model.commands.BeginTransactionStatement;
 import herddb.model.commands.CommitTransactionStatement;
+import herddb.model.commands.CreateTableSpaceStatement;
 import herddb.model.commands.CreateTableStatement;
 import herddb.model.commands.DeleteStatement;
 import herddb.model.commands.GetStatement;
@@ -33,9 +39,13 @@ import herddb.model.commands.InsertStatement;
 import herddb.model.commands.RollbackTransactionStatement;
 import herddb.model.commands.UpdateStatement;
 import herddb.utils.Bytes;
+import java.nio.file.Path;
+import java.util.Collections;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import org.junit.Ignore;
 import org.junit.Test;
 
 /**
@@ -260,11 +270,52 @@ public class SimpleTransactionTest extends BaseTestcase {
 
         GetResult get = manager.get(new GetStatement(tableSpace, tableName, key, null));
         assertTrue(get.found());
-        assertEquals(Bytes.from_int(0),get.getRecord().value);
-        
+        assertEquals(Bytes.from_int(0), get.getRecord().value);
+
         GetResult get2 = manager.get(new GetStatement(tableSpace, tableName2, key, null));
         assertTrue(get2.found());
-        assertEquals(Bytes.from_int(1),get2.getRecord().value);
+        assertEquals(Bytes.from_int(1), get2.getRecord().value);
+    }
+
+    @Test
+    public void rollbackCreateTable() throws Exception {
+
+        Bytes key = Bytes.from_int(1234);
+        Bytes value = Bytes.from_long(8888);
+
+        CreateTableStatement st2 = new CreateTableStatement(table);
+        manager.executeStatement(st2);
+        manager.flush();
+
+        Table transacted_table = Table
+                .builder()
+                .name("t2")
+                .column("id", ColumnTypes.STRING)
+                .column("name", ColumnTypes.STRING)
+                .primaryKey("id")
+                .build();
+
+        long tx = ((TransactionResult) manager.executeStatement(new BeginTransactionStatement(TableSpace.DEFAULT))).getTransactionId();
+        CreateTableStatement st_create = new CreateTableStatement(transacted_table).setTransactionId(tx);
+        manager.executeStatement(st_create);
+
+        InsertStatement insert = new InsertStatement(TableSpace.DEFAULT, "t2", new Record(key, value)).setTransactionId(tx);
+        assertEquals(1, manager.executeUpdate(insert).getUpdateCount());
+
+        GetStatement get = new GetStatement(TableSpace.DEFAULT, "t2", key, null).setTransactionId(tx);
+        GetResult result = manager.get(get);
+        assertTrue(result.found());
+        assertEquals(key, result.getRecord().key);
+        assertEquals(value, result.getRecord().value);
+
+        RollbackTransactionStatement rollback = new RollbackTransactionStatement(TableSpace.DEFAULT, tx);
+        manager.executeStatement(rollback);
+        try {
+            manager.get(new GetStatement(TableSpace.DEFAULT, "t2", key, null));
+            fail();
+        } catch (TableDoesNotExistException error) {
+        }
+
     }
 
 }