adapt Hbase put ls op (#117) (#144)

IHEII · eemjwu · shenyunlong · web-flow · commit 713b35194c57 · 2024-05-16T14:19:52.000+08:00
* habse put ls

* BatchOperationResult add interface to query result info

* fix add rowkey obj type change

* fix typo

* [Adjust] negate hbase timestamp in client side

---------

Co-authored-by: eemjwu &lt;34029771+eemjwu@users.noreply.github.com&gt;
Co-authored-by: shenyunlong.syl &lt;shenyunlong.syl@oceanbase.com&gt;
diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java b/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java
@@ -3169,6 +3169,10 @@ public void setRunningMode(RunningMode runningMode) {
         this.runningMode = runningMode;
     }
 
+    public RunningMode getRunningMode() {
+        return this.runningMode;
+    }
+
     public enum RunningMode {
         NORMAL, HBASE;
     }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/BatchOperation.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/BatchOperation.java
@@ -21,6 +21,8 @@
 import com.alipay.oceanbase.rpc.checkandmutate.CheckAndInsUp;
 import com.alipay.oceanbase.rpc.exception.ObTableException;
 import com.alipay.oceanbase.rpc.mutation.result.BatchOperationResult;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntityType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperationType;
 import com.alipay.oceanbase.rpc.table.ObTableClientLSBatchOpsImpl;
 import com.alipay.oceanbase.rpc.table.api.Table;
@@ -43,6 +45,7 @@ public class BatchOperation {
     boolean              hasGet           = false;
     ObTableOperationType lastType         = ObTableOperationType.INVALID;
     boolean              isSameType       = true;
+    protected ObTableEntityType entityType       = ObTableEntityType.DYNAMIC;
 
     /*
      * default constructor
@@ -138,6 +141,10 @@ public BatchOperation addOperation(CheckAndInsUp... insUps) {
         return this;
     }
 
+    public void setEntityType(ObTableEntityType entityType) {
+        this.entityType = entityType;
+    }
+
     public BatchOperation setIsAtomic(boolean isAtomic) {
         this.isAtomic = isAtomic;
         return this;
@@ -237,6 +244,7 @@ private BatchOperationResult executeWithNormalBatchOp() throws Exception {
                 throw new ObTableException("unknown operation " + operation);
             }
         }
+        batchOps.setEntityType(entityType);
         batchOps.setAtomicOperation(isAtomic);
         batchOps.setReturnOneResult(returnOneResult);
         return new BatchOperationResult(batchOps.executeWithResult());
@@ -265,6 +273,9 @@ private BatchOperationResult executeWithLSBatchOp() throws Exception {
                     }
                 } else if (operation instanceof Mutation) {
                     Mutation mutation = (Mutation) operation;
+                    if (((ObTableClient) client).getRunningMode() == ObTableClient.RunningMode.HBASE) {
+                        negateHbaseTimestamp(mutation);
+                    }
                     batchOps.addOperation(mutation);
                     if (!hasSetRowkeyElement && mutation.getRowKeyNames() != null) {
                         List<String> rowKeyNames = mutation.getRowKeyNames();
@@ -293,6 +304,17 @@ private BatchOperationResult executeWithLSBatchOp() throws Exception {
         batchOps.setReturningAffectedEntity(withResult);
         batchOps.setReturnOneResult(returnOneResult);
         batchOps.setAtomicOperation(isAtomic);
+        batchOps.setEntityType(entityType);
         return new BatchOperationResult(batchOps.executeWithResult());
     }
+
+    private void negateHbaseTimestamp(Mutation mutation) {
+        Object[] rowKey = mutation.getRowKey();
+        if (rowKey == null || rowKey.length != 3) {
+            throw new IllegalArgumentException("hbase rowkey length must be 3");
+        } else {
+            long ts = ((long) ((ObObj) mutation.getRowKey()[2]).getValue());
+            ((ObObj) mutation.getRowKey()[2]).setValue(-ts);
+        }
+    }
 }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/Increment.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/Increment.java
@@ -29,9 +29,7 @@
 import java.util.Map;
 
 public class Increment extends Mutation<Increment> {
-    private List<String> columns;
-    private List<Object> values;
-    boolean              withResult;
+    boolean withResult;
 
     /*
      * default constructor
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/Insert.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/Insert.java
@@ -31,9 +31,6 @@
 import java.util.Map;
 
 public class Insert extends Mutation<Insert> {
-    private List<String> columns = null;
-    private List<Object> values  = null;
-
     /*
      * default constructor
      */
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/InsertOrUpdate.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/InsertOrUpdate.java
@@ -28,9 +28,7 @@
 import java.util.Map;
 
 public class InsertOrUpdate extends Mutation<InsertOrUpdate> {
-    private boolean      usePut;
-    private List<String> columns;
-    private List<Object> values;
+    private boolean usePut;
 
     /*
      * default constructor
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/Mutation.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/Mutation.java
@@ -20,6 +20,12 @@
 import com.alipay.oceanbase.rpc.ObTableClient;
 import com.alipay.oceanbase.rpc.exception.ObTableException;
 import com.alipay.oceanbase.rpc.filter.*;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObjMeta;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObjType;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObITableEntity;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntity;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperation;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableOperationType;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObNewRange;
 import com.alipay.oceanbase.rpc.table.api.Table;
@@ -29,13 +35,18 @@
 import java.util.List;
 import java.util.Map;
 
+import static com.alipay.oceanbase.rpc.mutation.MutationFactory.colVal;
+import static com.alipay.oceanbase.rpc.mutation.MutationFactory.row;
+
 public class Mutation<T> {
     private String         tableName;
     private Table          client;
     protected Object[]     rowKey;
     private TableQuery     query;
     private boolean        hasSetRowKey = false;
     protected List<String> rowKeyNames  = null;
+    protected List<String> columns;
+    protected List<Object> values;
 
     /*
      * default constructor
@@ -47,6 +58,8 @@ public Mutation() {
         rowKey = null;
         query = null;
         rowKeyNames = null;
+        columns = null;
+        values = null;
     }
 
     /*
@@ -63,6 +76,8 @@ public Mutation(Table client, String tableName) {
         this.rowKey = null;
         this.query = null;
         this.rowKeyNames = null;
+        this.columns = null;
+        this.values = null;
     }
 
     /*
@@ -440,4 +455,48 @@ static void removeRowkeyFromMutateColval(List<String> columns, List<Object> valu
             }
         }
     }
+
+    public void addColVal(String propName, ObObj propValue) {
+        columns.add(propName);
+        values.add(propValue);
+    }
+
+    public static Mutation getInstance(ObTableOperationType type, String[] rowKeyNames,
+                                       Object[] rowKeys, String[] columns, Object[] properties) {
+        Mutation mutation = null;
+        switch (type) {
+            case INSERT_OR_UPDATE:
+                mutation = new InsertOrUpdate();
+                break;
+            case DEL:
+                mutation = new Delete();
+                break;
+            default:
+                throw new ObTableException("not support operation type " + type);
+        }
+
+        Row rowKeyRow = new Row();
+        if (rowKeys != null) {
+            for (int i = 0; i < rowKeys.length; i++) {
+                Object rowkey = rowKeys[i];
+                ObObj obj = ObObj.getInstance(rowkey);
+                rowKeyRow.add(rowKeyNames[i], obj);
+            }
+        }
+        mutation.setRowKey(rowKeyRow);
+
+        if (columns != null) {
+            for (int i = 0; i < columns.length; i++) {
+                String name = columns[i];
+                Object value = null;
+                if (properties != null) {
+                    value = properties[i];
+                }
+                ObObj c = ObObj.getInstance(value);
+                mutation.addColVal(name, c);
+            }
+        }
+
+        return mutation;
+    }
 }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/Put.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/Put.java
@@ -31,9 +31,6 @@
 import java.util.Map;
 
 public class Put extends Mutation<Put> {
-    private List<String> columns = null;
-    private List<Object> values  = null;
-
     /*
      * default constructor
      */
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/Replace.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/Replace.java
@@ -28,9 +28,6 @@
 import java.util.Map;
 
 public class Replace extends Mutation<Replace> {
-    private List<String> columns;
-    private List<Object> values;
-
     /*
      * default constructor
      */
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/Update.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/Update.java
@@ -29,9 +29,6 @@
 import java.util.Map;
 
 public class Update extends Mutation<Update> {
-    private List<String> columns;
-    private List<Object> values;
-
     /*
      * default constructor
      */
diff --git a/src/main/java/com/alipay/oceanbase/rpc/mutation/result/BatchOperationResult.java b/src/main/java/com/alipay/oceanbase/rpc/mutation/result/BatchOperationResult.java
@@ -18,6 +18,7 @@
 package com.alipay.oceanbase.rpc.mutation.result;
 
 import com.alipay.oceanbase.rpc.exception.ObTableException;
+import com.alipay.oceanbase.rpc.protocol.payload.ResultCodes;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -26,13 +27,56 @@ public class BatchOperationResult {
 
     private List<Object> results;
 
+    boolean              hasError = false;
+
     /*
      * construct with List of Object
      */
     public BatchOperationResult(List<Object> results) {
         this.results = results;
     }
 
+    public List<Object> getResults() {
+        return results;
+    }
+
+    public List<Integer> getErrorCodeList() {
+        List<Integer> errorCodeList = new ArrayList<Integer>();
+        for (Object item : results) {
+            int errorCode = ResultCodes.OB_SUCCESS.errorCode;
+            if (item instanceof ObTableException) {
+                errorCode = ((ObTableException) item).getErrorCode();
+                hasError = true;
+            }
+            errorCodeList.add(errorCode);
+        }
+        return errorCodeList;
+    }
+
+    public boolean hasError() {
+        if (!hasError) {
+            for (Object item : results) {
+                if (item instanceof ObTableException) {
+                    hasError = true;
+                    break;
+                }
+            }
+        }
+        return hasError;
+    }
+
+    public ObTableException getFirstException() {
+        ObTableException exception = null;
+        for (Object item : results) {
+            if (item instanceof ObTableException) {
+                exception = (ObTableException) item;
+                hasError = true;
+                break;
+            }
+        }
+        return exception;
+    }
+
     /*
      * get result
      */
@@ -59,6 +103,7 @@ public long getWrongCount() {
         for (Object item : results) {
             if (item instanceof ObTableException) {
                 ++wrongCount;
+                hasError = true;
             }
         }
         return wrongCount;
@@ -86,6 +131,7 @@ public int[] getWrongIdx() {
         for (Object item : results) {
             if (item instanceof ObTableException) {
                 wrongIdx.add(i);
+                hasError = true;
             }
             ++i;
         }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableOperation.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableOperation.java
@@ -130,10 +130,7 @@ public static ObTableOperation getInstance(ObTableOperationType type, Object[] r
         if (rowKeys != null) {
             for (int i = 0; i < rowKeys.length; i++) {
                 Object rowkey = rowKeys[i];
-                ObObjMeta rowkeyMeta = ObObjType.defaultObjMeta(rowkey);
-                ObObj obj = new ObObj();
-                obj.setMeta(rowkeyMeta);
-                obj.setValue(rowkey);
+                ObObj obj = ObObj.getInstance(rowkey);
                 entity.addRowKeyValue(obj);
             }
         }
@@ -145,11 +142,7 @@ public static ObTableOperation getInstance(ObTableOperationType type, Object[] r
                 if (properties != null) {
                     value = properties[i];
                 }
-                ObObjMeta meta = ObObjType.defaultObjMeta(value);
-
-                ObObj c = new ObObj();
-                c.setMeta(meta);
-                c.setValue(value);
+                ObObj c = ObObj.getInstance(value);
                 entity.setProperty(name, c);
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -3169,6 +3169,10 @@ public void setRunningMode(RunningMode runningMode) {`
`3169`	`3169`	`this.runningMode = runningMode;`
`3170`	`3170`	`}`
`3171`	`3171`
	`3172`	`+ public RunningMode getRunningMode() {`
	`3173`	`+ return this.runningMode;`
	`3174`	`+ }`
	`3175`	`+`
`3172`	`3176`	`public enum RunningMode {`
`3173`	`3177`	`NORMAL, HBASE;`
`3174`	`3178`	`}`