重构HBase Update/Delete

lealone-sql/src/main/java/org/lealone/command/Prepared.java

@@ -427,26 +427,6 @@ public Command getCommand() {
         return command;
     }
 
-    protected String[] localRegionNames;
-
-    public String[] getLocalRegionNames() {
-        return localRegionNames;
-    }
-
-    public void setLocalRegionNames(String[] localRegionNames) {
-        this.localRegionNames = localRegionNames;
-    }
-
-    private boolean executeDirec;
-
-    public boolean isExecuteDirec() {
-        return executeDirec;
-    }
-
-    public void setExecuteDirec(boolean executeDirec) {
-        this.executeDirec = executeDirec;
-    }
-
     private boolean local = true;
 
     public boolean isLocal() {

lealone-sql/src/main/java/org/lealone/dbobject/table/Table.java

@@ -1123,7 +1123,7 @@ public boolean supportsAlterColumnWithCopyData() {
         return true;
     }
 
-    public boolean isDistributed() {
+    public boolean supportsSharding() {
         return false;
     }
 

lealone-sql/src/main/java/org/lealone/dbobject/table/TableView.java

@@ -511,14 +511,14 @@ public String getTableName() {
     }
 
     @Override
-    public boolean isDistributed() {
+    public boolean supportsSharding() {
         for (Table t : tables) {
             if (t instanceof TableView) {
-                return ((TableView) t).isDistributed();
+                return ((TableView) t).supportsSharding();
             } else {
-                return t.isDistributed();
+                return t.supportsSharding();
             }
         }
-        return super.isDistributed();
+        return super.supportsSharding();
     }
 }

lealone-sql/src/main/java/org/lealone/engine/MetaRecord.java

@@ -58,7 +58,7 @@ public void execute(Database db, Session systemSession, DatabaseEventListener li
             Prepared command = systemSession.prepare(sql);
             //System.out.println("execute id: " + id + ", sql=" + sql);
             command.setObjectId(id);
-            command.setExecuteDirec(true);
+            command.setLocal(true);
             command.update();
         } catch (DbException e) {
             e = e.addSQL(sql);
@@ -91,6 +91,7 @@ public String getSQL() {
      * @param other the other record
      * @return -1, 0, or 1
      */
+    @Override
     public int compareTo(MetaRecord other) {
         int c1 = getCreateOrder();
         int c2 = other.getCreateOrder();

lealone-storage/cassandra/src/main/java/org/lealone/cassandra/dbobject/table/CassandraTable.java

@@ -110,7 +110,7 @@ public long getRowCountApproximation() {
     }
 
     @Override
-    public boolean isDistributed() {
+    public boolean supportsSharding() {
         return true;
     }
 

lealone-storage/cbase/src/main/java/org/lealone/cbase/dbobject/table/SkipListTable.java

@@ -129,7 +129,7 @@ public long getRowCountApproximation() {
     }
 
     @Override
-    public boolean isDistributed() {
+    public boolean supportsSharding() {
         return true;
     }
 

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/CommandParallel.java

@@ -17,6 +17,13 @@
  */
 package org.lealone.hbase.command;
 
+import static org.lealone.hbase.engine.HBaseConstants.COMMAND_PARALLEL_CORE_POOL_SIZE;
+import static org.lealone.hbase.engine.HBaseConstants.COMMAND_PARALLEL_KEEP_ALIVE_TIME;
+import static org.lealone.hbase.engine.HBaseConstants.COMMAND_PARALLEL_MAX_POOL_SIZE;
+import static org.lealone.hbase.engine.HBaseConstants.DEFAULT_COMMAND_PARALLEL_CORE_POOL_SIZE;
+import static org.lealone.hbase.engine.HBaseConstants.DEFAULT_COMMAND_PARALLEL_KEEP_ALIVE_TIME;
+import static org.lealone.hbase.engine.HBaseConstants.DEFAULT_COMMAND_PARALLEL_MAX_POOL_SIZE;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.Callable;
@@ -42,8 +49,6 @@
 import org.lealone.result.ResultInterface;
 import org.lealone.util.New;
 
-import static org.lealone.hbase.engine.HBaseConstants.*;
-
 public class CommandParallel {
     private final static ThreadPoolExecutor pool = initPool();
 
@@ -76,10 +81,10 @@ public static ResultInterface executeQuery(Session session, SQLRoutingInfo sqlRo
         if (sqlRoutingInfo.localRegions != null) {
             for (String regionName : sqlRoutingInfo.localRegions) {
                 Prepared p = session.prepare(HBaseUtils.getPlanSQL(select), true);
-                p.setExecuteDirec(true);
+                p.setLocal(true);
                 p.setFetchSize(select.getFetchSize());
                 if (p instanceof WithWhereClause) {
-                    ((WithWhereClause) p).getWhereClauseSupport().setRegionName(regionName);
+                    ((WithWhereClause) p).getWhereClauseSupport().setCurrentRegionName(regionName);
                 }
                 commands.add(new CommandWrapper(p));
             }
@@ -96,6 +101,7 @@ public static ResultInterface executeQuery(Session session, SQLRoutingInfo sqlRo
         for (int i = 0; i < size; i++) {
             final CommandInterface c = commands.get(i);
             futures.add(pool.submit(new Callable<ResultInterface>() {
+                @Override
                 public ResultInterface call() throws Exception {
                     return c.executeQuery(maxRows, scrollable);
                 }
@@ -114,7 +120,7 @@ public ResultInterface call() throws Exception {
 
         String newSQL = select.getPlanSQL(true);
         Select newSelect = (Select) session.prepare(newSQL, true);
-        newSelect.setExecuteDirec(true);
+        newSelect.setLocal(true);
 
         return new HBaseMergedResult(results, newSelect, select);
     }
@@ -130,6 +136,7 @@ public static int executeUpdate(List<CommandInterface> commands) {
         for (int i = 0; i < size; i++) {
             final CommandInterface c = commands.get(i);
             futures.add(pool.submit(new Callable<Integer>() {
+                @Override
                 public Integer call() throws Exception {
                     return c.executeUpdate();
                 }
@@ -154,6 +161,7 @@ public static int executeUpdate(SQLRoutingInfo sqlRoutingInfo, Callable<Integer>
         for (int i = 0; i < size - 1; i++) {
             final CommandInterface c = commands.get(i);
             futures.add(pool.submit(new Callable<Integer>() {
+                @Override
                 public Integer call() throws Exception {
                     return c.executeUpdate();
                 }

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/HBaseParser.java

@@ -47,6 +47,7 @@
 import org.lealone.hbase.command.dml.HBaseMerge;
 import org.lealone.hbase.command.dml.HBaseSelect;
 import org.lealone.hbase.command.dml.HBaseUpdate;
+import org.lealone.hbase.command.dml.WithWhereClause;
 import org.lealone.hbase.engine.HBaseConstants;
 import org.lealone.hbase.engine.HBaseDatabase;
 import org.lealone.hbase.engine.HBaseSession;
@@ -75,7 +76,7 @@ protected Prepared parsePrepared(char first) {
     private Prepared parseInTheRegion() {
         String[] regionNames = parseRegionNames();
         Prepared p = parsePrepared();
-        p.setLocalRegionNames(regionNames);
+        ((WithWhereClause) p).setLocalRegionNames(regionNames);
         return p;
     }
 

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/ddl/DefineCommandWrapper.java

@@ -23,11 +23,11 @@
 import org.lealone.command.Command;
 import org.lealone.command.FrontendCommand;
 import org.lealone.command.ddl.DefineCommand;
-import org.lealone.engine.Session;
 import org.lealone.engine.FrontendSession;
+import org.lealone.engine.Session;
 import org.lealone.expression.Parameter;
-import org.lealone.hbase.engine.HBaseSession;
 import org.lealone.hbase.engine.FrontendSessionPool;
+import org.lealone.hbase.engine.HBaseSession;
 import org.lealone.hbase.util.HBaseUtils;
 import org.lealone.message.DbException;
 import org.lealone.result.ResultInterface;
@@ -47,7 +47,7 @@ public DefineCommandWrapper(Session session, DefineCommand dc, String sql) {
 
     @Override
     public int update() {
-        if (isExecuteDirec()) {
+        if (isLocal()) {
             return dc.update();
         } else if (session.isMaster()) {
             try {

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/dml/HBaseDelete.java

@@ -20,39 +20,25 @@
 import org.lealone.command.dml.Delete;
 import org.lealone.engine.Session;
 
-public class HBaseDelete extends Delete implements UpdateOrDelete {
-    private final UpdateOrDeleteSupport updateOrDeleteSupport;
+public class HBaseDelete extends Delete implements WithWhereClause {
+    private final WhereClauseSupport whereClauseSupport = new WhereClauseSupport();
 
     public HBaseDelete(Session session) {
         super(session);
-        updateOrDeleteSupport = new UpdateOrDeleteSupport(session, this);
     }
 
     @Override
-    public void prepare() {
-        super.prepare();
-        if (tableFilter.getTable().isDistributed())
-            updateOrDeleteSupport.postPrepare(tableFilter);
-        else
-            setExecuteDirec(true);
-    }
-
-    @Override
-    public int update() {
-        if (isExecuteDirec())
-            return super.update();
-        else
-            return updateOrDeleteSupport.update();
+    public WhereClauseSupport getWhereClauseSupport() {
+        return whereClauseSupport;
     }
 
     @Override
-    public WhereClauseSupport getWhereClauseSupport() {
-        return updateOrDeleteSupport.getWhereClauseSupport();
+    public String[] getLocalRegionNames() {
+        return whereClauseSupport.getLocalRegionNames();
     }
 
     @Override
-    public int internalUpdate() {
-        return super.update();
+    public void setLocalRegionNames(String[] localRegionNames) {
+        whereClauseSupport.setLocalRegionNames(localRegionNames);
     }
-
 }

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/dml/HBaseInsert.java

@@ -39,15 +39,15 @@ public void setSortedInsertMode(boolean sortedInsertMode) {
     @Override
     public void prepare() {
         super.prepare();
-        if (table.isDistributed())
+        if (table.supportsSharding())
             insertOrMergeSupport.postPrepare(table, query, list, columns, null);
         else
-            setExecuteDirec(true);
+            setLocal(true);
     }
 
     @Override
     public int update() {
-        if (isExecuteDirec())
+        if (isLocal())
             return super.update();
         else
             return insertOrMergeSupport.update(insertFromSelect, sortedInsertMode, this);

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/dml/HBaseMerge.java

@@ -34,15 +34,15 @@ public HBaseMerge(Session session) {
     @Override
     public void prepare() {
         super.prepare();
-        if (table.isDistributed())
+        if (table.supportsSharding())
             insertOrMergeSupport.postPrepare(table, query, list, columns, keys);
         else
-            setExecuteDirec(true);
+            setLocal(true);
     }
 
     @Override
     public int update() {
-        if (isExecuteDirec())
+        if (isLocal())
             return super.update();
         else
             return insertOrMergeSupport.update(false, false, this);

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/dml/HBaseSelect.java

@@ -39,23 +39,24 @@ public HBaseSelect(Session session) {
     @Override
     public void prepare() {
         super.prepare();
-        if (topTableFilter.getTable().isDistributed())
+        if (topTableFilter.getTable().supportsSharding())
             whereClauseSupport.setTableFilter(topTableFilter);
         else
-            setExecuteDirec(true);
+            setLocal(true);
     }
 
     @Override
     public ResultInterface query(int limit, ResultTarget target) {
         boolean addRowToResultTarget = true;
         ResultInterface result;
 
-        if (isExecuteDirec()) {
+        String[] localRegionNames = whereClauseSupport.getLocalRegionNames();
+        if (isLocal()) {
             result = super.query(limit, target);
             addRowToResultTarget = false;
         } else if (localRegionNames != null && localRegionNames.length != 0) {
             if (localRegionNames.length == 1) {
-                whereClauseSupport.setRegionName(localRegionNames[0]);
+                whereClauseSupport.setCurrentRegionName(localRegionNames[0]);
                 result = super.query(limit, target);
                 addRowToResultTarget = false;
             } else {
@@ -71,7 +72,7 @@ public ResultInterface query(int limit, ResultTarget target) {
             }
 
             if (sqlRoutingInfo.localRegion != null) {
-                whereClauseSupport.setRegionName(sqlRoutingInfo.localRegion);
+                whereClauseSupport.setCurrentRegionName(sqlRoutingInfo.localRegion);
                 result = super.query(limit, target);
                 addRowToResultTarget = false;
             } else if (sqlRoutingInfo.remoteCommand != null) {
@@ -95,4 +96,13 @@ public WhereClauseSupport getWhereClauseSupport() {
         return whereClauseSupport;
     }
 
+    @Override
+    public String[] getLocalRegionNames() {
+        return whereClauseSupport.getLocalRegionNames();
+    }
+
+    @Override
+    public void setLocalRegionNames(String[] localRegionNames) {
+        whereClauseSupport.setLocalRegionNames(localRegionNames);
+    }
 }

lealone-storage/hbase/src/main/java/org/lealone/hbase/command/dml/HBaseUpdate.java

@@ -20,39 +20,25 @@
 import org.lealone.command.dml.Update;
 import org.lealone.engine.Session;
 
-public class HBaseUpdate extends Update implements UpdateOrDelete {
-    private final UpdateOrDeleteSupport updateOrDeleteSupport;
+public class HBaseUpdate extends Update implements WithWhereClause {
+    private final WhereClauseSupport whereClauseSupport = new WhereClauseSupport();
 
     public HBaseUpdate(Session session) {
         super(session);
-        updateOrDeleteSupport = new UpdateOrDeleteSupport(session, this);
     }
 
     @Override
-    public void prepare() {
-        super.prepare();
-        if (tableFilter.getTable().isDistributed())
-            updateOrDeleteSupport.postPrepare(tableFilter);
-        else
-            setExecuteDirec(true);
-    }
-
-    @Override
-    public int update() {
-        if (isExecuteDirec())
-            return super.update();
-        else
-            return updateOrDeleteSupport.update();
+    public WhereClauseSupport getWhereClauseSupport() {
+        return whereClauseSupport;
     }
 
     @Override
-    public WhereClauseSupport getWhereClauseSupport() {
-        return updateOrDeleteSupport.getWhereClauseSupport();
+    public String[] getLocalRegionNames() {
+        return whereClauseSupport.getLocalRegionNames();
     }
 
     @Override
-    public int internalUpdate() {
-        return super.update();
+    public void setLocalRegionNames(String[] localRegionNames) {
+        whereClauseSupport.setLocalRegionNames(localRegionNames);
     }
-
 }