adapt fts query request

GroundWu · GroundWu · commit 0c4cdf2aae72 · 2025-01-03T16:38:28.000+08:00
diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObClusterTableQuery.java b/src/main/java/com/alipay/oceanbase/rpc/ObClusterTableQuery.java
@@ -264,4 +264,10 @@ public void setEntityType(ObTableEntityType entityType) {
         super.setEntityType(entityType);
         tableClientQuery.setEntityType(entityType);
     }
+
+    @Override
+    public TableQuery setSearchText(String searchText) {
+        tableClientQuery.setSearchText(searchText);
+        return this;
+    }
 }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObIndexType.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObIndexType.java
@@ -21,16 +21,38 @@
 import java.util.Map;
 
 public enum ObIndexType {
-    IndexTypeIsNot(0), IndexTypeNormalLocal(1), IndexTypeUniqueLocal(2), IndexTypeNormalGlobal(3), IndexTypeUniqueGlobal(
-                                                                                                                         4), IndexTypePrimary(
-                                                                                                                                              5), IndexTypeDomainCtxcat(
-                                                                                                                                                                        6), IndexTypeNormalGlobalLocalStorage(
-                                                                                                                                                                                                              7), IndexTypeUniqueGlobalLocalStorage(
-                                                                                                                                                                                                                                                    8), IndexTypeSpatialLocal(
-                                                                                                                                                                                                                                                                              10), IndexTypeSpatialGlobal(
-                                                                                                                                                                                                                                                                                                          11), IndexTypeSpatialGlobalLocalStorage(
-                                                                                                                                                                                                                                                                                                                                                  12), IndexTypeMax(
-                                                                                                                                                                                                                                                                                                                                                                    13);
+    IndexTypeIsNot(0),
+    IndexTypeNormalLocal(1),
+    IndexTypeUniqueLocal(2),
+    IndexTypeNormalGlobal(3),
+    IndexTypeUniqueGlobal(4),
+    IndexTypePrimary(5),
+    IndexTypeDomainCtxcat(6),
+    IndexTypeNormalGlobalLocalStorage(7),
+    IndexTypeUniqueGlobalLocalStorage(8),
+    IndexTypeSpatialLocal(10),
+    IndexTypeSpatialGlobal(11),
+    IndexTypeSpatialGlobalLocalStorage(12),
+    IndexTypeRowkeyDocIdLocal(13),
+    IndexTypeDocIdRowkeyLocal(14),
+    IndexTypeFtsIndexLocal(15),
+    IndexTypeFtsDocWordLocal(16),
+    /*
+    IndexTypeDocIdRowkeyGlobal(17),
+    IndexTypeFtsIndexGlobal(18),
+    IndexTypeFtsDocWordGlobal(19),
+    IndexTypeDocIdRowkeyGlobalLocalStorage(20),
+    IndexTypeFtsIndexGlobalLocalStorage(21),
+    IndexTypeFtsDocWordGlobalLocalStorage(22),
+    IndexTypeNormalMultivalueLocal(23),
+    IndexTypeUniqueMultivalueLocal(24),
+    IndexTypeVecRowkeyVidLocal(25),
+    IndexTypeVecVidRowkeyLocal(26),
+    IndexTypeVecDeltaBufferLocal(27),
+    IndexTypeVecIndexIdLocal(28),
+    IndexTypeVecIndexSnapshotDataLocal(29),
+    */
+    IndexTypeMax(30);
 
     private int                              value;
     private static Map<Integer, ObIndexType> map = new HashMap<Integer, ObIndexType>();
diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQuery.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQuery.java
@@ -18,14 +18,16 @@
 package com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query;
 
 import com.alipay.oceanbase.rpc.exception.FeatureNotSupportedException;
+import com.alipay.oceanbase.rpc.table.ObFTSParams;
 import com.alipay.oceanbase.rpc.table.ObHBaseParams;
-import com.alipay.oceanbase.rpc.table.ObKVParams;
+import com.alipay.oceanbase.rpc.table. ObKVParams;
 import com.alipay.oceanbase.rpc.protocol.payload.AbstractPayload;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.ObRowKey;
 import com.alipay.oceanbase.rpc.protocol.payload.Constants;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.aggregation.ObTableAggregationSingle;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.aggregation.ObTableAggregationType;
+import com.alipay.oceanbase.rpc.table.ObKVParamsBase;
 import com.alipay.oceanbase.rpc.util.Serialization;
 import io.netty.buffer.ByteBuf;
 
@@ -74,7 +76,7 @@ public class ObTableQuery extends AbstractPayload {
 
     private Long partId = null;
 
-    protected ObKVParams obKVParams;
+    private ObKVParams obKVParams = null;
 
     public void adjustStartKey(List<ObObj> key) throws IllegalArgumentException {
         List<ObNewRange> keyRanges = getKeyRanges();
@@ -226,7 +228,7 @@ public byte[] encode() {
             idx += len;
         }
 
-        if (isHbaseQuery && obKVParams != null) {
+        if (obKVParams != null) { // hbaseQuery or FTSQuery will use obKVParams
             len = (int) obKVParams.getPayloadSize();
             System.arraycopy(obKVParams.encode(), 0, bytes, idx, len);
             idx += len;
@@ -293,7 +295,11 @@ public Object decode(ByteBuf buf) {
             String agg_column = Serialization.decodeVString(buf);
             this.aggregations.add(new ObTableAggregationSingle(ObTableAggregationType.fromByte(agg_type), agg_column));
         }
-        if (isHbaseQuery) {
+
+        buf.markReaderIndex();
+        if (buf.readByte() > 0) {
+            // read pType if is exists
+            buf.resetReaderIndex();
             obKVParams = new ObKVParams();
             this.obKVParams.decode(buf);
         }
@@ -328,7 +334,7 @@ public long getPayloadContentSize() {
         } else {
             contentSize += HTABLE_DUMMY_BYTES.length;
         }
-        if (isHbaseQuery && obKVParams != null) {
+        if (obKVParams != null) {
             contentSize += obKVParams.getPayloadSize();
         } else {
             contentSize += HTABLE_DUMMY_BYTES.length;
@@ -554,6 +560,18 @@ public void setObKVParams(ObKVParams obKVParams) {
         this.obKVParams = obKVParams;
     }
 
+    public void setSearchText(String searchText) {
+        if (isHbaseQuery) {
+            throw new FeatureNotSupportedException("Hbase query not support full text search currently");
+        }
+        if (obKVParams == null) {
+            obKVParams = new ObKVParams();
+        }
+        ObFTSParams ftsParams = (ObFTSParams)obKVParams.getObParams(ObKVParamsBase.paramType.FTS);
+        ftsParams.setSearchText(searchText);
+        obKVParams.setObParamsBase(ftsParams);
+    }
+
     public ObKVParams getObKVParams() {
         return obKVParams;
     }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/AbstractTableQueryImpl.java b/src/main/java/com/alipay/oceanbase/rpc/table/AbstractTableQueryImpl.java
@@ -23,6 +23,7 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObNewRange;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObScanOrder;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObTableQuery;
+import com.alipay.oceanbase.rpc.table.api.Table;
 import com.alipay.oceanbase.rpc.table.api.TableQuery;
 
 import java.util.Arrays;
@@ -184,6 +185,12 @@ public TableQuery setMaxResultSize(long maxResultSize) {
         return this;
     }
 
+    @Override
+    public TableQuery setSearchText(String searchText) {
+        this.tableQuery.setSearchText(searchText);
+        return this;
+    }
+
     public String getIndexTableName() {
         return indexTableName;
     }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/ObFTSParams.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObFTSParams.java
@@ -0,0 +1,64 @@
+/*-
+ * #%L
+ * com.oceanbase:obkv-table-client
+ * %%
+ * Copyright (C) 2021 - 2024 OceanBase
+ * %%
+ * OBKV Table 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.rpc.table;
+
+import com.alipay.oceanbase.rpc.util.Serialization;
+import io.netty.buffer.ByteBuf;
+
+public class ObFTSParams extends ObKVParamsBase {
+    String searchText = null;
+    public ObFTSParams() {
+        pType = paramType.FTS;
+    }
+
+    public paramType getType() {
+        return pType;
+    }
+
+    public void setSearchText(String searchText) {
+        this.searchText = searchText;
+    }
+
+    public String getSearchText() { return this.searchText; }
+
+    public byte[] encode() {
+        byte[] bytes = new byte[(int) getPayloadContentSize()];
+        int idx = 0;
+        byte[] b = new byte[] { (byte)pType.ordinal() };
+        System.arraycopy(b, 0, bytes, idx, 1);
+        idx += 1;
+        int len = Serialization.getNeedBytes(searchText);
+        System.arraycopy(Serialization.encodeVString(searchText), 0, bytes, idx, len);
+        return bytes;
+    }
+
+    public Object decode(ByteBuf buf) {
+        // pType is read by ObKVParams
+        this.searchText = Serialization.decodeVString(buf);
+        return this;
+    }
+
+    public long getPayloadContentSize() {
+       return 1 /* pType*/ + Serialization.getNeedBytes(searchText);
+    }
+
+    public String toString() {
+        return "ObFtsParams: {\n pType = " + pType + ", \n searchText = " + searchText
+                + "\n}\n";
+    }
+}
diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/ObKVParams.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObKVParams.java
@@ -32,6 +32,9 @@ public ObKVParamsBase getObParams(ObKVParamsBase.paramType pType) {
             case HBase:
                 return new ObHBaseParams();
             case Redis:
+                throw new RuntimeException("Currently does not support redis type");
+            case FTS:
+                return new ObFTSParams();
             default:
                 throw new RuntimeException("Currently does not support other types except HBase");
         }
diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/ObKVParamsBase.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObKVParamsBase.java
@@ -17,20 +17,31 @@
 
 package com.alipay.oceanbase.rpc.table;
 
+import com.alipay.oceanbase.rpc.direct_load.protocol.payload.ObTableLoadClientStatus;
 import io.netty.buffer.ByteBuf;
 
+import java.util.HashMap;
+import java.util.Map;
+
 public abstract class ObKVParamsBase {
     public enum paramType {
-        HBase((byte) 0), Redis((byte) 1);
+        HBase((byte) 0), Redis((byte) 1), FTS((byte) 2);
         private final byte value;
+        private static final Map<Integer, paramType> map = new HashMap<Integer, paramType>();
+
+        static {
+            for (paramType type : paramType.values()) {
+                map.put(type.ordinal(), type);
+            }
+        }
+
+        public static paramType valueOf(int value) { return map.get(value); }
 
         paramType(byte value) {
             this.value = value;
         }
 
-        public byte getValue() {
-            return value;
-        }
+        public byte getValue() { return value; }
     }
 
     public int       byteSize;
diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/api/TableQuery.java b/src/main/java/com/alipay/oceanbase/rpc/table/api/TableQuery.java
@@ -199,4 +199,6 @@ public interface TableQuery {
     TableQuery setScanRangeColumns(String... columns);
 
     void clear();
+
+    TableQuery setSearchText(String searchText);
 }
diff --git a/src/test/java/com/alipay/oceanbase/rpc/ObTableFullTextIndexTest.java b/src/test/java/com/alipay/oceanbase/rpc/ObTableFullTextIndexTest.java
@@ -3,6 +3,8 @@
 import com.alipay.oceanbase.rpc.exception.ObTableException;
 import com.alipay.oceanbase.rpc.mutation.result.MutationResult;
 import com.alipay.oceanbase.rpc.protocol.payload.ResultCodes;
+import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObTableQuery;
+import com.alipay.oceanbase.rpc.stream.QueryResultSet;
 import com.alipay.oceanbase.rpc.table.ObTable;
 import com.alipay.oceanbase.rpc.util.ObTableClientTestUtil;
 import com.google.protobuf.MapEntry;
@@ -323,6 +325,39 @@ public void testTTLInsert() throws Exception {
         }
     }
 
+    @Test
+    public void testFtsQuery() throws Exception {
+        try {
+            //sync query
+            QueryResultSet resultSet = client.query(tableName)
+                    .setSearchText("native")
+                    .indexName("full_idx1_tbl1")
+                    .execute();
+            while(resultSet.next()) {
+                Map<String, Object> row = resultSet.getRow();
+                for (Map.Entry<String, Object> entry: row.entrySet()) {
+                    System.out.println("colname: " + entry.getKey() + " \nvalue: " + entry.getValue());
+                }
+                System.out.println();
+            }
+            // async query
+            System.out.println("========async query:=========");
+            QueryResultSet asyncResultSet = client.query(tableName)
+                    .indexName("full_idx1_tbl1")
+                    .setSearchText("oceanbase")
+                    .asyncExecute();
+            while(asyncResultSet.next()) {
+                Map<String, Object> row = asyncResultSet.getRow();
+                for (Map.Entry<String, Object> entry: row.entrySet()) {
+                    System.out.println("colname: " + entry.getKey() + " \nvalue: " + entry.getValue());
+                }
+                System.out.println();
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
     private void executeSQL(String createSQL) throws SQLException {
         Connection connection = ObTableClientTestUtil.getConnection();
         Statement statement = connection.createStatement();
diff --git a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryPayloadTest.java b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryPayloadTest.java
@@ -20,11 +20,15 @@
 import com.alipay.oceanbase.rpc.protocol.payload.impl.*;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableConsistencyLevel;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntityType;
+import com.alipay.oceanbase.rpc.table.ObFTSParams;
 import com.alipay.oceanbase.rpc.table.ObHBaseParams;
 import com.alipay.oceanbase.rpc.table.ObKVParams;
+import com.alipay.oceanbase.rpc.table.ObKVParamsBase;
 import com.alipay.oceanbase.rpc.util.ObBytesString;
+import com.sun.xml.internal.ws.wsdl.writer.document.ParamType;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.PooledByteBufAllocator;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.util.ArrayList;
@@ -150,6 +154,37 @@ public void test_ObTableQueryResult() {
         buf.release();
     }
 
+    @Test
+    public void testFtsParam() {
+        ObFTSParams ftsParams = new ObFTSParams();
+        ftsParams.setSearchText("oceanbase");
+        byte[] bytes = ftsParams.encode();
+        ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer();
+        buf.writeBytes(bytes);
+        ObFTSParams newFtsParams = new ObFTSParams();
+        assertEquals(ftsParams.getType(), ObKVParamsBase.paramType.valueOf(buf.readByte()));
+        newFtsParams.decode(buf);
+        assertEquals(ftsParams.getSearchText(), newFtsParams.getSearchText());
+        buf.release();
+    }
+
+    @Test
+    public void testFtsQuery() {
+        ObTableQuery obTableQuery = getObTableQuery();
+        obTableQuery.setIndexName("ftx_idx");
+        obTableQuery.setSearchText("oceanbase");
+        byte[] bytes = obTableQuery.encode();
+        ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer();
+        buf.writeBytes(bytes);
+
+        ObTableQuery newObTableQuery = new ObTableQuery();
+        newObTableQuery.decode(buf);
+        ObKVParamsBase kv_params_base = newObTableQuery.getObKVParams().obKVParamsBase;
+        Assert.assertEquals(ObKVParamsBase.paramType.FTS, kv_params_base.getType());
+        ObFTSParams fts_params = (ObFTSParams) kv_params_base;
+        Assert.assertEquals("oceanbase", fts_params.getSearchText());
+    }
+
     private ObTableQuery getObTableQuery() {
         ObTableQuery obTableQuery = new ObTableQuery();
         obTableQuery.addKeyRange(getObNewRange());

Original file line number	Diff line number	Diff line change
`@@ -264,4 +264,10 @@ public void setEntityType(ObTableEntityType entityType) {`
`264`	`264`	`super.setEntityType(entityType);`
`265`	`265`	`tableClientQuery.setEntityType(entityType);`
`266`	`266`	`}`
	`267`	`+`
	`268`	`+ @Override`
	`269`	`+ public TableQuery setSearchText(String searchText) {`
	`270`	`+ tableClientQuery.setSearchText(searchText);`
	`271`	`+ return this;`
	`272`	`+ }`
`267`	`273`	`}`
Original file line number	Diff line number	Diff line change
`@@ -199,4 +199,6 @@ public interface TableQuery {`
`199`	`199`	`TableQuery setScanRangeColumns(String... columns);`
`200`	`200`
`201`	`201`	`void clear();`
	`202`	`+`
	`203`	`+ TableQuery setSearchText(String searchText);`
`202`	`204`	`}`