Merge branch 'master' into secondary_part-merge-master

Author: maochongxin

src/main/java/com/alipay/oceanbase/hbase/OHTable.java

@@ -26,6 +26,7 @@
 import com.google.protobuf.Service;
 import com.google.protobuf.ServiceException;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.*;
@@ -67,7 +68,7 @@ public class OHTablePool implements Closeable {
      * Default Constructor. Default HBaseConfiguration and no limit on pool size.
      */
     public OHTablePool() {
-        this(new Configuration(), Integer.MAX_VALUE);
+        this(HBaseConfiguration.create(), Integer.MAX_VALUE);
     }
 
     /**

src/main/java/com/alipay/oceanbase/hbase/OHTablePool.java

@@ -143,10 +143,6 @@ public final class OHConstants {
 
     public static final String   MAX_KEYVALUE_SIZE_KEY                       = "hbase.client.keyvalue.maxsize";
 
-    public static final String   HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK         = "hbase.htable.put.write.buffer.check";
-
-    public static final int      DEFAULT_HBASE_HTABLE_PUT_WRITE_BUFFER_CHECK = 10;
-
     public static final long     WRITE_BUFFER_SIZE_DEFAULT                   = 2097152L;
 
     public static final int      MAX_KEYVALUE_SIZE_DEFAULT                   = -1;

src/main/java/com/alipay/oceanbase/hbase/constants/OHConstants.java

@@ -1,3 +1,20 @@
+/*-
+ * #%L
+ * com.oceanbase:obkv-hbase-client
+ * %%
+ * Copyright (C) 2022 - 2025 OceanBase Group
+ * %%
+ * OBKV HBase Client Framework  is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *          http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * #L%
+ */
+
 package com.alipay.oceanbase.hbase.util;
 
 import org.apache.hadoop.hbase.ServerName;

src/main/java/com/alipay/oceanbase/hbase/util/BatchError.java

@@ -35,10 +35,6 @@ public static byte[][] extractFamilyFromQualifier(byte[] qualifier) throws Excep
             throw new RuntimeException("Cannot get family name");
         }
         byte[] family = Arrays.copyOfRange(qualifier, 0, familyLen);
-        int qualifierLen = qualifier.length - familyLen - 1;
-        if (qualifierLen <= 0) {
-            throw new RuntimeException("Cannot get qualifier name");
-        }
         byte[] newQualifier = Arrays.copyOfRange(qualifier, familyLen + 1, qualifier.length);
         return new byte[][] { family, newQualifier };
     }

src/main/java/com/alipay/oceanbase/hbase/util/OHBaseFuncUtils.java

@@ -18,9 +18,11 @@
 package com.alipay.oceanbase.hbase.util;
 
 import com.alipay.oceanbase.hbase.OHTable;
-import com.google.common.annotations.VisibleForTesting;
+import com.alipay.oceanbase.rpc.exception.ObTableUnexpectedException;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableBatchOperation;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.*;
 import org.slf4j.Logger;
@@ -31,7 +33,9 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
-import static com.alipay.oceanbase.rpc.util.TableClientLoggerFactory.LCD;
+
+import static com.alipay.oceanbase.hbase.util.TableHBaseLoggerFactory.LCD;
+import static com.alipay.oceanbase.rpc.ObGlobal.*;
 
 @InterfaceAudience.Private
 public class OHBufferedMutatorImpl implements BufferedMutator {
@@ -40,13 +44,11 @@ public class OHBufferedMutatorImpl implements BufferedMutator {
 
     private final ExceptionListener       listener;
 
-    private final OHTable                 ohTable;
     private final TableName               tableName;
     private volatile Configuration        conf;
 
-    @VisibleForTesting
+    private OHTable                       ohTable;
     final ConcurrentLinkedQueue<Mutation> asyncWriteBuffer       = new ConcurrentLinkedQueue<Mutation>();
-    @VisibleForTesting
     AtomicLong                            currentAsyncBufferSize = new AtomicLong(0);
 
     private long                          writeBufferSize;
@@ -55,9 +57,11 @@ public class OHBufferedMutatorImpl implements BufferedMutator {
     private final ExecutorService         pool;
     private int                           rpcTimeout;
     private int                           operationTimeout;
+    private static final long             OB_VERSION_4_2_5_1     = calcVersion(4, (short) 2,
+                                                                     (byte) 5, (byte) 1);
 
-    public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParams params)
-                                                                                             throws IOException {
+    public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParams params,
+                                 OHTable ohTable) throws IOException {
         if (ohConnection == null || ohConnection.isClosed()) {
             throw new IllegalArgumentException("Connection is null or closed.");
         }
@@ -77,7 +81,38 @@ public OHBufferedMutatorImpl(OHConnectionImpl ohConnection, BufferedMutatorParam
             .getMaxKeyValueSize() : connectionConfig.getMaxKeyValueSize();
 
         // create an OHTable object to do batch work
-        this.ohTable = new OHTable(tableName, ohConnection, connectionConfig, pool);
+        if (ohTable != null) {
+            this.ohTable = ohTable;
+        } else {
+            this.ohTable = new OHTable(tableName, ohConnection, connectionConfig, pool);
+        }
+    }
+
+    /**
+     * only used for OHTable get bufferedMutator
+     * */
+    public OHBufferedMutatorImpl(Configuration conf, BufferedMutatorParams params, OHTable ohTable)
+                                                                                                   throws IOException {
+        // create an OHTable object to do batch work
+        if (ohTable == null) {
+            throw new ObTableUnexpectedException("The ohTable is null.");
+        }
+        this.ohTable = ohTable;
+        // init params in OHBufferedMutatorImpl
+        this.tableName = params.getTableName();
+        this.conf = conf;
+        this.listener = params.getListener();
+
+        OHConnectionConfiguration connectionConfig = new OHConnectionConfiguration(conf);
+        this.pool = params.getPool();
+        this.rpcTimeout = connectionConfig.getRpcTimeout();
+        this.operationTimeout = connectionConfig.getOperationTimeout();
+
+        this.writeBufferSize = params.getWriteBufferSize() != OHConnectionImpl.BUFFERED_PARAM_UNSET ? params
+            .getWriteBufferSize() : connectionConfig.getWriteBufferSize();
+        this.maxKeyValueSize = params.getMaxKeyValueSize() != OHConnectionImpl.BUFFERED_PARAM_UNSET ? params
+            .getMaxKeyValueSize() : connectionConfig.getMaxKeyValueSize();
+
     }
 
     @Override
@@ -119,14 +154,12 @@ public void mutate(List<? extends Mutation> mutations) throws IOException {
             validateOperation(m);
             toAddSize += m.heapSize();
         }
-
         currentAsyncBufferSize.addAndGet(toAddSize);
         asyncWriteBuffer.addAll(mutations);
 
         if (currentAsyncBufferSize.get() > writeBufferSize) {
-            execute(false);
+            batchExecute(false);
         }
-
     }
 
     /**
@@ -142,10 +175,18 @@ private void validateOperation(Mutation mt) throws IllegalArgumentException {
         }
         if (mt instanceof Put) {
             // family empty check is in validatePut
-            HTable.validatePut((Put) mt, maxKeyValueSize);
-            OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), true);
+            OHTable.validatePut((Put) mt, maxKeyValueSize);
+            if (isMultiFamilySupport()) {
+                OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), true);
+            } else {
+                OHTable.checkFamilyViolationForOneFamily(mt.getFamilyMap().keySet());
+            }
         } else {
-            OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), false);
+            if (isMultiFamilySupport()) {
+                OHTable.checkFamilyViolation(mt.getFamilyMap().keySet(), false);
+            } else {
+                OHTable.checkFamilyViolationForOneFamily(mt.getFamilyMap().keySet());
+            }
         }
     }
 
@@ -156,7 +197,7 @@ private void validateOperation(Mutation mt) throws IllegalArgumentException {
      * @param flushAll - if true, sends all the writes and wait for all of them to finish before
      *        returning.
      */
-    private void execute(boolean flushAll) throws IOException {
+    private void batchExecute(boolean flushAll) throws IOException {
         LinkedList<Mutation> execBuffer = new LinkedList<>();
         long dequeuedSize = 0L;
         try {
@@ -172,19 +213,16 @@ private void execute(boolean flushAll) throws IOException {
             if (execBuffer.isEmpty()) {
                 return;
             }
-            ohTable.batch(execBuffer);
+            Object[] results = new Object[execBuffer.size()];
+            ohTable.batch(execBuffer, results);
             // if commit all successfully, clean execBuffer
             execBuffer.clear();
         } catch (Exception ex) {
-            LOGGER.error(LCD.convert("01-00026"), ex);
+            // do not recollect error operations, notify outside
+            LOGGER.error("error happens: table name = ", tableName.getNameAsString(), ex);
             if (ex.getCause() instanceof RetriesExhaustedWithDetailsException) {
-                LOGGER.error(tableName + ": One or more of the operations have failed after retries.");
+                LOGGER.error(tableName.getNameAsString() + ": One or more of the operations have failed after retries.", ex);
                 RetriesExhaustedWithDetailsException retryException = (RetriesExhaustedWithDetailsException) ex.getCause();
-                // recollect mutations
-                execBuffer.clear();
-                for (int i = 0; i < retryException.getNumExceptions(); ++i) {
-                    execBuffer.add((Mutation) retryException.getRow(i));
-                }
                 if (listener != null) {
                     listener.onException(retryException, this);
                 } else {
@@ -194,12 +232,6 @@ private void execute(boolean flushAll) throws IOException {
                 LOGGER.error("Errors unrelated to operations occur during mutation operation", ex);
                 throw ex;
             }
-        } finally {
-            for (Mutation mutation : execBuffer) {
-                long size = mutation.heapSize();
-                currentAsyncBufferSize.addAndGet(size);
-                asyncWriteBuffer.add(mutation);
-            }
         }
     }
 
@@ -209,7 +241,7 @@ public void close() throws IOException {
             return;
         }
         try {
-            execute(true);
+            batchExecute(true);
         } finally {
             // the pool in ObTableClient will be shut down too
             this.pool.shutdown();
@@ -234,13 +266,21 @@ public void setWriteBufferSize(long writeBufferSize) throws IOException {
         }
     }
 
+    /**
+     * Only 4_2_5 BP1 - 4_3_0 and after 4_3_4 support multi-cf
+     * */
+    boolean isMultiFamilySupport() {
+        return (OB_VERSION >= OB_VERSION_4_2_5_1 && OB_VERSION < OB_VERSION_4_3_0_0)
+               || (OB_VERSION >= OB_VERSION_4_3_4_0);
+    }
+
     /**
      * Force to commit all operations
      * do not care whether the pool is shut down or this BufferedMutator is closed
      */
     @Override
     public void flush() throws IOException {
-        execute(true);
+        batchExecute(true);
     }
 
     @Override
@@ -258,6 +298,10 @@ public void setOperationTimeout(int operationTimeout) {
         this.ohTable.setOperationTimeout(operationTimeout);
     }
 
+    public long getCurrentBufferSize() {
+        return currentAsyncBufferSize.get();
+    }
+
     @Deprecated
     public List<Row> getWriteBuffer() {
         return Arrays.asList(asyncWriteBuffer.toArray(new Row[0]));

src/main/java/com/alipay/oceanbase/hbase/util/OHBufferedMutatorImpl.java

@@ -126,6 +126,11 @@ public BufferedMutator getBufferedMutator(TableName tableName) throws IOExceptio
 
     @Override
     public BufferedMutator getBufferedMutator(BufferedMutatorParams params) throws IOException {
+        return getBufferedMutator(params, null);
+    }
+
+    public BufferedMutator getBufferedMutator(BufferedMutatorParams params, OHTable ohTable)
+                                                                                            throws IOException {
         if (params.getTableName() == null) {
             throw new IllegalArgumentException("TableName cannot be null.");
         }
@@ -138,7 +143,7 @@ public BufferedMutator getBufferedMutator(BufferedMutatorParams params) throws I
         if (params.getMaxKeyValueSize() == BUFFERED_PARAM_UNSET) {
             params.maxKeyValueSize(connectionConfig.getMaxKeyValueSize());
         }
-        return new OHBufferedMutatorImpl(this, params);
+        return new OHBufferedMutatorImpl(this, params, ohTable);
     }
 
     @Override

src/main/java/com/alipay/oceanbase/hbase/util/OHConnectionImpl.java

@@ -21,6 +21,7 @@
 import com.alipay.oceanbase.hbase.OHTablePool;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.client.HTableFactory;
 import org.apache.hadoop.hbase.client.HTableInterface;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -170,7 +171,7 @@ private Configuration adjustConfiguration(Configuration configuration, String ta
     }
 
     private Configuration copyConfiguration(Configuration origin) {
-        Configuration copy = new Configuration();
+        Configuration copy = HBaseConfiguration.create();
 
         for (Map.Entry<String, String> entry : origin) {
             copy.set(entry.getKey(), entry.getValue());

src/main/java/com/alipay/oceanbase/hbase/util/OHTableFactory.java

@@ -1522,6 +1522,38 @@ public void testMultiColumnFamilyGet() throws Exception {
             }
         }
         assertEquals(5, keyValues.length);
+
+        // test multi-cf get checkExistenceOnly
+        Delete delKey1 = new Delete(toBytes("Key1"));
+        multiCfHTable.delete(delKey1);
+        get = new Get(toBytes("Key1"));
+        Result allKey1Result = multiCfHTable.get(get);
+        Assert.assertEquals(0, allKey1Result.size());
+
+        boolean exist = multiCfHTable.exists(get);
+        assertFalse(exist);
+        Put put = new Put(toBytes("Key1"));
+        put.add(family1, family1_column1, family1_value);
+        multiCfHTable.put(put);
+        exist = multiCfHTable.exists(get);
+        assertTrue(exist);
+
+        // test existsAll
+        multiCfHTable.delete(delKey1);
+        allKey1Result = multiCfHTable.get(get);
+        Assert.assertEquals(0, allKey1Result.size());
+
+        get = new Get(toBytes("Key0"));
+        get2 = new Get(toBytes("Key1"));
+        get3 = new Get(toBytes("Key2"));
+        ArrayList<Get> gets = new ArrayList<>();
+        gets.add(get);
+        gets.add(get2);
+        gets.add(get3);
+        boolean[] exists = multiCfHTable.existsAll(gets);
+        Assert.assertTrue(exists[0]);
+        Assert.assertFalse(exists[1]);
+        Assert.assertTrue(exists[2]);
     }
 
     @Test