Merge lp:~mmcm/akiban-server/pg-periodically-commit into lp:~akiban-technologies/akiban-server/trunk

[Mike McMahon]

Add a mode where explicit transaction commits from time to time.

This is useful for initial data loads from a program that believes that it is doing all the loading with auto-commit off but failing due to the 5 second FDB limitation.

It is controlled by a connection property. It might be cleaner to have it be an extension to SET SESSION CHARACTERISTICS AS TRANSACTION. However, a property has the advantage that is can be set by another process, thereby avoiding the need to alter the application.

The check is done at the end of each statement, where auto-commit would be done. This means that it really only works for a series of isolated statements, like a bunch of INSERTs. It does not work, in particular, for a large DELETE or UPDATE. These are problematic because there is a scan, too, so the underlying transaction cannot be commited invisibly. I am open to suggestions for how to extend / replace this to handle those cases.

The implementation for the default store (Persistit) in this branch is empty. An FDB branch will actually do something.

[Nathan Williams (nwilliams)]

Looks good.

review: Approve

=== modified file 'src/main/java/com/akiban/server/service/transaction/TransactionService.java'
--- src/main/java/com/akiban/server/service/transaction/TransactionService.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/transaction/TransactionService.java	2013-08-12 23:49:27 +0000
@@ -84,6 +84,9 @@
      */
     int incrementTransactionStep(Session session);
 
+    /** Commit the transaction if this is a good time. */
+    void periodicallyCommit(Session session);
+
     /** Add a callback to transaction. */
     void addCallback(Session session, CallbackType type, Callback callback);
 
 
=== modified file 'src/main/java/com/akiban/server/store/PersistitTransactionService.java'
--- src/main/java/com/akiban/server/store/PersistitTransactionService.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/store/PersistitTransactionService.java	2013-08-12 23:49:27 +0000
@@ -159,6 +159,11 @@
     }
 
     @Override
+    public void periodicallyCommit(Session session) {
+        // Persistit can mostly manage with a long-running transaction.
+    }
+
+    @Override
     public void addCallback(Session session, CallbackType type, Callback callback) {
         session.push(getCallbackKey(type), callback);
     }
 
=== modified file 'src/main/java/com/akiban/sql/embedded/JDBCConnection.java'
--- src/main/java/com/akiban/sql/embedded/JDBCConnection.java	2013-06-11 16:30:17 +0000
+++ src/main/java/com/akiban/sql/embedded/JDBCConnection.java	2013-08-12 23:49:27 +0000
@@ -143,7 +143,7 @@
             }
             sessionMonitor.enterStage(MonitorStage.OPTIMIZE);
             if (transaction == null)
-                localTransaction = new ServerTransaction(this, true);
+                localTransaction = new ServerTransaction(this, true, false);
             if ((sqlStmt instanceof DMLStatementNode) && 
                 !(sqlStmt instanceof CallStatementNode))
                 return compiler.compileExecutableStatement((DMLStatementNode)sqlStmt, parser.getParameterList(), getParameterNames, autoGeneratedKeys, context);
@@ -182,7 +182,7 @@
             }
             sessionMonitor.enterStage(MonitorStage.OPTIMIZE);
             if (transaction == null)
-                localTransaction = new ServerTransaction(this, true);
+                localTransaction = new ServerTransaction(this, true, false);
             ExplainPlanContext context = new ExplainPlanContext(compiler, reqs.serviceManager(), session);
             Explainable explainable;
             if ((sqlStmt instanceof DMLStatementNode) && 
 
=== modified file 'src/main/java/com/akiban/sql/pg/PostgresSessionStatement.java'
--- src/main/java/com/akiban/sql/pg/PostgresSessionStatement.java	2013-07-06 17:46:02 +0000
+++ src/main/java/com/akiban/sql/pg/PostgresSessionStatement.java	2013-08-12 23:49:27 +0000
@@ -53,7 +53,7 @@
       "client_encoding", "DateStyle", "geqo", "ksqo",
       "queryTimeoutSec", "zeroDateTimeBehavior", "maxNotificationLevel", "OutputFormat",
       "parserInfixBit", "parserInfixLogical", "parserDoubleQuoted",
-      "newtypes"
+      "newtypes", "transactionPeriodicallyCommit"
     };
 
     private Operation operation;
 
=== modified file 'src/main/java/com/akiban/sql/server/ServerSession.java'
--- src/main/java/com/akiban/sql/server/ServerSession.java	2013-07-13 20:04:24 +0000
+++ src/main/java/com/akiban/sql/server/ServerSession.java	2013-08-12 23:49:27 +0000
@@ -131,6 +131,9 @@
     /** Set following transaction to read-only / read-write. */
     public void setTransactionDefaultReadOnly(boolean readOnly);
 
+    /** Set following transaction to commit as determined by store. */
+    public void setTransactionPeriodicallyCommit(boolean periodicallyCommit);
+
     /** Get the functions registry. */
     public FunctionsRegistry functionsRegistry();
 
 
=== modified file 'src/main/java/com/akiban/sql/server/ServerSessionBase.java'
--- src/main/java/com/akiban/sql/server/ServerSessionBase.java	2013-07-13 20:04:24 +0000
+++ src/main/java/com/akiban/sql/server/ServerSessionBase.java	2013-08-12 23:49:27 +0000
@@ -60,6 +60,7 @@
         new HashMap<>();
     protected ServerTransaction transaction;
     protected boolean transactionDefaultReadOnly = false;
+    protected boolean transactionPeriodicallyCommit = false;
     protected ServerSessionMonitor sessionMonitor;
 
     protected Long queryTimeoutMilli = null;
@@ -124,6 +125,12 @@
                 queryTimeoutMilli = (long)(Double.parseDouble(value) * 1000);
             return true;
         }
+        if ("transactionPeriodicallyCommit".equals(key)) {
+            boolean periodicallyCommit = (value != null) && Boolean.parseBoolean(value);
+            transactionPeriodicallyCommit = periodicallyCommit;
+            if (transaction != null)
+                transaction.setPeriodicallyCommit(periodicallyCommit);
+        }
         return false;
     }
 
@@ -210,7 +217,7 @@
     public void beginTransaction() {
         if (transaction != null)
             throw new TransactionInProgressException();
-        transaction = new ServerTransaction(this, transactionDefaultReadOnly);
+        transaction = new ServerTransaction(this, transactionDefaultReadOnly, transactionPeriodicallyCommit);
     }
 
     @Override
@@ -250,6 +257,11 @@
     }
 
     @Override
+    public void setTransactionPeriodicallyCommit(boolean periodicallyCommit) {
+        this.transactionPeriodicallyCommit = periodicallyCommit;
+    }
+
+    @Override
     public FunctionsRegistry functionsRegistry() {
         return reqs.functionsRegistry();
     }
@@ -331,14 +343,14 @@
                 throw new NoTransactionInProgressException();
             case READ:
             case NEW:
-                localTransaction = new ServerTransaction(this, true);
+                localTransaction = new ServerTransaction(this, true, false);
                 break;
             case WRITE:
             case NEW_WRITE:
             case WRITE_STEP_ISOLATED:
                 if (transactionDefaultReadOnly)
                     throw new TransactionReadOnlyException();
-                localTransaction = new ServerTransaction(this, false);
+                localTransaction = new ServerTransaction(this, false, false);
                 localTransaction.beforeUpdate(true);
                 break;
             }
@@ -369,17 +381,18 @@
             else
                 localTransaction.abort();
         }
-        else {
+        else if (transaction != null) {
             // Make changes visible in open global transaction.
             ServerStatement.TransactionMode transactionMode = stmt.getTransactionMode();
             switch (transactionMode) {
             case REQUIRED_WRITE:
             case WRITE:
             case WRITE_STEP_ISOLATED:
-                if (transaction != null)
-                    transaction.afterUpdate(transactionMode == ServerStatement.TransactionMode.WRITE_STEP_ISOLATED);
+                transaction.afterUpdate(transactionMode == ServerStatement.TransactionMode.WRITE_STEP_ISOLATED);
                 break;
             }
+            // Give periodic commit a chance if enabled.
+            transaction.checkPeriodicallyCommit();
         }
     }
 
 
=== modified file 'src/main/java/com/akiban/sql/server/ServerTransaction.java'
--- src/main/java/com/akiban/sql/server/ServerTransaction.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/sql/server/ServerTransaction.java	2013-08-12 23:49:27 +0000
@@ -29,14 +29,16 @@
 {
     private final Session session;
     private final TransactionService txnService;
-    private boolean readOnly;
+    private boolean readOnly, periodicallyCommit;
     private Date transactionTime;
     
     /** Begin a new transaction or signal an exception. */
-    public ServerTransaction(ServerSession server, boolean readOnly) {
+    public ServerTransaction(ServerSession server, 
+                             boolean readOnly, boolean periodicallyCommit) {
         this.session = server.getSession();
         this.txnService = server.getTransactionService();
         this.readOnly = readOnly;
+        this.periodicallyCommit = periodicallyCommit;
         txnService.beginTransaction(session);
     }
 
@@ -48,6 +50,14 @@
         this.readOnly = readOnly;
     }
 
+    public boolean isPeriodicallyCommit() {
+        return periodicallyCommit;
+    }
+
+    public void setPeriodicallyCommit(boolean periodicallyCommit) {
+        this.periodicallyCommit = periodicallyCommit;
+    }
+
     public void checkTransactionMode(ServerStatement.TransactionMode transactionMode) {
         switch (transactionMode) {
         case NONE:
@@ -95,6 +105,12 @@
         return txnService.isRollbackPending(session);
     }
 
+    public void checkPeriodicallyCommit() {
+        if (periodicallyCommit) {
+            txnService.periodicallyCommit(session);
+        }
+    }
+
     /** Return the transaction's time, which is fixed the first time
      * something asks for it. */
     public Date getTime(ServerSession server) {