优化HBase事务实现，使用本地事务状态表替换HTable的方式

lealone · Jan 12, 2015 · f0913a3 · f0913a3
1 parent 58a99f9
commit f0913a3
Show file tree

Hide file tree

Showing 11 changed files with 153 additions and 214 deletions.
diff --git a/lealone-client/src/main/java/org/lealone/engine/FrontendSession.java b/lealone-client/src/main/java/org/lealone/engine/FrontendSession.java
@@ -67,6 +67,8 @@ public class FrontendSession extends SessionWithState implements DataHandler {
     public static final int COMMAND_EXECUTE_DISTRIBUTED_SAVEPOINT_ADD = 104;
     public static final int COMMAND_EXECUTE_DISTRIBUTED_SAVEPOINT_ROLLBACK = 105;
 
+    public static final int COMMAND_EXECUTE_TRANSACTION_VALIDATE = 106;
+
     public static final int COMMAND_EXECUTE_BATCH_UPDATE_STATEMENT = 120;
     public static final int COMMAND_EXECUTE_BATCH_UPDATE_PREPAREDSTATEMENT = 121;
 
@@ -345,7 +347,7 @@ private void connectServer(ConnectionInfo ci) {
     }
 
     //TODO
-    public void handleException(IOException e) {
+    public void handleException(Exception e) {
         checkClosed();
     }
 
@@ -686,6 +688,18 @@ public synchronized void rollbackToSavepoint(String name) {
         }
     }
 
+    public synchronized boolean validateTransaction(String localTransactionName) {
+        checkClosed();
+        try {
+            transfer.writeInt(FrontendSession.COMMAND_EXECUTE_TRANSACTION_VALIDATE).writeString(localTransactionName);
+            done(transfer);
+            return transfer.readBoolean();
+        } catch (Exception e) {
+            handleException(e);
+            return false;
+        }
+    }
+
     public void setTransaction(TransactionInterface transaction) {
         this.transaction = transaction;
     }

diff --git a/lealone-server/src/main/java/org/lealone/server/TcpServerThread.java b/lealone-server/src/main/java/org/lealone/server/TcpServerThread.java
@@ -32,6 +32,7 @@
 import org.lealone.result.ResultColumn;
 import org.lealone.result.ResultInterface;
 import org.lealone.store.LobStorage;
+import org.lealone.transaction.TransactionStatusTable;
 import org.lealone.util.IOUtils;
 import org.lealone.util.New;
 import org.lealone.util.SmallLRUCache;
@@ -477,6 +478,20 @@ private void process() throws IOException {
             transfer.flush();
             break;
         }
+        case FrontendSession.COMMAND_EXECUTE_TRANSACTION_VALIDATE: {
+            int old = session.getModificationId();
+            boolean isValid = TransactionStatusTable.isValid(transfer.readString());
+            int status;
+            if (session.isClosed()) {
+                status = FrontendSession.STATUS_CLOSED;
+            } else {
+                status = getState(old);
+            }
+            transfer.writeInt(status);
+            transfer.writeBoolean(isValid);
+            transfer.flush();
+            break;
+        }
         case FrontendSession.COMMAND_EXECUTE_BATCH_UPDATE_STATEMENT: {
             int size = transfer.readInt();
             ArrayList<String> batchCommands = New.arrayList(size);

diff --git a/lealone-storage/hbase/src/main/java/org/lealone/hbase/command/dml/HBaseSelect.java b/lealone-storage/hbase/src/main/java/org/lealone/hbase/command/dml/HBaseSelect.java
@@ -51,10 +51,7 @@ public ResultInterface query(int limit, ResultTarget target) {
         ResultInterface result;
 
         String[] localRegionNames = whereClauseSupport.getLocalRegionNames();
-        if (isLocal()) {
+        if (localRegionNames != null && localRegionNames.length != 0) {
-            result = super.query(limit, target);
-            addRowToResultTarget = false;
-        } else if (localRegionNames != null && localRegionNames.length != 0) {
             if (localRegionNames.length == 1) {
                 whereClauseSupport.setCurrentRegionName(localRegionNames[0]);
                 result = super.query(limit, target);
@@ -64,6 +61,9 @@ public ResultInterface query(int limit, ResultTarget target) {
                 sqlRoutingInfo.localRegions = Arrays.asList(localRegionNames);
                 result = CommandParallel.executeQuery(session, sqlRoutingInfo, this, limit, false);
             }
+        } else if (isLocal()) {
+            result = super.query(limit, target);
+            addRowToResultTarget = false;
         } else {
             try {
                 sqlRoutingInfo = HBaseUtils.getSQLRoutingInfo((HBaseSession) session, whereClauseSupport, this);

diff --git a/lealone-storage/hbase/src/main/java/org/lealone/hbase/dbobject/table/HBaseTable.java b/lealone-storage/hbase/src/main/java/org/lealone/hbase/dbobject/table/HBaseTable.java
@@ -55,6 +55,7 @@
 import org.lealone.message.DbException;
 import org.lealone.result.Row;
 import org.lealone.result.RowList;
+import org.lealone.transaction.TransactionInterface;
 import org.lealone.util.New;
 import org.lealone.util.StatementBuilder;
 import org.lealone.value.Value;
@@ -790,4 +791,9 @@ public String getPrimaryKeyName() {
 
         return primaryKeyName;
     }
+
+    @Override
+    public TransactionInterface getTransaction(Session session) {
+        return session.getTransaction();
+    }
 }
diff --git a/lealone-storage/hbase/src/main/java/org/lealone/hbase/engine/HBaseSession.java b/lealone-storage/hbase/src/main/java/org/lealone/hbase/engine/HBaseSession.java
@@ -30,9 +30,7 @@
 import org.lealone.hbase.dbobject.HBaseSequence;
 import org.lealone.hbase.result.HBaseRow;
 import org.lealone.hbase.transaction.HBaseTransaction;
-import org.lealone.message.DbException;
 import org.lealone.result.Row;
-import org.lealone.transaction.TransactionInterface;
 
 public class HBaseSession extends Session {
 
@@ -46,8 +44,6 @@ public class HBaseSession extends Session {
      */
     private HRegionServer regionServer;
 
-    private volatile HBaseTransaction transaction;
-
     public HBaseSession(Database database, User user, int id) {
         super(database, user, id);
     }
@@ -110,23 +106,14 @@ public String getHostAndPort() {
     }
 
     public void log(HBaseRow row) {
-        if (transaction == null)
+        this.getTransaction().log(row);
-            throw DbException.throwInternalError();
-        transaction.log(row);
     }
 
     @Override
     public HBaseTransaction getTransaction() {
-        if (transaction == null) {
+        if (super.getTransaction() == null) {
-            transaction = new HBaseTransaction(this);
+            super.setTransaction(new HBaseTransaction(this));
-            super.setTransaction(transaction);
         }
-        return transaction;
+        return (HBaseTransaction) super.getTransaction();
-    }
-
-    @Override
-    public void setTransaction(TransactionInterface t) {
-        transaction = (HBaseTransaction) t;
-        super.setTransaction(t);
     }
 }
diff --git a/lealone-storage/hbase/src/main/java/org/lealone/hbase/metadata/TransactionStatusTable.java b/lealone-storage/hbase/src/main/java/org/lealone/hbase/metadata/TransactionStatusTable.java
diff --git a/lealone-storage/hbase/src/main/java/org/lealone/hbase/transaction/HBaseTransaction.java b/lealone-storage/hbase/src/main/java/org/lealone/hbase/transaction/HBaseTransaction.java
@@ -24,6 +24,7 @@
 import org.lealone.hbase.result.HBaseRow;
 import org.lealone.hbase.util.HBaseUtils;
 import org.lealone.transaction.GlobalTransaction;
+import org.lealone.transaction.TransactionManager;
 import org.lealone.value.Value;
 
 public class HBaseTransaction extends GlobalTransaction {
@@ -32,7 +33,7 @@ public class HBaseTransaction extends GlobalTransaction {
 
     public HBaseTransaction(Session session) {
         super(session);
-        String hostAndPort = session.getHostAndPort();
+        String hostAndPort = TransactionManager.getHostAndPort();
 
         transactionMetaAdd = Bytes.toBytes(hostAndPort + "," + transactionId + "," + HBaseConstants.Tag.ADD);
         transactionMetaDelete = Bytes.toBytes(hostAndPort + "," + transactionId + "," + HBaseConstants.Tag.DELETE);

diff --git a/lealone-storage/hbase/src/main/java/org/lealone/hbase/transaction/ValidityChecker.java b/lealone-storage/hbase/src/main/java/org/lealone/hbase/transaction/ValidityChecker.java
@@ -29,7 +29,7 @@
 import org.apache.hadoop.hbase.util.Bytes;
 import org.lealone.hbase.engine.HBaseConstants;
 import org.lealone.hbase.engine.HBaseSession;
-import org.lealone.hbase.metadata.TransactionStatusTable;
+import org.lealone.transaction.TransactionStatusTable;
 import org.lealone.util.StringUtils;
 
 /**
@@ -38,7 +38,6 @@
  *
  */
 public class ValidityChecker {
-    private final static TransactionStatusTable transactionStatusTable = TransactionStatusTable.getInstance();
 
     public static Result checkResult(byte[] defaultColumnFamilyName, HBaseSession session, HRegionServer regionServer,
             byte[] regionName, HBaseTransaction t, Result r) throws IOException {
@@ -68,7 +67,7 @@ public static Result checkResult(byte[] defaultColumnFamilyName, HBaseSession se
             return r;
         }
 
-        if ((oldTid % 2 == 0) || !transactionStatusTable.isValid(hostAndPort, oldTid, t)) {
+        if ((oldTid % 2 == 0) || !TransactionStatusTable.isValid(session, hostAndPort, oldTid, t)) {
             Get get = new Get(r.getRow());
             get.setMaxVersions(1);
             get.setTimeRange(0, oldTid - 1);