Repository: genjidb/genji
Pull request #151 (2020-08-17 21:54:13)

Fix bad tableInfoStore state on rollback
Since we started keeping table information in memory, we introduced a bug that prevents us from creating a table if this one has already been created during a transaction, then that transaction rolls back:

- Tx A creates a table named `foo`
- Tx A rolls back
- Tx B creates a table name `foo` -> Table already exists

This PR introduces transaction isolation at the tableInfoStore level by using transaction ids to hide tables that haven't been committed from other transactions.
That way, transactions are not aware of tables that have been created in transactions that haven't been committed yet, EXCEPT when other transactions attempt to create a table with the same name.



Patch:
diff --git a/database/config.go b/database/config.go
index bd6058e2..1ec207b5 100644
--- a/database/config.go
+++ b/database/config.go
@@ -66,6 +66,9 @@ func (f *FieldConstraint) ScanDocument(d document.Document) error {
 }
 
 type TableInfo struct {
+	// if non-zero, this tableInfo has been created during the current transaction.
+	// it will be removed if the transaction is rolled back or set to false if its commited.
+	transactionID int64
 	// storeID is a generated ID that acts as a key to reference a table.
 	// The first-4 bytes represents the timestamp in second and the last-2 bytes are
 	// randomly generated.
@@ -148,12 +151,18 @@ func newTableInfoStore(tx engine.Transaction) (*tableInfoStore, error) {
 }
 
 // Insert a new tableInfo for the given table name.
-func (t *tableInfoStore) Insert(tx engine.Transaction, tableName string, info *TableInfo) error {
+func (t *tableInfoStore) Insert(tx *Transaction, tableName string, info *TableInfo) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
 	_, ok := t.tableInfos[tableName]
 	if ok {
+		// TODO(asdine): if a table already exists but is uncommited,
+		// there is a chance the other transaction will be rolled back.
+		// Instead of returning an error, wait until the other transaction is
+		// either commited or rolled back.
+		// If it is commited, return an error here
+		// If not, create the table in this transaction.
 		return ErrTableAlreadyExists
 	}
 
@@ -162,7 +171,7 @@ func (t *tableInfoStore) Insert(tx engine.Transaction, tableName string, info *T
 		return err
 	}
 
-	st, err := tx.GetStore([]byte(tableInfoStoreName))
+	st, err := tx.Tx.GetStore([]byte(tableInfoStoreName))
 	if err != nil {
 		return err
 	}
@@ -172,11 +181,12 @@ func (t *tableInfoStore) Insert(tx engine.Transaction, tableName string, info *T
 		return err
 	}
 
+	info.transactionID = tx.id
 	t.tableInfos[tableName] = *info
 	return nil
 }
 
-func (t *tableInfoStore) Get(tableName string) (*TableInfo, error) {
+func (t *tableInfoStore) Get(tx *Transaction, tableName string) (*TableInfo, error) {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 
@@ -185,14 +195,27 @@ func (t *tableInfoStore) Get(tableName string) (*TableInfo, error) {
 		return nil, ErrTableNotFound
 	}
 
+	if info.transactionID != 0 && info.transactionID != tx.id {
+		return nil, ErrTableNotFound
+	}
+
 	return &info, nil
 }
 
-func (t *tableInfoStore) Delete(tx engine.Transaction, tableName string) error {
+func (t *tableInfoStore) Delete(tx *Transaction, tableName string) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
-	st, err := tx.GetStore([]byte(tableInfoStoreName))
+	info, ok := t.tableInfos[tableName]
+	if !ok {
+		return ErrTableNotFound
+	}
+
+	if info.transactionID != 0 && info.transactionID != tx.id {
+		return ErrTableNotFound
+	}
+
+	st, err := tx.Tx.GetStore([]byte(tableInfoStoreName))
 	if err != nil {
 		return err
 	}
@@ -244,15 +267,48 @@ func (t *tableInfoStore) loadAllTableInfo(tx engine.Transaction) error {
 	return nil
 }
 
-// ListTables lists all the tables.
+// remove all tableInfo whose transaction id is equal to the given transacrion id.
+// this is called when a read/write transaction is being rolled back.
+func (t *tableInfoStore) rollback(tx *Transaction) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	for k, info := range t.tableInfos {
+		if info.transactionID == tx.id {
+			delete(t.tableInfos, k)
+		}
+	}
+}
+
+// set all the tableInfo created by this transaction to 0.
+// this is called when a read/write transaction is being commited.
+func (t *tableInfoStore) commit(tx *Transaction) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	for k := range t.tableInfos {
+		if t.tableInfos[k].transactionID == tx.id {
+			info := t.tableInfos[k]
+			info.transactionID = 0
+			t.tableInfos[k] = info
+		}
+	}
+}
+
+// ListTables lists all the tables. It ignores tables created by
+// other transactions that haven't been commited yet.
 // The returned slice is lexicographically ordered.
-func (t *tableInfoStore) ListTables() []string {
+func (t *tableInfoStore) ListTables(tx *Transaction) []string {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
 
 	names := make([]string, len(t.tableInfos))
 	var i int
 	for k := range t.tableInfos {
+		if t.tableInfos[k].transactionID != 0 && t.tableInfos[k].transactionID != tx.id {
+			continue
+		}
+
 		names[i] = k
 		i++
 	}
@@ -262,19 +318,6 @@ func (t *tableInfoStore) ListTables() []string {
 	return names
 }
 
-// GetTableInfo returns a copy of all the table information.
-func (t *tableInfoStore) GetTableInfo() map[string]TableInfo {
-	t.mu.RLock()
-	defer t.mu.RUnlock()
-
-	ti := make(map[string]TableInfo, len(t.tableInfos))
-	for k, v := range t.tableInfos {
-		ti[k] = v
-	}
-
-	return ti
-}
-
 func (t *tableInfoStore) generateStoreID() [6]byte {
 	t.mu.RLock()
 	defer t.mu.RUnlock()
diff --git a/database/database.go b/database/database.go
index 7263fc83..04391a40 100644
--- a/database/database.go
+++ b/database/database.go
@@ -3,6 +3,7 @@ package database
 
 import (
 	"sync"
+	"sync/atomic"
 
 	"github.com/genjidb/genji/engine"
 )
@@ -21,6 +22,11 @@ type Database struct {
 	// and is used by every call to table#Insert to generate the
 	// docid if the table doesn't have a primary key.
 	tableDocids map[string]int64
+
+	// This stores the last transaction id created.
+	// It starts at 0 at database startup and is
+	// incremented atomically every time Begin is called.
+	lastTransactionID int64
 }
 
 // New initializes the DB using the given engine.
@@ -84,6 +90,7 @@ func (db *Database) Begin(writable bool) (*Transaction, error) {
 	}
 
 	tx := Transaction{
+		id:             atomic.AddInt64(&db.lastTransactionID, 1),
 		db:             db,
 		Tx:             ntx,
 		writable:       writable,
@@ -97,8 +104,3 @@ func (db *Database) Begin(writable bool) (*Transaction, error) {
 
 	return &tx, nil
 }
-
-// GetTableInfo returns a map containing information about all the tables.
-func (db *Database) GetTableInfo() map[string]TableInfo {
-	return db.tableInfoStore.GetTableInfo()
-}
diff --git a/database/table.go b/database/table.go
index 02d1a3b2..80c487a5 100644
--- a/database/table.go
+++ b/database/table.go
@@ -24,12 +24,7 @@ type Table struct {
 
 // Info of the table.
 func (t *Table) Info() (*TableInfo, error) {
-	ti, err := t.infoStore.Get(t.name)
-	if err != nil {
-		return nil, err
-	}
-
-	return ti, nil
+	return t.infoStore.Get(t.tx, t.name)
 }
 
 // Name returns the name of the table.
@@ -348,7 +343,7 @@ func (t *Table) GetDocument(key []byte) (document.Document, error) {
 // if there are no primary key in the table, a default
 // key is generated, called the docid.
 func (t *Table) generateKey(d document.Document) ([]byte, error) {
-	ti, err := t.infoStore.Get(t.name)
+	ti, err := t.infoStore.Get(t.tx, t.name)
 	if err != nil {
 		return nil, err
 	}
diff --git a/database/transaction.go b/database/transaction.go
index d6b6cc9f..016a6fe6 100644
--- a/database/transaction.go
+++ b/database/transaction.go
@@ -1,7 +1,6 @@
 package database
 
 import (
-	"errors"
 	"fmt"
 
 	"github.com/genjidb/genji/document"
@@ -20,6 +19,7 @@ var (
 // Transaction is either read-only or read/write. Read-only can be used to read tables
 // and read/write can be used to read, create, delete and modify tables.
 type Transaction struct {
+	id       int64
 	db       *Database
 	Tx       engine.Transaction
 	writable bool
@@ -30,11 +30,17 @@ type Transaction struct {
 
 // Rollback the transaction. Can be used safely after commit.
 func (tx *Transaction) Rollback() error {
+	if tx.writable {
+		tx.tableInfoStore.rollback(tx)
+	}
 	return tx.Tx.Rollback()
 }
 
 // Commit the transaction.
 func (tx *Transaction) Commit() error {
+	if tx.writable {
+		tx.tableInfoStore.commit(tx)
+	}
 	return tx.Tx.Commit()
 }
 
@@ -43,36 +49,15 @@ func (tx *Transaction) Writable() bool {
 	return tx.writable
 }
 
-// Promote rollsback a read-only transaction and begins a read-write transaction transparently.
-// It returns an error if the current transaction is already writable.
-func (tx *Transaction) Promote() error {
-	if tx.writable {
-		return errors.New("cannot promote a writable transaction")
-	}
-
-	err := tx.Rollback()
-	if err != nil {
-		return err
-	}
-
-	newTransaction, err := tx.db.Begin(true)
-	if err != nil {
-		return err
-	}
-
-	*tx = *newTransaction
-	return nil
-}
-
 // CreateTable creates a table with the given name.
 // If it already exists, returns ErrTableAlreadyExists.
-func (tx Transaction) CreateTable(name string, info *TableInfo) error {
+func (tx *Transaction) CreateTable(name string, info *TableInfo) error {
 	if info == nil {
 		info = new(TableInfo)
 	}
 
 	info.storeID = tx.tableInfoStore.generateStoreID()
-	err := tx.tableInfoStore.Insert(tx.Tx, name, info)
+	err := tx.tableInfoStore.Insert(tx, name, info)
 	if err != nil {
 		return err
 	}
@@ -86,8 +71,8 @@ func (tx Transaction) CreateTable(name string, info *TableInfo) error {
 }
 
 // GetTable returns a table by name. The table instance is only valid for the lifetime of the transaction.
-func (tx Transaction) GetTable(name string) (*Table, error) {
-	ti, err := tx.tableInfoStore.Get(name)
+func (tx *Transaction) GetTable(name string) (*Table, error) {
+	ti, err := tx.tableInfoStore.Get(tx, name)
 	if err != nil {
 		return nil, err
 	}
@@ -98,7 +83,7 @@ func (tx Transaction) GetTable(name string) (*Table, error) {
 	}
 
 	return &Table{
-		tx:        &tx,
+		tx:        tx,
 		Store:     s,
 		name:      name,
 		infoStore: tx.tableInfoStore,
@@ -107,14 +92,14 @@ func (tx Transaction) GetTable(name string) (*Table, error) {
 
 // RenameTable renames a table.
 // If it doesn't exist, it returns ErrTableNotFound.
-func (tx Transaction) RenameTable(oldName, newName string) error {
-	ti, err := tx.tableInfoStore.Get(oldName)
+func (tx *Transaction) RenameTable(oldName, newName string) error {
+	ti, err := tx.tableInfoStore.Get(tx, oldName)
 	if err != nil {
 		return err
 	}
 
 	// Insert the TableInfo keyed by the newName name.
-	err = tx.tableInfoStore.Insert(tx.Tx, newName, ti)
+	err = tx.tableInfoStore.Insert(tx, newName, ti)
 	if err != nil {
 		return err
 	}
@@ -135,11 +120,11 @@ func (tx Transaction) RenameTable(oldName, newName string) error {
 	}
 
 	// Delete the old reference from the tableInfoStore.
-	return tx.tableInfoStore.Delete(tx.Tx, oldName)
+	return tx.tableInfoStore.Delete(tx, oldName)
 }
 
 // DropTable deletes a table from the database.
-func (tx Transaction) DropTable(name string) error {
+func (tx *Transaction) DropTable(name string) error {
 	it := tx.indexStore.st.NewIterator(engine.IteratorConfig{})
 
 	var buf msgpack.EncodedDocument
@@ -175,12 +160,12 @@ func (tx Transaction) DropTable(name string) error {
 		return err
 	}
 
-	ti, err := tx.tableInfoStore.Get(name)
+	ti, err := tx.tableInfoStore.Get(tx, name)
 	if err != nil {
 		return err
 	}
 
-	err = tx.tableInfoStore.Delete(tx.Tx, name)
+	err = tx.tableInfoStore.Delete(tx, name)
 	if err != nil {
 		return err
 	}
@@ -190,13 +175,13 @@ func (tx Transaction) DropTable(name string) error {
 
 // ListTables lists all the tables.
 // The returned slice is lexicographically ordered.
-func (tx Transaction) ListTables() []string {
-	return tx.tableInfoStore.ListTables()
+func (tx *Transaction) ListTables() []string {
+	return tx.tableInfoStore.ListTables(tx)
 }
 
 // CreateIndex creates an index with the given name.
 // If it already exists, returns ErrIndexAlreadyExists.
-func (tx Transaction) CreateIndex(opts IndexConfig) error {
+func (tx *Transaction) CreateIndex(opts IndexConfig) error {
 	_, err := tx.GetTable(opts.TableName)
 	if err != nil {
 		return err
@@ -206,7 +191,7 @@ func (tx Transaction) CreateIndex(opts IndexConfig) error {
 }
 
 // GetIndex returns an index by name.
-func (tx Transaction) GetIndex(name string) (*Index, error) {
+func (tx *Transaction) GetIndex(name string) (*Index, error) {
 	opts, err := tx.indexStore.Get(name)
 	if err != nil {
 		return nil, err
@@ -226,7 +211,7 @@ func (tx Transaction) GetIndex(name string) (*Index, error) {
 }
 
 // DropIndex deletes an index from the database.
-func (tx Transaction) DropIndex(name string) error {
+func (tx *Transaction) DropIndex(name string) error {
 	opts, err := tx.indexStore.Get(name)
 	if err != nil {
 		return err
@@ -247,12 +232,12 @@ func (tx Transaction) DropIndex(name string) error {
 }
 
 // ListIndexes lists all indexes.
-func (tx Transaction) ListIndexes() ([]*IndexConfig, error) {
+func (tx *Transaction) ListIndexes() ([]*IndexConfig, error) {
 	return tx.indexStore.ListAll()
 }
 
 // ReIndex truncates and recreates selected index from scratch.
-func (tx Transaction) ReIndex(indexName string) error {
+func (tx *Transaction) ReIndex(indexName string) error {
 	idx, err := tx.GetIndex(indexName)
 	if err != nil {
 		return err
@@ -279,7 +264,7 @@ func (tx Transaction) ReIndex(indexName string) error {
 }
 
 // ReIndexAll truncates and recreates all indexes of the database from scratch.
-func (tx Transaction) ReIndexAll() error {
+func (tx *Transaction) ReIndexAll() error {
 	var indexes []string
 
 	it := tx.indexStore.st.NewIterator(engine.IteratorConfig{})
diff --git a/db.go b/db.go
index cade1d92..631a98b5 100644
--- a/db.go
+++ b/db.go
@@ -134,7 +134,7 @@ func (tx *Tx) Query(q string, args ...interface{}) (*query.Result, error) {
 		return nil, err
 	}
 
-	return pq.Exec(tx.Transaction, argsToParams(args), false)
+	return pq.Exec(tx.Transaction, argsToParams(args))
 }
 
 // QueryDocument runs the query and returns the first document.
diff --git a/sql/driver/driver.go b/sql/driver/driver.go
index 5c7d7fb8..26af9fb2 100644
--- a/sql/driver/driver.go
+++ b/sql/driver/driver.go
@@ -82,10 +82,9 @@ func (c *conn) Prepare(q string) (driver.Stmt, error) {
 	}
 
 	return stmt{
-		db:            c.db,
-		tx:            c.tx,
-		q:             pq,
-		nonPromotable: c.nonPromotable,
+		db: c.db,
+		tx: c.tx,
+		q:  pq,
 	}, nil
 }
 
@@ -113,37 +112,35 @@ func (c *conn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, e
 
 	var err error
 
-	// if the ReadOnly flag is explicitly specified, create a non promotable transaction,
-	// otherwise start with a promotable read only transaction.
+	// if the ReadOnly flag is explicitly specified, create a read-only transaction,
+	// otherwise create a read/write transaction.
 	if opts.ReadOnly {
-		c.nonPromotable = true
+		c.tx, err = c.db.Begin(false)
+	} else {
+		c.tx, err = c.db.Begin(true)
 	}
 
-	c.tx, err = c.db.Begin(false)
 	return c, err
 }
 
 func (c *conn) Commit() error {
 	err := c.tx.Commit()
 	c.tx = nil
-	c.nonPromotable = false
 	return err
 }
 
 func (c *conn) Rollback() error {
 	err := c.tx.Rollback()
 	c.tx = nil
-	c.nonPromotable = false
 	return err
 }
 
 // Stmt is a prepared statement. It is bound to a Conn and not
 // used by multiple goroutines concurrently.
 type stmt struct {
-	db            *genji.DB
-	tx            *genji.Tx
-	q             query.Query
-	nonPromotable bool
+	db *genji.DB
+	tx *genji.Tx
+	q  query.Query
 }
 
 // NumInput returns the number of placeholder parameters.
@@ -192,7 +189,7 @@ func (s stmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver
 	// if calling ExecContext within a transaction, use it,
 	// otherwise use DB.
 	if s.tx != nil {
-		res, err = s.q.Exec(s.tx.Transaction, driverNamedValueToParams(args), s.nonPromotable)
+		res, err = s.q.Exec(s.tx.Transaction, driverNamedValueToParams(args))
 	} else {
 		res, err = s.q.Run(s.db.DB, driverNamedValueToParams(args))
 	}
@@ -242,7 +239,7 @@ func (s stmt) QueryContext(ctx context.Context, args []driver.NamedValue) (drive
 	// if calling QueryContext within a transaction, use it,
 	// otherwise use DB.
 	if s.tx != nil {
-		res, err = s.q.Exec(s.tx.Transaction, driverNamedValueToParams(args), s.nonPromotable)
+		res, err = s.q.Exec(s.tx.Transaction, driverNamedValueToParams(args))
 	} else {
 		res, err = s.q.Run(s.db.DB, driverNamedValueToParams(args))
 	}
@@ -343,7 +340,6 @@ func (rs *documentStream) iterate(ctx context.Context) {
 				return nil
 			}
 		}
-
 	})
 
 	if err == errStop || err == nil {
diff --git a/sql/query/query.go b/sql/query/query.go
index f17dfb1a..204bd474 100644
--- a/sql/query/query.go
+++ b/sql/query/query.go
@@ -61,22 +61,12 @@ func (q Query) Run(db *database.Database, args []expr.Param) (*Result, error) {
 	return &res, nil
 }
 
-// Exec the query within the given transaction. If the one of the statements requires a read-write
-// transaction and tx is not, tx will get promoted.
-func (q Query) Exec(tx *database.Transaction, args []expr.Param, forceReadOnly bool) (*Result, error) {
+// Exec the query within the given transaction.
+func (q Query) Exec(tx *database.Transaction, args []expr.Param) (*Result, error) {
 	var res Result
 	var err error
 
 	for _, stmt := range q.Statements {
-		// if the statement requires a writable transaction,
-		// promote the current transaction.
-		if !forceReadOnly && !tx.Writable() && !stmt.IsReadOnly() {
-			err := tx.Promote()
-			if err != nil {
-				return nil, err
-			}
-		}
-
 		res, err = stmt.Run(tx, args)
 		if err != nil {
 			return nil, err


Test patch:
diff --git a/database/config_test.go b/database/config_test.go
index b1eb3dd0..f75210fc 100644
--- a/database/config_test.go
+++ b/database/config_test.go
@@ -9,54 +9,113 @@ import (
 )
 
 func TestTableInfoStore(t *testing.T) {
-	ng := memoryengine.NewEngine()
-	defer ng.Close()
+	t.Run("ok", func(t *testing.T) {
+		ng := memoryengine.NewEngine()
+		defer ng.Close()
 
-	db, err := New(ng)
-	require.NoError(t, err)
-	defer db.Close()
+		db, err := New(ng)
+		require.NoError(t, err)
+		defer db.Close()
 
-	tx, err := db.Begin(true)
-	require.NoError(t, err)
-	defer tx.Rollback()
+		tx, err := db.Begin(true)
+		require.NoError(t, err)
+		defer tx.Rollback()
 
-	info := &TableInfo{
-		FieldConstraints: []FieldConstraint{
-			{Path: []string{"k"}, Type: document.DoubleValue, IsPrimaryKey: true},
-		},
-	}
+		info := &TableInfo{
+			FieldConstraints: []FieldConstraint{
+				{Path: []string{"k"}, Type: document.DoubleValue, IsPrimaryKey: true},
+			},
+		}
 
-	// Inserting one TableInfo should work.
-	err = tx.tableInfoStore.Insert(tx.Tx, "foo1", info)
-	require.NoError(t, err)
+		// Inserting one TableInfo should work.
+		err = tx.tableInfoStore.Insert(tx, "foo1", info)
+		require.NoError(t, err)
 
-	// Inserting an existing TableInfo should not work.
-	err = tx.tableInfoStore.Insert(tx.Tx, "foo1", info)
-	require.Equal(t, err, ErrTableAlreadyExists)
+		// Inserting an existing TableInfo should not work.
+		err = tx.tableInfoStore.Insert(tx, "foo1", info)
+		require.Equal(t, err, ErrTableAlreadyExists)
 
-	// Listing all tables should return their name
-	// lexicographically ordered.
-	_ = tx.tableInfoStore.Insert(tx.Tx, "foo3", info)
-	_ = tx.tableInfoStore.Insert(tx.Tx, "foo2", info)
-	lt := tx.tableInfoStore.ListTables()
-	require.Equal(t, []string{"foo1", "foo2", "foo3"}, lt)
+		// Listing all tables should return their name
+		// lexicographically ordered.
+		_ = tx.tableInfoStore.Insert(tx, "foo3", info)
+		_ = tx.tableInfoStore.Insert(tx, "foo2", info)
+		lt := tx.tableInfoStore.ListTables(tx)
+		require.Equal(t, []string{"foo1", "foo2", "foo3"}, lt)
 
-	// Getting an existing TableInfo should work.
-	received, err := tx.tableInfoStore.Get("foo1")
-	require.NoError(t, err)
-	require.NotNil(t, received.storeID)
+		// Getting an existing TableInfo should work.
+		received, err := tx.tableInfoStore.Get(tx, "foo1")
+		require.NoError(t, err)
+		require.NotNil(t, received.storeID)
 
-	// Getting a non-existing TableInfo should not work.
-	_, err = tx.tableInfoStore.Get("unknown")
-	require.Equal(t, ErrTableNotFound, err)
+		// Getting a non-existing TableInfo should not work.
+		_, err = tx.tableInfoStore.Get(tx, "unknown")
+		require.Equal(t, ErrTableNotFound, err)
 
-	// Deleting an existing TableInfo should work.
-	err = tx.tableInfoStore.Delete(tx.Tx, "foo1")
-	require.NoError(t, err)
+		// Deleting an existing TableInfo should work.
+		err = tx.tableInfoStore.Delete(tx, "foo1")
+		require.NoError(t, err)
+
+		// Deleting a non-existing TableInfo should not work.
+		err = tx.tableInfoStore.Delete(tx, "foo1")
+		require.Equal(t, ErrTableNotFound, err)
+	})
+
+	t.Run("on rollback", func(t *testing.T) {
+		ng := memoryengine.NewEngine()
+		defer ng.Close()
+
+		db, err := New(ng)
+		require.NoError(t, err)
+		defer db.Close()
 
-	// Deleting a non-existing TableInfo should not work.
-	err = tx.tableInfoStore.Delete(tx.Tx, "foo1")
-	require.Equal(t, ErrTableNotFound, err)
+		info := &TableInfo{
+			FieldConstraints: []FieldConstraint{
+				{Path: []string{"k"}, Type: document.DoubleValue, IsPrimaryKey: true},
+			},
+		}
+
+		insertAndRollback := func() {
+			tx, err := db.Begin(true)
+			require.NoError(t, err)
+			err = tx.tableInfoStore.Insert(tx, "foo", info)
+			require.NoError(t, err)
+			err = tx.Rollback()
+			require.NoError(t, err)
+		}
+
+		insertAndRollback()
+		insertAndRollback()
+	})
+
+	t.Run("on commit", func(t *testing.T) {
+		ng := memoryengine.NewEngine()
+		defer ng.Close()
+
+		db, err := New(ng)
+		require.NoError(t, err)
+		defer db.Close()
+
+		info := &TableInfo{
+			FieldConstraints: []FieldConstraint{
+				{Path: []string{"k"}, Type: document.DoubleValue, IsPrimaryKey: true},
+			},
+		}
+
+		insertAndCommit := func() error {
+			tx, err := db.Begin(true)
+			require.NoError(t, err)
+			defer tx.Rollback()
+
+			err = tx.tableInfoStore.Insert(tx, "foo", info)
+			if err != nil {
+				return err
+			}
+			return tx.Commit()
+		}
+
+		require.NoError(t, insertAndCommit())
+		require.Error(t, insertAndCommit())
+	})
 }
 
 func TestIndexStore(t *testing.T) {
diff --git a/database/transaction_test.go b/database/transaction_test.go
index d0975b6b..c91a2e75 100644
--- a/database/transaction_test.go
+++ b/database/transaction_test.go
@@ -41,6 +41,27 @@ func TestTxTable(t *testing.T) {
 		require.EqualError(t, err, database.ErrTableAlreadyExists.Error())
 	})
 
+	t.Run("Create and rollback", func(t *testing.T) {
+		db, err := database.New(memoryengine.NewEngine())
+		require.NoError(t, err)
+		defer db.Close()
+
+		check := func() {
+			tx, err := db.Begin(true)
+			require.NoError(t, err)
+			defer func() {
+				err = tx.Rollback()
+				require.NoError(t, err)
+			}()
+
+			err = tx.CreateTable("test", nil)
+			require.NoError(t, err)
+		}
+
+		check()
+		check()
+	})
+
 	t.Run("Get", func(t *testing.T) {
 		tx, cleanup := newTestDB(t)
 		defer cleanup()
diff --git a/db_test.go b/db_test.go
index 545bbefc..eba49b22 100644
--- a/db_test.go
+++ b/db_test.go
@@ -18,7 +18,7 @@ func ExampleTx() {
 	}
 	defer db.Close()
 
-	tx, err := db.Begin(false)
+	tx, err := db.Begin(true)
 	if err != nil {
 		panic(err)
 	}