diff --git a/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java b/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java index bae57eaf..625df441 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java +++ b/src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java @@ -2258,8 +2258,8 @@ public ObPayload execute(final ObHbaseRequest request) throws Exception { String realTableName = request.getCfRows().get(0).getRealTableName(); int keyIdx = request.getCfRows().get(0).getKeyIndex(0); row.add("K", request.getKeys().get(keyIdx).getValue()); - row.add("Q", request.getCfRows().get(0).getCells().get(0).getQ().getValue()); - row.add("T", request.getCfRows().get(0).getCells().get(0).getT().getValue()); + row.add("Q", request.getCfRows().get(0).getFirstCellQualifierValue()); + row.add("T", request.getCfRows().get(0).getFirstCellTimestampValue()); return execute(realTableName, new OperationExecuteCallback(row, null) { @Override diff --git a/src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemoting.java b/src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemoting.java index 3cf8bbd5..b58d516b 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemoting.java +++ b/src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemoting.java @@ -22,8 +22,11 @@ import com.alipay.oceanbase.rpc.exception.*; import com.alipay.oceanbase.rpc.protocol.packet.ObCompressType; import com.alipay.oceanbase.rpc.protocol.payload.*; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.OHOperationType; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableEntityType; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableLSOpRequest; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.ObTableLSOpResult; import com.alipay.oceanbase.rpc.protocol.payload.impl.login.ObTableLoginRequest; -import com.alipay.oceanbase.rpc.util.ObPureCrc32C; import com.alipay.oceanbase.rpc.util.TableClientLoggerFactory; import com.alipay.oceanbase.rpc.util.TraceUtil; import com.alipay.remoting.*; @@ -102,17 +105,6 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques throw new FeatureNotSupportedException(errMessage); } ByteBuf buf = response.getPacketContentBuf(); - // verify checksum - long expected_checksum = response.getHeader().getChecksum(); - byte[] content = new byte[buf.readableBytes()]; - buf.getBytes(buf.readerIndex(), content); - if (ObPureCrc32C.calculate(content) != expected_checksum) { - String errMessage = TraceUtil.formatTraceMessage(conn, request, - "get response with checksum error: " + response.getMessage()); - ExceptionUtil.throwObTableTransportException(errMessage, - TransportCodes.BOLT_CHECKSUM_ERR); - return null; - } // decode ResultCode for response packet boolean isRoutingWrong = false; @@ -165,6 +157,14 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques "receive unexpected command code: " + response.getCmdCode().value()); throw new ObTableUnexpectedException(errMessage, resultCode.getRcode()); } + if (payload instanceof ObTableLSOpResult && request instanceof ObTableLSOpRequest) { + ObTableLSOpRequest lsRequest = (ObTableLSOpRequest) request; + OHOperationType hbaseOpType = lsRequest.getHbaseOpType(); + boolean eligibleHBaseBatchGet = lsRequest.getEntityType() == ObTableEntityType.HKV + && (hbaseOpType == OHOperationType.GET_LIST + || hbaseOpType == OHOperationType.BATCH); + ((ObTableLSOpResult) payload).setDecodeHBaseKqtv(eligibleHBaseBatchGet); + } try { payload.decode(buf); } catch (Exception e) { diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObj.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObj.java index 7fa4e1b5..43a0cb89 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObj.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObj.java @@ -29,6 +29,13 @@ public class ObObj implements ObSimplePayload { private static long MAX_OBJECT_VALUE = -2L; private static long MIN_OBJECT_VALUE = -3L; + /** Cached meta for HBase Put V2 hot path (Q/V rowkey bytes or slice view). */ + private static final ObObjMeta HBASE_PUT_VARCHAR_META = ObObjType.ObVarcharType + .getDefaultObjMeta(); + /** Cached meta for HBase Put V2 timestamp / TTL (signed int64). */ + private static final ObObjMeta HBASE_PUT_INT64_META = ObObjType.ObInt64Type + .getDefaultObjMeta(); + static { MAX_OBJECT = new ObObj(ObObjType.ObExtendType.getDefaultObjMeta(), MAX_OBJECT_VALUE); MIN_OBJECT = new ObObj(ObObjType.ObExtendType.getDefaultObjMeta(), MIN_OBJECT_VALUE); @@ -166,6 +173,20 @@ public static ObObj getInstance(Object value) { } } + /** + * HBase Put V2 Q/V/rowkey: skip {@link ObObjType#defaultObjMeta(Object)} dispatch. + */ + public static ObObj hbasePutVarchar(Object value) { + return new ObObj(HBASE_PUT_VARCHAR_META, value); + } + + /** + * HBase Put V2 timestamp / cell TTL: skip meta lookup; value autoboxes once to Long. + */ + public static ObObj hbasePutInt64(long value) { + return new ObObj(HBASE_PUT_INT64_META, value); + } + /* * Get max. */ diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjType.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjType.java index 0fc7a7cb..0bbd9a96 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjType.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjType.java @@ -1270,7 +1270,9 @@ public Comparable parseToComparable(Object o, ObCollationType ct) */ @Override public byte[] encode(Object obj) { - if (obj instanceof byte[]) { + if (obj instanceof ObBytesString) { + return Serialization.encodeBytesString((ObBytesString) obj); + } else if (obj instanceof byte[]) { ObBytesString bytesString = new ObBytesString((byte[]) obj); return Serialization.encodeBytesString(bytesString); } else if (obj instanceof ObVString) { @@ -1282,7 +1284,9 @@ public byte[] encode(Object obj) { @Override public void encode(ObByteBuf buf, Object obj) { - if (obj instanceof byte[]) { + if (obj instanceof ObBytesString) { + Serialization.encodeBytesString(buf, (ObBytesString) obj); + } else if (obj instanceof byte[]) { ObBytesString bytesString = new ObBytesString((byte[]) obj); Serialization.encodeBytesString(buf, bytesString); } else if (obj instanceof ObVString) { @@ -1976,7 +1980,9 @@ public Date parseToComparable(Object o, ObCollationType ct) */ private int value; - private static Map map = new HashMap(); + // Object types encoded as i8 use the array fast path; extended type ids use the map. + private static final ObObjType[] VALUE_LOOKUP = new ObObjType[128]; + private static final Map EXTENDED_VALUE_LOOKUP = new HashMap(); ObObjType(int value) { this.value = value; @@ -1984,7 +1990,17 @@ public Date parseToComparable(Object o, ObCollationType ct) static { for (ObObjType type : ObObjType.values()) { - map.put(type.value, type); + int value = type.value; + if (value < 0) { + throw new IllegalStateException("ObObjType value must not be negative: " + value); + } else if (value < VALUE_LOOKUP.length) { + if (VALUE_LOOKUP[value] != null) { + throw new IllegalStateException("duplicate ObObjType value: " + value); + } + VALUE_LOOKUP[value] = type; + } else if (EXTENDED_VALUE_LOOKUP.put(value, type) != null) { + throw new IllegalStateException("duplicate ObObjType value: " + value); + } } } @@ -2016,6 +2032,8 @@ public static ObObjType valueOfType(Object object) { return ObVarcharType; } else if (object instanceof byte[]) { return ObVarcharType; + } else if (object instanceof ObBytesString) { + return ObVarcharType; } else if (object instanceof ObVString) { return ObVarcharType; } else if (object instanceof Double) { @@ -2039,7 +2057,11 @@ public static ObObjType valueOfType(Object object) { * Value of. */ public static ObObjType valueOf(int value) { - return map.get(value); + if (value < 0) { + return null; + } + return value < VALUE_LOOKUP.length ? VALUE_LOOKUP[value] : EXTENDED_VALUE_LOOKUP + .get(value); } /* @@ -2113,7 +2135,9 @@ public Object decodeText(ByteBuf buf, ObCollationType type) { * Get text encoded size. */ public static int getTextEncodedSize(Object obj) { - if (obj instanceof byte[]) { + if (obj instanceof ObBytesString) { + return Serialization.getNeedBytes((ObBytesString) obj); + } else if (obj instanceof byte[]) { ObBytesString bytesString = new ObBytesString((byte[]) obj); return Serialization.getNeedBytes(bytesString); } else if (obj instanceof ObVString) { @@ -2130,7 +2154,7 @@ public static byte[] parseTextToBytes(ObObjType obObjType, Object object, ObCollationType collationType) { if (collationType == ObCollationType.CS_TYPE_BINARY) { if (object instanceof ObBytesString) { - return ((ObBytesString) object).bytes; + return materializeBytesString((ObBytesString) object); } if (object instanceof byte[]) { @@ -2150,7 +2174,8 @@ public static byte[] parseTextToBytes(ObObjType obObjType, Object object, return ((String) object).getBytes(); } if (object instanceof ObBytesString) { - return (Serialization.decodeVString(((ObBytesString) object).bytes)).getBytes(); + return Serialization.decodeVString(materializeBytesString((ObBytesString) object)) + .getBytes(); } if (object instanceof byte[]) { @@ -2201,7 +2226,7 @@ public static Comparable parseTextToComparable(ObObjType obObjType, Object objec return (String) object; } if (object instanceof ObBytesString) { - return Serialization.decodeVString(((ObBytesString) object).bytes); + return Serialization.decodeVString(materializeBytesString((ObBytesString) object)); } if (object instanceof byte[]) { @@ -2224,6 +2249,14 @@ public static Comparable parseTextToComparable(ObObjType obObjType, Object objec + object); } + private static byte[] materializeBytesString(ObBytesString bytesString) { + if (bytesString.offset == 0 && bytesString.length() == bytesString.bytes.length) { + return bytesString.bytes; + } + return Arrays.copyOfRange(bytesString.bytes, bytesString.offset, + bytesString.offset + bytesString.length()); + } + /* * Parse timestamp. */ diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableObjType.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableObjType.java index fda8d354..e448940f 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableObjType.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableObjType.java @@ -202,9 +202,11 @@ public void decode(ByteBuf buf, ObObj obj) { ObTableInvalidType(26) { }; + private static final int LOOKUP_SIZE = 128; private int value; - // mapping from value to enum - private static Map valueMap = new HashMap(); + // Keep current low ids on the fast path; reserve a map for future ids outside the array. + private static final ObTableObjType[] VALUE_LOOKUP = new ObTableObjType[LOOKUP_SIZE]; + private static final Map OVERFLOW_VALUE_LOOKUP = new HashMap<>(); // mapping from ObTableObjType to ObObjType private static Map tableObjTypeMap = new HashMap<>(); @@ -214,7 +216,21 @@ public void decode(ByteBuf buf, ObObj obj) { static { for (ObTableObjType type : ObTableObjType.values()) { - valueMap.put(type.value, type); + registerLookup(type.value, type); + } + } + + private static void registerLookup(int value, ObTableObjType type) { + if (value < 0) { + throw new IllegalStateException("Invalid table object type id: " + value); + } + if (value < VALUE_LOOKUP.length) { + if (VALUE_LOOKUP[value] != null) { + throw new IllegalStateException("Duplicate table object type id: " + value); + } + VALUE_LOOKUP[value] = type; + } else if (OVERFLOW_VALUE_LOOKUP.put(value, type) != null) { + throw new IllegalStateException("Duplicate table object type id: " + value); } } @@ -298,7 +314,11 @@ public static ObObjType getObjType(ObTableObjType tableObjType) { * Value of. */ public static ObTableObjType valueOf(int value) { - return valueMap.get(value); + if (value < 0) { + return null; + } + return value < VALUE_LOOKUP.length ? VALUE_LOOKUP[value] + : OVERFLOW_VALUE_LOOKUP.get(value); } /* diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableSerialUtil.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableSerialUtil.java index c8deb039..af5e2f6a 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableSerialUtil.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableSerialUtil.java @@ -48,7 +48,7 @@ public static ObTableObjType decodeTableObjType(ByteBuf buf) { if (buf == null) { throw new IllegalArgumentException("cannot get ObTableObjType, buf is null"); } - byte type = Serialization.decodeI8(buf); + int type = Serialization.decodeI8(buf) & 0xFF; ObTableObjType objType = ObTableObjType.valueOf(type); if (objType == null) { throw new IllegalArgumentException("cannot get table object type from value, type: " + type); diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbaseCfRows.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbaseCfRows.java index 42421b9e..96fee606 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbaseCfRows.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbaseCfRows.java @@ -19,9 +19,11 @@ import com.alipay.oceanbase.rpc.protocol.payload.AbstractPayload; import com.alipay.oceanbase.rpc.util.ObByteBuf; +import com.alipay.oceanbase.rpc.util.ObBytesString; import com.alipay.oceanbase.rpc.util.Serialization; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; public class ObHbaseCfRows extends AbstractPayload { @@ -30,6 +32,22 @@ public class ObHbaseCfRows extends AbstractPayload { private List cellNumArray = new ArrayList<>(); // the number of original cells to each key private List cells = new ArrayList<>(); // original cells + private byte[][] compactQualifierArrays; + private int[] compactQualifierOffsets; + private int[] compactQualifierLengths; + private long[] compactTimestamps; + private byte[][] compactValueArrays; + private int[] compactValueOffsets; + private int[] compactValueLengths; + private int[] compactCellContentSizes; + private long[] compactKeyRunTtls; + private boolean compactMode; + private int compactCellCount; + private int compactKeyRunCount; + private int compactRunRemainingCells; + private long compactRunTtl = Long.MAX_VALUE; + private long compactCellsPayloadSize; + public ObHbaseCfRows() {} public ObHbaseCfRows(List keyIndex, List cellNumArray, List cells) { @@ -47,11 +65,143 @@ public void setRealTableName(String realTableName) { } public void add(Integer index, Integer cellNum, List cells) { + ensureLegacyMode(); this.keyIndex.add(index); this.cellNumArray.add(cellNum); this.cells.addAll(cells); } + /** + * Begin a key's cell run without a temporary {@code List}. + * Call {@link #appendCell(ObHbaseCell)} {@code cellCount} times next. + */ + public void beginKeyCells(int index, int cellCount) { + ensureLegacyMode(); + this.keyIndex.add(index); + this.cellNumArray.add(cellCount); + } + + /** Append one cell after {@link #beginKeyCells(int, int)}. */ + public void appendCell(ObHbaseCell cell) { + ensureLegacyMode(); + this.cells.add(cell); + } + + /** Begin a key's compact Put cell run. TTL is shared by all cells in this run. */ + public void beginCompactKeyCells(int index, int cellCount, long ttl) { + ensureCompactMode(); + compactMode = true; + if (cellCount < 0) { + throw new IllegalArgumentException("cell count is negative: " + cellCount); + } + if (compactRunRemainingCells != 0) { + throw new IllegalStateException("previous compact key run is incomplete: " + + compactRunRemainingCells); + } + ensureCompactKeyRunCapacity(compactKeyRunCount + 1); + keyIndex.add(index); + cellNumArray.add(cellCount); + compactKeyRunTtls[compactKeyRunCount++] = ttl; + compactRunRemainingCells = cellCount; + compactRunTtl = ttl; + resetPayloadContentSize(); + } + + /** Append one compact Put cell after {@link #beginCompactKeyCells(int, int, long)}. */ + public void appendCompactCell(byte[] qualifier, int qualifierOffset, int qualifierLength, + long timestamp, byte[] value, int valueOffset, int valueLength) { + ensureCompactMode(); + if (compactRunRemainingCells <= 0) { + throw new IllegalStateException("no compact key run is awaiting cells"); + } + checkSlice(qualifier, qualifierOffset, qualifierLength, "qualifier"); + checkSlice(value, valueOffset, valueLength, "value"); + ensureCompactCellCapacity(compactCellCount + 1); + + compactQualifierArrays[compactCellCount] = qualifier; + compactQualifierOffsets[compactCellCount] = qualifierOffset; + compactQualifierLengths[compactCellCount] = qualifierLength; + compactTimestamps[compactCellCount] = timestamp; + compactValueArrays[compactCellCount] = value; + compactValueOffsets[compactCellCount] = valueOffset; + compactValueLengths[compactCellCount] = valueLength; + boolean hasTtl = compactRunTtl != Long.MAX_VALUE; + long cellContentSize = ObHbasePutCellCodec.getCellContentSize(qualifierLength, timestamp, + valueLength, hasTtl, compactRunTtl); + if (cellContentSize > Integer.MAX_VALUE) { + throw new IllegalArgumentException("compact cell content is too large: " + + cellContentSize); + } + compactCellContentSizes[compactCellCount] = (int) cellContentSize; + long cellPayloadSize = ObHbasePutCellCodec.getCellPayloadSize(cellContentSize); + if (compactCellsPayloadSize > Long.MAX_VALUE - cellPayloadSize) { + throw new IllegalArgumentException("compact cells payload size overflow"); + } + compactCellsPayloadSize += cellPayloadSize; + compactCellCount++; + compactRunRemainingCells--; + payLoadContentSize = INVALID_PAYLOAD_CONTENT_SIZE; + } + + /** Reserve compact cell arrays without creating per-cell objects. */ + public void reserveAdditionalCompactCells(int additionalCapacity) { + ensureCompactMode(); + compactMode = true; + if (additionalCapacity < 0 || additionalCapacity > Integer.MAX_VALUE - compactCellCount) { + throw new IllegalArgumentException("invalid additional compact cell capacity: " + + additionalCapacity); + } + ensureCompactCellCapacity(compactCellCount + additionalCapacity); + } + + public boolean hasCompactCells() { + return compactMode; + } + + /** Return the first cell qualifier in the representation used by table routing. */ + public Object getFirstCellQualifierValue() { + if (!hasCompactCells()) { + return getFirstLegacyCell().getQ().getValue(); + } + validateCompactFirstCell(); + byte[] qualifier = compactQualifierArrays[0]; + int offset = compactQualifierOffsets[0]; + int length = compactQualifierLengths[0]; + return offset == 0 && length == qualifier.length ? qualifier : new ObBytesString(qualifier, + offset, length); + } + + /** Return the first cell timestamp in the representation used by table routing. */ + public Object getFirstCellTimestampValue() { + if (!hasCompactCells()) { + return getFirstLegacyCell().getT().getValue(); + } + validateCompactFirstCell(); + return compactTimestamps[0]; + } + + /** Hint the number of key runs that will be appended. */ + public void reserveKeyRuns(int minCapacity) { + if (keyIndex instanceof ArrayList) { + ((ArrayList) keyIndex).ensureCapacity(minCapacity); + } + if (cellNumArray instanceof ArrayList) { + ((ArrayList) cellNumArray).ensureCapacity(minCapacity); + } + } + + /** Reserve room for another cell run without over-allocating other families. */ + public void reserveAdditionalCells(int additionalCapacity) { + ensureLegacyMode(); + if (additionalCapacity < 0 || additionalCapacity > Integer.MAX_VALUE - cells.size()) { + throw new IllegalArgumentException("invalid additional cell capacity: " + + additionalCapacity); + } + if (cells instanceof ArrayList) { + ((ArrayList) cells).ensureCapacity(cells.size() + additionalCapacity); + } + } + public List getKeyIndex() { return keyIndex; } @@ -92,52 +242,29 @@ public void encode(ObByteBuf buf) { } // 4. encode cells without length - for (ObHbaseCell cell : cells) { - cell.encode(buf); + if (hasCompactCells()) { + encodeCompactCells(buf); + } else { + for (ObHbaseCell cell : cells) { + cell.encode(buf); + } } } @Override public byte[] encode() { - byte[] bytes = new byte[(int) getPayloadSize()]; - int idx = 0; - - // 0. encode header - idx = encodeHeader(bytes, idx); - - // 1. encode family - byte[] strbytes = Serialization.encodeVString(realTableName); - System.arraycopy(strbytes, 0, bytes, idx, strbytes.length); - idx += strbytes.length; - - // 2. encode keyIndex - int len = Serialization.getNeedBytes(keyIndex.size()); - System.arraycopy(Serialization.encodeVi64(keyIndex.size()), 0, bytes, idx, len); - idx += len; - for (long index : keyIndex) { - len = Serialization.getNeedBytes(index); - System.arraycopy(Serialization.encodeVi64(index), 0, bytes, idx, len); - idx += len; - } - - // 3. encode cellNumArray - len = Serialization.getNeedBytes(cellNumArray.size()); - System.arraycopy(Serialization.encodeVi64(cellNumArray.size()), 0, bytes, idx, len); - idx += len; - for (long cellNum : cellNumArray) { - len = Serialization.getNeedBytes(cellNum); - System.arraycopy(Serialization.encodeVi64(cellNum), 0, bytes, idx, len); - idx += len; + long payloadSize = getPayloadSize(); + if (payloadSize > Integer.MAX_VALUE) { + throw new IllegalArgumentException("ObHbaseCfRows payload is too large: " + + payloadSize); } - - // 4. encode cells without length - for (ObHbaseCell cell : cells) { - len = (int) cell.getPayloadSize(); - System.arraycopy(cell.encode(), 0, bytes, idx, len); - idx += len; + ObByteBuf buf = new ObByteBuf((int) payloadSize); + encode(buf); + if (buf.pos != buf.bytes.length) { + throw new IllegalArgumentException("error in encode ObHbaseCfRows (pos:" + buf.pos + + ", capacity:" + buf.bytes.length + ")"); } - - return bytes; + return buf.bytes; } @Override @@ -160,8 +287,13 @@ public long getPayloadContentSize() { } // only add cells size - for (ObHbaseCell cell : cells) { - payloadContentSize += cell.getPayloadSize(); + if (hasCompactCells()) { + validateCompactState(); + payloadContentSize += compactCellsPayloadSize; + } else { + for (ObHbaseCell cell : cells) { + payloadContentSize += cell.getPayloadSize(); + } } this.payLoadContentSize = payloadContentSize; } @@ -171,11 +303,130 @@ public long getPayloadContentSize() { @Override public void resetPayloadContentSize() { super.resetPayloadContentSize(); - for (ObHbaseCell cell : cells) { - if (cell != null) { - cell.resetPayloadContentSize(); + if (!hasCompactCells()) { + for (ObHbaseCell cell : cells) { + if (cell != null) { + cell.resetPayloadContentSize(); + } + } + } + } + + private void encodeCompactCells(ObByteBuf buf) { + validateCompactState(); + int cellIndex = 0; + for (int runIndex = 0; runIndex < compactKeyRunCount; runIndex++) { + long ttl = compactKeyRunTtls[runIndex]; + int cellCount = cellNumArray.get(runIndex); + for (int i = 0; i < cellCount; i++, cellIndex++) { + ObHbasePutCellCodec.encodeCell(buf, compactQualifierArrays[cellIndex], + compactQualifierOffsets[cellIndex], compactQualifierLengths[cellIndex], + compactTimestamps[cellIndex], compactValueArrays[cellIndex], + compactValueOffsets[cellIndex], compactValueLengths[cellIndex], + ttl != Long.MAX_VALUE, ttl, compactCellContentSizes[cellIndex]); } } } + private void validateCompactState() { + if (compactRunRemainingCells != 0) { + throw new IllegalStateException("compact key run is incomplete: " + + compactRunRemainingCells); + } + if (compactKeyRunCount != cellNumArray.size()) { + throw new IllegalStateException("compact key run count mismatch: runs=" + + compactKeyRunCount + ", cellNumArray=" + + cellNumArray.size()); + } + long expectedCellCount = 0; + for (Integer cellCount : cellNumArray) { + expectedCellCount += cellCount; + } + if (expectedCellCount != compactCellCount) { + throw new IllegalStateException("compact cell count mismatch: expected=" + + expectedCellCount + ", actual=" + compactCellCount); + } + } + + private void validateCompactFirstCell() { + if (compactCellCount == 0) { + throw new IllegalStateException("compact cell list is empty"); + } + } + + private ObHbaseCell getFirstLegacyCell() { + if (cells.isEmpty()) { + throw new IllegalStateException("cell list is empty"); + } + return cells.get(0); + } + + private void ensureLegacyMode() { + if (hasCompactCells()) { + throw new IllegalStateException("legacy and compact cells cannot be mixed"); + } + } + + private void ensureCompactMode() { + if (!cells.isEmpty()) { + throw new IllegalStateException("legacy and compact cells cannot be mixed"); + } + } + + private void ensureCompactCellCapacity(int requiredCapacity) { + int oldCapacity = compactQualifierArrays == null ? 0 : compactQualifierArrays.length; + if (requiredCapacity <= oldCapacity) { + return; + } + int newCapacity = expandedCapacity(oldCapacity, requiredCapacity); + compactQualifierArrays = copyOf(compactQualifierArrays, newCapacity); + compactQualifierOffsets = copyOf(compactQualifierOffsets, newCapacity); + compactQualifierLengths = copyOf(compactQualifierLengths, newCapacity); + compactTimestamps = copyOf(compactTimestamps, newCapacity); + compactValueArrays = copyOf(compactValueArrays, newCapacity); + compactValueOffsets = copyOf(compactValueOffsets, newCapacity); + compactValueLengths = copyOf(compactValueLengths, newCapacity); + compactCellContentSizes = copyOf(compactCellContentSizes, newCapacity); + } + + private void ensureCompactKeyRunCapacity(int requiredCapacity) { + int oldCapacity = compactKeyRunTtls == null ? 0 : compactKeyRunTtls.length; + if (requiredCapacity > oldCapacity) { + compactKeyRunTtls = copyOf(compactKeyRunTtls, + expandedCapacity(oldCapacity, requiredCapacity)); + } + } + + private static int expandedCapacity(int oldCapacity, int requiredCapacity) { + int candidate = oldCapacity == 0 ? 8 : oldCapacity + (oldCapacity >> 1); + if (candidate < requiredCapacity) { + candidate = requiredCapacity; + } + if (candidate < 0) { + throw new IllegalArgumentException("compact cell capacity overflow: " + + requiredCapacity); + } + return candidate; + } + + private static byte[][] copyOf(byte[][] source, int length) { + return source == null ? new byte[length][] : Arrays.copyOf(source, length); + } + + private static int[] copyOf(int[] source, int length) { + return source == null ? new int[length] : Arrays.copyOf(source, length); + } + + private static long[] copyOf(long[] source, int length) { + return source == null ? new long[length] : Arrays.copyOf(source, length); + } + + private static void checkSlice(byte[] bytes, int offset, int length, String name) { + if (bytes == null || offset < 0 || length < 0 || offset > bytes.length - length) { + throw new IllegalArgumentException("invalid " + name + " slice: offset=" + offset + + ", length=" + length + ", capacity=" + + (bytes == null ? -1 : bytes.length)); + } + } + } diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbasePutCellCodec.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbasePutCellCodec.java new file mode 100644 index 00000000..a2f75574 --- /dev/null +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbasePutCellCodec.java @@ -0,0 +1,113 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2021 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.protocol.payload.impl.execute; + +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObTableObjType; +import com.alipay.oceanbase.rpc.util.ObByteBuf; +import com.alipay.oceanbase.rpc.util.Serialization; + +/** Encodes the fixed HBase Put Q/T/V/(TTL) cell protocol without temporary ObObj objects. */ +public final class ObHbasePutCellCodec { + private static final long CELL_VERSION = 1; + + private ObHbasePutCellCodec() { + } + + public static long getCellPayloadSize(int qualifierLength, long timestamp, int valueLength, + boolean hasTtl, long ttl) { + long contentSize = getCellContentSize(qualifierLength, timestamp, valueLength, hasTtl, ttl); + return getCellPayloadSize(contentSize); + } + + public static void encodeCell(ObByteBuf buf, byte[] qualifier, int qualifierOffset, + int qualifierLength, long timestamp, byte[] value, + int valueOffset, int valueLength, boolean hasTtl, long ttl) { + checkSlice(qualifier, qualifierOffset, qualifierLength, "qualifier"); + checkSlice(value, valueOffset, valueLength, "value"); + long contentSize = getCellContentSize(qualifierLength, timestamp, valueLength, hasTtl, ttl); + encodeCell(buf, qualifier, qualifierOffset, qualifierLength, timestamp, value, valueOffset, + valueLength, hasTtl, ttl, contentSize); + } + + static void encodeCell(ObByteBuf buf, byte[] qualifier, int qualifierOffset, + int qualifierLength, long timestamp, byte[] value, int valueOffset, + int valueLength, boolean hasTtl, long ttl, long contentSize) { + Serialization.encodeObUniVersionHeader(buf, CELL_VERSION, contentSize); + Serialization.encodeVi64(buf, hasTtl ? 4 : 3); + encodeVarchar(buf, qualifier, qualifierOffset, qualifierLength); + encodeInt64(buf, timestamp); + encodeVarchar(buf, value, valueOffset, valueLength); + if (hasTtl) { + encodeInt64(buf, ttl); + } + } + + static long getCellContentSize(int qualifierLength, long timestamp, int valueLength, + boolean hasTtl, long ttl) { + checkLength(qualifierLength, "qualifier"); + checkLength(valueLength, "value"); + long size = Serialization.getNeedBytes(hasTtl ? 4 : 3); + size += getVarcharEncodedSize(qualifierLength); + size += ObTableObjType.DEFAULT_TABLE_OBJ_TYPE_SIZE + Serialization.getNeedBytes(timestamp); + size += getVarcharEncodedSize(valueLength); + if (hasTtl) { + size += ObTableObjType.DEFAULT_TABLE_OBJ_TYPE_SIZE + Serialization.getNeedBytes(ttl); + } + return size; + } + + static long getCellPayloadSize(long contentSize) { + return Serialization.getObUniVersionHeaderLength(CELL_VERSION, contentSize) + contentSize; + } + + private static long getVarcharEncodedSize(int length) { + return ObTableObjType.DEFAULT_TABLE_OBJ_TYPE_SIZE + Serialization.getNeedBytes(length) + + (long) length + 1; + } + + private static void encodeVarchar(ObByteBuf buf, byte[] bytes, int offset, int length) { + Serialization.encodeI8(buf, ObTableObjType.ObTableVarcharType.getValue()); + Serialization.encodeVi32(buf, length); + if (length > 0) { + buf.writeBytes(bytes, offset, length); + } + buf.writeByte((byte) 0); + } + + private static void encodeInt64(ObByteBuf buf, long value) { + Serialization.encodeI8(buf, ObTableObjType.ObTableInt64Type.getValue()); + Serialization.encodeVi64(buf, value); + } + + private static void checkSlice(byte[] bytes, int offset, int length, String name) { + if (bytes == null) { + throw new IllegalArgumentException(name + " bytes is null"); + } + if (offset < 0 || length < 0 || offset > bytes.length - length) { + throw new IllegalArgumentException("invalid " + name + " slice: offset=" + offset + + ", length=" + length + ", capacity=" + + bytes.length); + } + } + + private static void checkLength(int length, String name) { + if (length < 0) { + throw new IllegalArgumentException(name + " length is negative: " + length); + } + } +} diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpRequest.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpRequest.java index 90e1d925..feb052f5 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpRequest.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpRequest.java @@ -170,6 +170,10 @@ public void setHbaseOpType(OHOperationType hbaseOpType) { this.hbaseOpType = hbaseOpType; } + public OHOperationType getHbaseOpType() { + return hbaseOpType; + } + /** * Reset the cached payload content size and propagate to child objects */ diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpResult.java index 1305cad7..52e9913d 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpResult.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableLSOpResult.java @@ -32,6 +32,7 @@ public class ObTableLSOpResult extends AbstractPayload { private List results = new ArrayList(); private List propertiesColumnNames = new ArrayList<>(); + private boolean decodeHBaseKqtv = false; /* * Get pcode. @@ -96,6 +97,7 @@ public Object decode(ByteBuf buf) { for (int i = 0; i < len; i++) { ObTableTabletOpResult tabletOpResult = new ObTableTabletOpResult(); tabletOpResult.setPropertiesColumnNames(this.propertiesColumnNames); + tabletOpResult.setDecodeHBaseKqtv(decodeHBaseKqtv); tabletOpResult.decode(buf); results.add(tabletOpResult); } @@ -145,4 +147,8 @@ public void addAllResults(List results) { this.results.addAll(results); } + public void setDecodeHBaseKqtv(boolean decodeHBaseKqtv) { + this.decodeHBaseKqtv = decodeHBaseKqtv; + } + } diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpEntity.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpEntity.java index fedf1c08..53b83dfc 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpEntity.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpEntity.java @@ -18,10 +18,13 @@ package com.alipay.oceanbase.rpc.protocol.payload.impl.execute; import com.alipay.oceanbase.rpc.protocol.payload.AbstractPayload; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType; 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.ObTableSerialUtil; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObTableObjType; +import com.alipay.oceanbase.rpc.protocol.payload.impl.execute.query.ObHBaseCellBatch; import com.alipay.oceanbase.rpc.util.ObByteBuf; import com.alipay.oceanbase.rpc.util.Serialization; import io.netty.buffer.ByteBuf; @@ -41,6 +44,12 @@ public class ObTableSingleOpEntity extends AbstractPayload { private List aggPropertiesNames = new ArrayList<>(); private List propertiesValues = new ArrayList<>(); + private boolean decodeHBaseKqtv = false; + private ObHBaseCellBatch hbaseCellBatch = null; + + private static final int HBASE_KQTV_COLUMN_COUNT = 4; + private static final int HBASE_TIMESTAMP_COLUMN_INDEX = 2; + private boolean ignoreEncodePropertiesColumnNames = false; public ObTableSingleOpEntity() {} @@ -173,6 +182,7 @@ public Object decode(ByteBuf buf) { // 2. rowkey obobj rowkeyLen = (int) Serialization.decodeVi64(buf); + rowkey = new ArrayList<>(rowkeyLen); for (int i = 0; i < rowkeyLen; i++) { ObObj obj = new ObObj(); ObTableSerialUtil.decode(buf, obj); @@ -185,10 +195,16 @@ public Object decode(ByteBuf buf) { // 4. properties obobj propLen = (int) Serialization.decodeVi64(buf); - for (int i = 0; i < propLen; i++) { - ObObj obj = new ObObj(); - ObTableSerialUtil.decode(buf, obj); - propertiesValues.add(obj); + int propertiesStartIndex = buf.readerIndex(); + if (decodeHBaseKqtv && isHBaseKqtvSchema() + && (propLen & (HBASE_KQTV_COLUMN_COUNT - 1)) == 0) { + hbaseCellBatch = tryDecodeHBaseKqtvBatch(buf, propLen); + if (hbaseCellBatch == null) { + buf.readerIndex(propertiesStartIndex); + decodeGenericProperties(buf, propLen); + } + } else { + decodeGenericProperties(buf, propLen); } } catch (Exception e) { String errMsg = String.format("ObTableSingleOpEntity decode exception: rowkeyBitLen=%d, rowkeyLen=%d, propertiesBitLen=%d, propertiesLen=%d" @@ -199,6 +215,83 @@ public Object decode(ByteBuf buf) { return this; } + private void decodeGenericProperties(ByteBuf buf, int propLen) { + propertiesValues = new ArrayList<>(propLen); + for (int i = 0; i < propLen; i++) { + ObObj obj = new ObObj(); + ObTableSerialUtil.decode(buf, obj); + propertiesValues.add(obj); + } + } + + private boolean isHBaseKqtvSchema() { + return propertiesNames.size() == HBASE_KQTV_COLUMN_COUNT + && "K".equals(propertiesNames.get(0)) + && "Q".equals(propertiesNames.get(1)) + && "T".equals(propertiesNames.get(2)) + && "V".equals(propertiesNames.get(3)); + } + + private ObHBaseCellBatch tryDecodeHBaseKqtvBatch(ByteBuf buf, int propLen) { + int cellCount = propLen / HBASE_KQTV_COLUMN_COUNT; + ObHBaseCellBatch batch = new ObHBaseCellBatch(cellCount); + ObObjMeta binaryMeta = ObObjType.ObVarcharType.getDefaultObjMeta(); + binaryMeta.setCsType(ObCollationType.CS_TYPE_BINARY); + batch.setMeta(0, binaryMeta); + batch.setMeta(1, binaryMeta); + batch.setMeta(HBASE_TIMESTAMP_COLUMN_INDEX, + ObObjType.ObInt64Type.getDefaultObjMeta()); + batch.setMeta(3, binaryMeta); + + for (int cellIndex = 0; cellIndex < cellCount; cellIndex++) { + byte[] rowKey = decodeHBaseBinary(buf, cellIndex, 0); + if (rowKey == null) { + return null; + } + byte[] qualifier = decodeHBaseBinary(buf, cellIndex, 1); + if (qualifier == null) { + if (cellIndex == 0) { + return null; + } + throw unexpectedHBaseType(cellIndex, 1); + } + ObTableObjType timestampType = ObTableSerialUtil.decodeTableObjType(buf); + if (timestampType != ObTableObjType.ObTableInt64Type) { + if (cellIndex == 0) { + return null; + } + throw unexpectedHBaseType(cellIndex, HBASE_TIMESTAMP_COLUMN_INDEX); + } + long timestamp = Serialization.decodeVi64(buf); + byte[] value = decodeHBaseBinary(buf, cellIndex, 3); + if (value == null) { + if (cellIndex == 0) { + return null; + } + throw unexpectedHBaseType(cellIndex, 3); + } + batch.setCell(cellIndex, rowKey, qualifier, timestamp, value); + } + propertiesValues = Collections.emptyList(); + return batch; + } + + private byte[] decodeHBaseBinary(ByteBuf buf, int cellIndex, int columnIndex) { + ObTableObjType type = ObTableSerialUtil.decodeTableObjType(buf); + if (type != ObTableObjType.ObTableVarbinaryType) { + if (cellIndex == 0) { + return null; + } + throw unexpectedHBaseType(cellIndex, columnIndex); + } + return Serialization.decodeBinaryColumn(buf); + } + + private IllegalArgumentException unexpectedHBaseType(int cellIndex, int columnIndex) { + return new IllegalArgumentException("HBase Batch Get KQTV type changed at cell " + + cellIndex + ", column " + columnIndex); + } + /* * Get payload content size. */ @@ -445,4 +538,12 @@ public List getPropertiesValues() { return this.propertiesValues; } + public void setDecodeHBaseKqtv(boolean decodeHBaseKqtv) { + this.decodeHBaseKqtv = decodeHBaseKqtv; + } + + public ObHBaseCellBatch getHBaseCellBatch() { + return hbaseCellBatch; + } + } diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpResult.java index 2695e526..d4d620e6 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpResult.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpResult.java @@ -33,6 +33,7 @@ public class ObTableSingleOpResult extends AbstractPayload { private String executeHost; private int executePort; private List propertiesColumnNames = new ArrayList<>(); + private boolean decodeHBaseKqtv = false; /* * Get pcode. */ @@ -92,6 +93,8 @@ public Object decode(ByteBuf buf) { // 3. decode Entity this.entity.setAggPropertiesNames(propertiesColumnNames); + this.entity.setDecodeHBaseKqtv( + decodeHBaseKqtv && operationType == ObTableOperationType.GET); this.entity.decode(buf); // 4. decode affected rows @@ -158,6 +161,10 @@ public void setEntity(ObTableSingleOpEntity entity) { this.entity = entity; } + public void setDecodeHBaseKqtv(boolean decodeHBaseKqtv) { + this.decodeHBaseKqtv = decodeHBaseKqtv; + } + /* * Get affected rows. */ diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableTabletOpResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableTabletOpResult.java index 4585a3b0..6955dce9 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableTabletOpResult.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableTabletOpResult.java @@ -31,6 +31,7 @@ public class ObTableTabletOpResult extends AbstractPayload { private List results = new ArrayList(); private List propertiesColumnNames = new ArrayList<>(); + private boolean decodeHBaseKqtv = false; @Override @@ -81,6 +82,7 @@ public Object decode(ByteBuf buf) { for (int i = 0; i < len; i++) { ObTableSingleOpResult obTableSingleOpResult = new ObTableSingleOpResult(); obTableSingleOpResult.setPropertiesColumnNames(propertiesColumnNames); + obTableSingleOpResult.setDecodeHBaseKqtv(decodeHBaseKqtv); obTableSingleOpResult.decode(buf); results.add(obTableSingleOpResult); } @@ -126,5 +128,8 @@ public void addResult(ObTableSingleOpResult result) { public void setPropertiesColumnNames(List propertiesColumnNames) { this.propertiesColumnNames = propertiesColumnNames; } -} + public void setDecodeHBaseKqtv(boolean decodeHBaseKqtv) { + this.decodeHBaseKqtv = decodeHBaseKqtv; + } +} diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/QueryStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/QueryStreamResult.java index 0b3072f7..81cb0b7c 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/QueryStreamResult.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/QueryStreamResult.java @@ -33,5 +33,9 @@ public interface QueryStreamResult extends Lifecycle { LinkedList> getCacheRows(); + default int getCachedRowCount() { + return getCacheRows().size(); + } + List getCacheProperties(); } diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java index 4cc2447c..1f1b32cf 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java @@ -64,6 +64,10 @@ public abstract class AbstractQueryStreamResult extends AbstractPayload implemen protected Map> expectant; protected List cacheProperties = new LinkedList(); protected LinkedList> cacheRows = new LinkedList>(); + protected ArrayDeque cacheHBaseCellBatches = new ArrayDeque(); + protected ObHBaseCellBatch currentHBaseCellBatch; + protected int currentHBaseCellIndex = -1; + protected boolean currentHBaseCell; private LinkedList, ObTableQueryResult>> partitionLastResult = new LinkedList, ObTableQueryResult>>(); private ObReadConsistency readConsistency = ObReadConsistency.STRONG; // ObRowKey objs: [startKey, MIN_OBJECT, MIN_OBJECT] @@ -312,7 +316,7 @@ public boolean next() throws Exception { lock.lock(); try { // firstly, refer to the cache - if (!cacheRows.isEmpty()) { + if (hasCachedRows()) { nextRow(); return true; } @@ -341,10 +345,10 @@ public boolean next() throws Exception { Map.Entry> entry = it.next(); referPartition.add(entry); try { - // Mark the refer partition + // Mark the refer partition referPartition.add(entry); - // Try accessing the new partition + // Try accessing the new partition ObTableQueryResult tableQueryResult = (ObTableQueryResult) referToNewPartition(entry .getValue()); @@ -360,7 +364,7 @@ public boolean next() throws Exception { if (shouldRetry(e)) { // TODO: need to skip over the partitions that have been scanned setExpectant(refreshPartition(tableQuery, tableName)); - // Reset the iterator to start over + // Reset the iterator to start over it = expectant.entrySet().iterator(); referPartition.clear(); // Clear the referPartition if needed } else { @@ -473,10 +477,35 @@ protected Map> buildFirstPartitions(ObTableClie protected void nextRow() { rowIndex = rowIndex + 1; - row = cacheRows.poll(); - if (row != null) { + row = null; + currentStartKey = null; + currentHBaseCell = false; + + if (!cacheRows.isEmpty()) { + row = cacheRows.poll(); currentStartKey = row; + return; + } + + while (currentHBaseCellBatch == null + || currentHBaseCellIndex + 1 >= currentHBaseCellBatch.size()) { + currentHBaseCellBatch = cacheHBaseCellBatches.pollFirst(); + currentHBaseCellIndex = -1; + if (currentHBaseCellBatch == null) { + return; } + } + + currentHBaseCellIndex++; + currentHBaseCell = true; + } + + protected boolean hasCachedRows() { + if (!cacheRows.isEmpty() || !cacheHBaseCellBatches.isEmpty()) { + return true; + } + return currentHBaseCellBatch != null + && currentHBaseCellIndex + 1 < currentHBaseCellBatch.size(); } protected void checkStatus() throws IllegalStateException { @@ -562,12 +591,22 @@ protected abstract Map> refreshPartition(ObTabl private void resetCachedResult() { cacheRows.clear(); + cacheHBaseCellBatches.clear(); + currentHBaseCellBatch = null; + currentHBaseCellIndex = -1; + currentHBaseCell = false; cacheProperties.clear(); partitionLastResult.clear(); } protected void cacheResultRows(ObTableQueryResult tableQueryResult) { - cacheRows.addAll(tableQueryResult.getPropertiesRows()); + ObHBaseCellBatch batch = tableQueryResult.getHBaseCellBatch(); + if (batch != null && cacheRows.isEmpty()) { + cacheHBaseCellBatches.addLast(batch); + } else { + materializeCachedHBaseCells(); + cacheRows.addAll(tableQueryResult.getPropertiesRows()); + } cacheProperties = tableQueryResult.getPropertiesNames(); } @@ -581,8 +620,7 @@ protected void cacheStreamNext(ObPair partIdWithObTable, } private void cacheResultRows(ObTableQueryAsyncResult tableQueryAsyncResult) { - cacheRows.addAll(tableQueryAsyncResult.getAffectedEntity().getPropertiesRows()); - cacheProperties = tableQueryAsyncResult.getAffectedEntity().getPropertiesNames(); + cacheResultRows(tableQueryAsyncResult.getAffectedEntity()); } protected void cacheStreamNext(ObPair partIdWithObTable, @@ -606,9 +644,92 @@ public List getRow() { if (rowIndex == -1) { throw new IllegalStateException("before result set start"); } + if (row == null && currentHBaseCell) { + row = currentHBaseCellBatch.materializeRow(currentHBaseCellIndex); + currentStartKey = row; + } return row; } + public boolean isCurrentHBaseCell() { + return currentHBaseCell; + } + + public ObHBaseCellBatch getCurrentHBaseCellBatch() { + if (!currentHBaseCell) { + throw new IllegalStateException("current row is not a compact HBase cell"); + } + return currentHBaseCellBatch; + } + + public int getCurrentHBaseCellIndex() { + if (!currentHBaseCell) { + throw new IllegalStateException("current row is not a compact HBase cell"); + } + return currentHBaseCellIndex; + } + + /** + * Drain the compact cells for the current HBase row without materializing them as ObObj rows. + * The method only consumes already cached batches and leaves the first cell of the next row + * unread. Fetching another stream page remains the responsibility of {@link #next()}. + */ + public ObHBaseCellRow drainCurrentHBaseRow() { + lock.lock(); + try { + checkStatus(); + if (!currentHBaseCell || currentHBaseCellBatch == null || currentHBaseCellIndex < 0) { + throw new IllegalStateException("current row is not a compact HBase cell"); + } + + byte[] rowKey = currentHBaseCellBatch.getRowKey(currentHBaseCellIndex); + ObHBaseCellRow hbaseRow = new ObHBaseCellRow(rowKey); + + while (true) { + ObHBaseCellBatch batch = currentHBaseCellBatch; + int fromIndex = currentHBaseCellIndex; + int toIndex = fromIndex + 1; + while (toIndex < batch.size() && Arrays.equals(rowKey, batch.getRowKey(toIndex))) { + toIndex++; + } + + hbaseRow.addSlice(batch, fromIndex, toIndex); + currentHBaseCellIndex = toIndex - 1; + + if (toIndex < batch.size()) { + break; + } + + ObHBaseCellBatch nextBatch = cacheHBaseCellBatches.peekFirst(); + while (nextBatch != null && nextBatch.size() == 0) { + cacheHBaseCellBatches.pollFirst(); + nextBatch = cacheHBaseCellBatches.peekFirst(); + } + if (nextBatch == null || !Arrays.equals(rowKey, nextBatch.getRowKey(0))) { + break; + } + + currentHBaseCellBatch = cacheHBaseCellBatches.pollFirst(); + currentHBaseCellIndex = 0; + } + + rowIndex += hbaseRow.getCellCount() - 1; + row = null; + currentStartKey = null; + currentHBaseCell = true; + return hbaseRow; + } finally { + lock.unlock(); + } + } + + protected List getCurrentStartKeyForRetry() { + if (currentStartKey == null && currentHBaseCell) { + currentStartKey = currentHBaseCellBatch.materializeRow(currentHBaseCellIndex); + } + return currentStartKey; + } + /* * Get row index. */ @@ -694,9 +815,39 @@ public List getCacheProperties() { * Get cache rows. */ public LinkedList> getCacheRows() { + materializeCachedHBaseCells(); return cacheRows; } + @Override + public int getCachedRowCount() { + int cachedRowCount = cacheRows.size(); + if (currentHBaseCellBatch != null) { + cachedRowCount += currentHBaseCellBatch.size() - currentHBaseCellIndex - 1; + } + for (ObHBaseCellBatch batch : cacheHBaseCellBatches) { + cachedRowCount += batch.size(); + } + return cachedRowCount; + } + + private void materializeCachedHBaseCells() { + if (currentHBaseCellBatch != null) { + if (currentHBaseCell && row == null) { + row = currentHBaseCellBatch.materializeRow(currentHBaseCellIndex); + currentStartKey = row; + } + cacheRows.addAll(currentHBaseCellBatch.materializeRows(currentHBaseCellIndex + 1)); + currentHBaseCellBatch = null; + currentHBaseCellIndex = -1; + currentHBaseCell = false; + } + ObHBaseCellBatch batch; + while ((batch = cacheHBaseCellBatches.pollFirst()) != null) { + cacheRows.addAll(batch.materializeRows(0)); + } + } + public LinkedList, ObTableQueryResult>> getPartitionLastResult() { return partitionLastResult; } diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHBaseCellBatch.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHBaseCellBatch.java new file mode 100644 index 00000000..e35c22d0 --- /dev/null +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHBaseCellBatch.java @@ -0,0 +1,104 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2021 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.protocol.payload.impl.execute.query; + +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObjMeta; + +import java.util.ArrayList; +import java.util.List; + +/** + * Compact representation of one HBase K/Q/T/V response page. Binary values own their byte arrays + * and remain valid after the response buffer is released. + */ +public final class ObHBaseCellBatch { + + private static final int HBASE_KQTV_COLUMN_COUNT = 4; + + private final byte[][] rowKeys; + private final byte[][] qualifiers; + private final long[] timestamps; + private final byte[][] values; + private final ObObjMeta[] metas = new ObObjMeta[HBASE_KQTV_COLUMN_COUNT]; + + public ObHBaseCellBatch(int size) { + if (size < 0) { + throw new IllegalArgumentException("negative HBase cell batch size: " + size); + } + this.rowKeys = new byte[size][]; + this.qualifiers = new byte[size][]; + this.timestamps = new long[size]; + this.values = new byte[size][]; + } + + public void setMeta(int columnIndex, ObObjMeta meta) { + metas[columnIndex] = meta; + } + + ObObjMeta getMeta(int columnIndex) { + return metas[columnIndex]; + } + + public void setCell(int index, byte[] rowKey, byte[] qualifier, long timestamp, byte[] value) { + rowKeys[index] = rowKey; + qualifiers[index] = qualifier; + timestamps[index] = timestamp; + values[index] = value; + } + + public int size() { + return timestamps.length; + } + + public byte[] getRowKey(int index) { + return rowKeys[index]; + } + + public byte[] getQualifier(int index) { + return qualifiers[index]; + } + + public long getTimestamp(int index) { + return timestamps[index]; + } + + public byte[] getValue(int index) { + return values[index]; + } + + List materializeRow(int index) { + List row = new ArrayList(HBASE_KQTV_COLUMN_COUNT); + row.add(new ObObj(metas[0], rowKeys[index])); + row.add(new ObObj(metas[1], qualifiers[index])); + row.add(new ObObj(metas[2], timestamps[index])); + row.add(new ObObj(metas[3], values[index])); + return row; + } + + List> materializeRows(int fromIndex) { + if (fromIndex < 0 || fromIndex > size()) { + throw new IndexOutOfBoundsException("invalid materialize start index: " + fromIndex); + } + List> rows = new ArrayList>(size() - fromIndex); + for (int i = fromIndex; i < size(); i++) { + rows.add(materializeRow(i)); + } + return rows; + } +} diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHBaseCellRow.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHBaseCellRow.java new file mode 100644 index 00000000..06346cc9 --- /dev/null +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHBaseCellRow.java @@ -0,0 +1,88 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2021 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.protocol.payload.impl.execute.query; + +import java.util.ArrayList; +import java.util.List; + +/** + * Read-only view of the compact cells that belong to one HBase row. A row may span multiple + * decoded batches, so the view keeps batch ranges instead of copying fields into per-cell DTOs. + */ +public final class ObHBaseCellRow { + + private final byte[] rowKey; + private final List slices = new ArrayList(1); + private int cellCount; + + ObHBaseCellRow(byte[] rowKey) { + if (rowKey == null) { + throw new NullPointerException("rowKey is null"); + } + this.rowKey = rowKey; + } + + void addSlice(ObHBaseCellBatch batch, int fromIndex, int toIndex) { + if (batch == null) { + throw new NullPointerException("batch is null"); + } + if (fromIndex < 0 || toIndex <= fromIndex || toIndex > batch.size()) { + throw new IndexOutOfBoundsException("invalid HBase cell batch slice [" + fromIndex + + ", " + toIndex + ") for batch size " + + batch.size()); + } + slices.add(new ObHBaseCellBatchSlice(batch, fromIndex, toIndex)); + cellCount += toIndex - fromIndex; + } + + public byte[] getRowKey() { + return rowKey; + } + + public int getCellCount() { + return cellCount; + } + + public int getSliceCount() { + return slices.size(); + } + + public ObHBaseCellBatch getBatch(int sliceIndex) { + return slices.get(sliceIndex).batch; + } + + public int getFromIndex(int sliceIndex) { + return slices.get(sliceIndex).fromIndex; + } + + public int getToIndex(int sliceIndex) { + return slices.get(sliceIndex).toIndex; + } + + private static final class ObHBaseCellBatchSlice { + private final ObHBaseCellBatch batch; + private final int fromIndex; + private final int toIndex; + + private ObHBaseCellBatchSlice(ObHBaseCellBatch batch, int fromIndex, int toIndex) { + this.batch = batch; + this.fromIndex = fromIndex; + this.toIndex = toIndex; + } + } +} diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHTableFilter.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHTableFilter.java index f5e43ba0..da53e97c 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHTableFilter.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObHTableFilter.java @@ -303,9 +303,7 @@ public byte[] getFilterString() { * Set filter string. */ public void setFilterString(byte[] filterString) { - if (this.filterString == null) { - this.filterString = new ObBytesString(); - } - this.filterString.bytes = filterString; + this.filterString = filterString == null ? null : new ObBytesString(filterString); + resetPayloadContentSize(); } } diff --git a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryResult.java b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryResult.java index 689315b4..f3af044d 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryResult.java +++ b/src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryResult.java @@ -20,23 +20,35 @@ import com.alipay.oceanbase.rpc.protocol.packet.ObRpcPacketHeader; import com.alipay.oceanbase.rpc.protocol.payload.AbstractPayload; import com.alipay.oceanbase.rpc.protocol.payload.Pcodes; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType; 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.util.Serialization; import io.netty.buffer.ByteBuf; -import java.util.LinkedList; +import java.util.ArrayList; import java.util.List; public class ObTableQueryResult extends AbstractPayload { + private static final String HBASE_COL_K = "K"; + private static final String HBASE_COL_Q = "Q"; + private static final String HBASE_COL_T = "T"; + private static final String HBASE_COL_V = "V"; + private static final int HBASE_KQTV_COL_COUNT = 4; + private static final int HBASE_TS_COL_INDEX = 2; + private static final int OB_OBJ_META_SIZE = 4; + private ObRpcPacketHeader header; - private List propertiesNames = new LinkedList(); + private List propertiesNames = new ArrayList(); private long rowCount = 0; // decode to propertiesRows from dataBuffer directly // byte[] dataBuffer; // TODO 需要做成流式的，目前 OB 还不支持流式协议，单个 packet 大小过大会失败 - private List> propertiesRows = new LinkedList>(); + private List> propertiesRows = new ArrayList>(); + private ObHBaseCellBatch hbaseCellBatch; /* * Ob table query result. @@ -88,17 +100,33 @@ public byte[] encode() { System.arraycopy(Serialization.encodeVi64(rowCount), 0, bytes, idx, len); idx += len; - len = Serialization.getNeedBytes(propertiesRows.size()); - System.arraycopy(Serialization.encodeVi64(propertiesRows.size()), 0, bytes, idx, len); + int resultRowCount = getResultRowCount(); + len = Serialization.getNeedBytes(resultRowCount); + System.arraycopy(Serialization.encodeVi64(resultRowCount), 0, bytes, idx, len); idx += len; // ObDataBuffer - for (List row : propertiesRows) { + if (hbaseCellBatch != null) { + for (int rowIndex = 0; rowIndex < hbaseCellBatch.size(); rowIndex++) { + for (int columnIndex = 0; columnIndex < HBASE_KQTV_COL_COUNT; columnIndex++) { + ObObjMeta meta = hbaseCellBatch.getMeta(columnIndex); + byte[] metaBytes = meta.encode(); + System.arraycopy(metaBytes, 0, bytes, idx, metaBytes.length); + idx += metaBytes.length; + Object value = getHBaseCellValue(hbaseCellBatch, rowIndex, columnIndex); + byte[] valueBytes = meta.getType().encode(value); + System.arraycopy(valueBytes, 0, bytes, idx, valueBytes.length); + idx += valueBytes.length; + } + } + } else { + for (List row : propertiesRows) { for (ObObj obObj : row) { len = obObj.getEncodedSize(); System.arraycopy(obObj.encode(), 0, bytes, idx, len); idx += len; } } + } return bytes; } @@ -115,25 +143,130 @@ public Object decode(ByteBuf buf) { // this.header.decode(buf); // 2. decode itself - long size = Serialization.decodeVi64(buf); - for (int i = 0; i < size; i++) { - this.propertiesNames.add(Serialization.decodeVString(buf)); + int propertyCount = checkedCount(Serialization.decodeVi64(buf), "property count"); + List decodedPropertiesNames = new ArrayList(propertyCount); + for (int i = 0; i < propertyCount; i++) { + decodedPropertiesNames.add(Serialization.decodeVString(buf)); } - this.rowCount = Serialization.decodeVi64(buf); + long decodedRowCount = Serialization.decodeVi64(buf); + int resultRowCount = checkedCount(decodedRowCount, "row count"); // ObDataBuffer Serialization.decodeVi64(buf); // dataBuffer length + List> decodedPropertiesRows = new ArrayList>(0); + ObHBaseCellBatch decodedHBaseCellBatch = null; + if (resultRowCount > 0 && isHBaseKqtvSchema(decodedPropertiesNames)) { + int rowsStartIndex = buf.readerIndex(); + decodedHBaseCellBatch = tryDecodeHBaseKqtvBatch(buf, resultRowCount); + if (decodedHBaseCellBatch == null) { + buf.readerIndex(rowsStartIndex); + decodedPropertiesRows = decodeGenericRows(buf, resultRowCount, propertyCount); + } + } else { + decodedPropertiesRows = decodeGenericRows(buf, resultRowCount, propertyCount); + } + + this.propertiesNames = decodedPropertiesNames; + this.rowCount = decodedRowCount; + this.propertiesRows = decodedPropertiesRows; + this.hbaseCellBatch = decodedHBaseCellBatch; + + return this; + } + + private static List> decodeGenericRows(ByteBuf buf, int rowCount, int columnCount) { + List> rows = new ArrayList>(rowCount); for (int r = 0; r < rowCount; r++) { - List row = new LinkedList(); - for (int i = 0; i < propertiesNames.size(); i++) { + List row = new ArrayList(columnCount); + for (int i = 0; i < columnCount; i++) { ObObj obObj = new ObObj(); obObj.decode(buf); row.add(obObj); } - addPropertiesRow(row); + rows.add(row); } - return this; + return rows; + } + + private static boolean isHBaseKqtvSchema(List propertiesNames) { + return propertiesNames.size() == HBASE_KQTV_COL_COUNT + && HBASE_COL_K.equals(propertiesNames.get(0)) + && HBASE_COL_Q.equals(propertiesNames.get(1)) + && HBASE_COL_T.equals(propertiesNames.get(2)) + && HBASE_COL_V.equals(propertiesNames.get(3)); + } + + private static ObHBaseCellBatch tryDecodeHBaseKqtvBatch(ByteBuf buf, int rowCount) { + HBaseKqtvMetaCache metaCache = new HBaseKqtvMetaCache(); + ObHBaseCellBatch batch = new ObHBaseCellBatch(rowCount); + for (int r = 0; r < rowCount; r++) { + byte[] rowKey = null; + byte[] qualifier = null; + long timestamp = 0; + byte[] value = null; + for (int c = 0; c < HBASE_KQTV_COL_COUNT; c++) { + ObObjMeta meta; + if (r == 0) { + int metaBits = readMetaBits(buf); + meta = new ObObjMeta(); + meta.decode(buf); + if (!isExpectedHBaseMeta(c, meta)) { + return null; + } + metaCache.metaBits[c] = metaBits; + metaCache.metas[c] = meta; + batch.setMeta(c, meta); + } else { + int actualMetaBits = readMetaBits(buf); + if (actualMetaBits != metaCache.metaBits[c]) { + throw new IllegalStateException("HBase KQTV meta changed at row " + r + + ", column " + c); + } + buf.skipBytes(OB_OBJ_META_SIZE); + meta = metaCache.metas[c]; + } + + if (c == 0) { + rowKey = Serialization.decodeBinaryColumn(buf); + } else if (c == 1) { + qualifier = Serialization.decodeBinaryColumn(buf); + } else if (c == HBASE_TS_COL_INDEX) { + timestamp = Serialization.decodeVi64(buf); + } else { + value = Serialization.decodeBinaryColumn(buf); + } + } + batch.setCell(r, rowKey, qualifier, timestamp, value); + } + return batch; + } + + private static int readMetaBits(ByteBuf buf) { + if (buf.readableBytes() < OB_OBJ_META_SIZE) { + throw new IllegalArgumentException("not enough bytes to decode ObObjMeta"); + } + return buf.getInt(buf.readerIndex()); + } + + private static boolean isExpectedHBaseMeta(int columnIndex, ObObjMeta meta) { + if (columnIndex == HBASE_TS_COL_INDEX) { + return meta.getType() == ObObjType.ObInt64Type; + } + return meta.getType() == ObObjType.ObVarcharType + && meta.getCsType() == ObCollationType.CS_TYPE_BINARY; + } + + private static final class HBaseKqtvMetaCache { + private final int[] metaBits = new int[HBASE_KQTV_COL_COUNT]; + private final ObObjMeta[] metas = new ObObjMeta[HBASE_KQTV_COL_COUNT]; + } + + private static int checkedCount(long count, String fieldName) { + if (count < 0 || count > Integer.MAX_VALUE) { + throw new IllegalArgumentException("invalid " + fieldName + ": " + count); + } + return (int) count; } /* @@ -149,10 +282,20 @@ public long getPayloadContentSize() { } size += Serialization.getNeedBytes(rowCount); - size += Serialization.getNeedBytes(propertiesRows.size()); - for (List row : propertiesRows) { + size += Serialization.getNeedBytes(getResultRowCount()); + if (hbaseCellBatch != null) { + for (int rowIndex = 0; rowIndex < hbaseCellBatch.size(); rowIndex++) { + for (int columnIndex = 0; columnIndex < HBASE_KQTV_COL_COUNT; columnIndex++) { + ObObjMeta meta = hbaseCellBatch.getMeta(columnIndex); + Object value = getHBaseCellValue(hbaseCellBatch, rowIndex, columnIndex); + size += meta.getEncodedSize() + meta.getType().getEncodedSize(value); + } + } + } else { + for (List row : propertiesRows) { for (ObObj obObj : row) { size += obObj.getEncodedSize(); + } } } @@ -198,28 +341,61 @@ public void setRowCount(long rowCount) { * Get properties rows. */ public List> getPropertiesRows() { + if (hbaseCellBatch != null) { + propertiesRows = hbaseCellBatch.materializeRows(0); + hbaseCellBatch = null; + } return propertiesRows; } + public ObHBaseCellBatch getHBaseCellBatch() { + return hbaseCellBatch; + } + + public boolean hasHBaseCellBatch() { + return hbaseCellBatch != null; + } + /* * Set properties rows. */ public void setPropertiesRows(List> propertiesRows) { this.propertiesRows = propertiesRows; + this.hbaseCellBatch = null; } /* * Add properties row. */ public void addPropertiesRow(List propertiesRow) { - this.propertiesRows.add(propertiesRow); + getPropertiesRows().add(propertiesRow); } /* * Add all properties rows. */ public void addAllPropertiesRows(List> propertiesRows) { - this.propertiesRows.addAll(propertiesRows); + getPropertiesRows().addAll(propertiesRows); + } + + private int getResultRowCount() { + return hbaseCellBatch == null ? propertiesRows.size() : hbaseCellBatch.size(); + } + + private static Object getHBaseCellValue(ObHBaseCellBatch batch, int rowIndex, + int columnIndex) { + switch (columnIndex) { + case 0: + return batch.getRowKey(rowIndex); + case 1: + return batch.getQualifier(rowIndex); + case HBASE_TS_COL_INDEX: + return batch.getTimestamp(rowIndex); + case 3: + return batch.getValue(rowIndex); + default: + throw new IllegalArgumentException("invalid HBase KQTV column: " + columnIndex); + } } /* diff --git a/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java b/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java index da48edf5..ed16fbfe 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java +++ b/src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java @@ -107,8 +107,11 @@ public void init() throws Exception { protected void cacheResultRows(ObTableQueryAsyncResult tableQueryResult) { cacheRows.clear(); - cacheRows.addAll(tableQueryResult.getAffectedEntity().getPropertiesRows()); - cacheProperties = tableQueryResult.getAffectedEntity().getPropertiesNames(); + cacheHBaseCellBatches.clear(); + currentHBaseCellBatch = null; + currentHBaseCellIndex = -1; + currentHBaseCell = false; + super.cacheResultRows(tableQueryResult.getAffectedEntity()); } protected ObTableQueryAsyncResult referToNewPartition(ObPair partIdWithObTable) @@ -197,7 +200,7 @@ public boolean queryLastStreamResultInNext() throws Exception { && entry.getPartitionInfo().getLevel() == ObPartitionLevel.LEVEL_ONE && entry.getPartitionInfo().getFirstPartDesc().getPartFuncType().isRangePart()) { this.asyncRequest.getObTableQueryRequest().getTableQuery() - .adjustStartKey(currentStartKey); + .adjustStartKey(getCurrentStartKeyForRetry()); setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest() .getTableQuery(), realTableName)); setEnd(true); @@ -213,7 +216,7 @@ public boolean queryLastStreamResultInNext() throws Exception { if (isEnd()) { it.remove(); } - if (!cacheRows.isEmpty()) { + if (hasCachedRows()) { nextRow(); return true; } @@ -241,7 +244,7 @@ public boolean queryNewStreamResultInNext() throws Exception { && tableEntry.getPartitionInfo().getFirstPartDesc().getPartFuncType() .isRangePart()) { this.asyncRequest.getObTableQueryRequest().getTableQuery() - .adjustStartKey(currentStartKey); + .adjustStartKey(getCurrentStartKeyForRetry()); setExpectant(refreshPartition(this.asyncRequest.getObTableQueryRequest() .getTableQuery(), realTableName)); } else { @@ -267,7 +270,7 @@ public boolean queryNewStreamResultInNext() throws Exception { if (isEnd()) { it.remove(); } - if (!cacheRows.isEmpty()) { + if (hasCachedRows()) { hasNext = true; nextRow(); break; @@ -322,7 +325,7 @@ public boolean next() throws Exception { try { hasDoneRpc = false; // firstly, refer to the cache - if (!cacheRows.isEmpty()) { + if (hasCachedRows()) { nextRow(); return true; } @@ -337,7 +340,7 @@ public boolean next() throws Exception { // new server does not store the current session_id // only support range-partitioned table, check in server this.asyncRequest.getObTableQueryRequest().getTableQuery() - .adjustStartKey(currentStartKey); + .adjustStartKey(getCurrentStartKeyForRetry()); // just need to asjust startKey to anchor the correct position // no need to refresh partition id for session_id missing hasNext = queryNewStreamResultInNext(); diff --git a/src/main/java/com/alipay/oceanbase/rpc/stream/QueryResultSet.java b/src/main/java/com/alipay/oceanbase/rpc/stream/QueryResultSet.java index 29df3998..6f8404d5 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/stream/QueryResultSet.java +++ b/src/main/java/com/alipay/oceanbase/rpc/stream/QueryResultSet.java @@ -68,7 +68,7 @@ public Row getResultRow() { * Cache size. */ public int cacheSize() { - return queryStreamResult.getCacheRows().size(); + return queryStreamResult.getCachedRowCount(); } /* diff --git a/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java b/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java index 2bb90e28..fb4dd19b 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java +++ b/src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java @@ -618,7 +618,6 @@ public void partitionExecute(ObTableSingleOpResult[] results, tableLsOpRequest.setConsistencyLevel(ObReadConsistency.STRONG); } tableLsOpRequest.setHbaseOpType(hbaseOpType); - ObTableLSOpResult subLSOpResult; boolean needRefreshPartitionLocation = false; int tryTimes = 0; diff --git a/src/main/java/com/alipay/oceanbase/rpc/util/MonitorUtil.java b/src/main/java/com/alipay/oceanbase/rpc/util/MonitorUtil.java index 8e806306..c05fec7a 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/util/MonitorUtil.java +++ b/src/main/java/com/alipay/oceanbase/rpc/util/MonitorUtil.java @@ -261,7 +261,7 @@ private static String logMessage(String traceId, String database, String tableNa String argsValue = buildParamsString(params); - String res = String.valueOf(result.getCacheRows().size()); + String res = String.valueOf(result.getCachedRowCount()); StringBuilder stringBuilder = new StringBuilder(); stringBuilder.append(traceId).append(",").append(database).append(",").append(tableName) diff --git a/src/main/java/com/alipay/oceanbase/rpc/util/ObBytesString.java b/src/main/java/com/alipay/oceanbase/rpc/util/ObBytesString.java index 6619d671..f4283be2 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/util/ObBytesString.java +++ b/src/main/java/com/alipay/oceanbase/rpc/util/ObBytesString.java @@ -25,9 +25,13 @@ public class ObBytesString implements Comparable { public byte[] bytes; + public int offset; + private int stringLength; public ObBytesString() { this.bytes = new byte[0]; + this.offset = 0; + this.stringLength = 0; } public ObBytesString(byte[] bytes) { @@ -35,6 +39,25 @@ public ObBytesString(byte[] bytes) { throw new IllegalArgumentException("ObBytesString bytes can not be null "); } this.bytes = bytes; + this.offset = 0; + this.stringLength = bytes.length; + } + + /** + * View into {@code bytes[offset, offset+length)}. Encode copies only this region. + */ + public ObBytesString(byte[] bytes, int offset, int length) { + if (bytes == null) { + throw new IllegalArgumentException("ObBytesString bytes can not be null "); + } + if (offset < 0 || length < 0 || offset > bytes.length || length > bytes.length - offset) { + throw new IllegalArgumentException("ObBytesString invalid range offset=" + offset + + " length=" + length + " bytes.length=" + + bytes.length); + } + this.bytes = bytes; + this.offset = offset; + this.stringLength = length; } public ObBytesString(String str) { @@ -42,6 +65,8 @@ public ObBytesString(String str) { throw new IllegalArgumentException("ObBytesString str can not be null "); } this.bytes = Serialization.strToBytes(str); + this.offset = 0; + this.stringLength = this.bytes.length; } /** @@ -49,7 +74,7 @@ public ObBytesString(String str) { * @return length */ public int length() { - return bytes.length; + return stringLength; } /** @@ -64,7 +89,7 @@ public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) return false; ObBytesString that = (ObBytesString) o; - return compare(bytes, that.bytes) == 0; + return compare(this, that) == 0; } /** @@ -74,17 +99,17 @@ public boolean equals(Object o) { */ @Override public int compareTo(ObBytesString another) { - return compare(bytes, another.bytes); + return compare(this, another); } - private int compare(byte[] s, byte[] t) { - int len1 = s.length; - int len2 = t.length; + private static int compare(ObBytesString s, ObBytesString t) { + int len1 = s.stringLength; + int len2 = t.stringLength; int lim = Math.min(len1, len2); int k = 0; while (k < lim) { - byte c1 = s[k]; - byte c2 = t[k]; + byte c1 = s.bytes[s.offset + k]; + byte c2 = t.bytes[t.offset + k]; if (c1 != c2) { return c1 - c2; } @@ -95,21 +120,20 @@ private int compare(byte[] s, byte[] t) { public ObBytesString[] split(byte delim) { ArrayList list = new ArrayList<>(); - int start = 0; - for (int i = 0; i < bytes.length; ++i) { + int start = offset; + int end = offset + stringLength; + for (int i = offset; i < end; ++i) { if (bytes[i] == delim) { byte[] data = new byte[i - start]; System.arraycopy(bytes, start, data, 0, data.length); - ObBytesString str = new ObBytesString(data); - list.add(str); + list.add(new ObBytesString(data)); start = i + 1; } } - if (start < bytes.length) { - byte[] data = new byte[bytes.length - start]; + if (start < end) { + byte[] data = new byte[end - start]; System.arraycopy(bytes, start, data, 0, data.length); - ObBytesString str = new ObBytesString(data); - list.add(str); + list.add(new ObBytesString(data)); } return list.toArray(new ObBytesString[0]); } diff --git a/src/main/java/com/alipay/oceanbase/rpc/util/ObHashUtils.java b/src/main/java/com/alipay/oceanbase/rpc/util/ObHashUtils.java index e7b709ca..3e2fa458 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/util/ObHashUtils.java +++ b/src/main/java/com/alipay/oceanbase/rpc/util/ObHashUtils.java @@ -30,6 +30,7 @@ import java.sql.Date; import java.sql.Timestamp; import java.time.OffsetDateTime; +import java.util.Arrays; import static com.alipay.oceanbase.rpc.protocol.payload.impl.ObObjType.*; @@ -59,7 +60,12 @@ public static long varcharHash(Object varchar, ObCollationType collationType, lo } else if (varchar instanceof byte[]) { bytes = (byte[]) varchar; } else if (varchar instanceof ObBytesString) { - bytes = ((ObBytesString) varchar).bytes; + ObBytesString bytesString = (ObBytesString) varchar; + bytes = bytesString.bytes; + if (bytesString.offset != 0 || bytesString.length() != bytes.length) { + bytes = Arrays.copyOfRange(bytes, bytesString.offset, + bytesString.offset + bytesString.length()); + } } else { throw new IllegalArgumentException("varchar not supported , ObCollationType = " + collationType + " Object =" + varchar); diff --git a/src/main/java/com/alipay/oceanbase/rpc/util/ObPureCrc32C.java b/src/main/java/com/alipay/oceanbase/rpc/util/ObPureCrc32C.java index b7492489..2e0dbd18 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/util/ObPureCrc32C.java +++ b/src/main/java/com/alipay/oceanbase/rpc/util/ObPureCrc32C.java @@ -28,6 +28,8 @@ */ public class ObPureCrc32C { + private static final int SLICING_THRESHOLD = 64; + /** * crc32tab is generated by: * // bit-reversed poly 0x1EDC6F41 @@ -84,6 +86,8 @@ public class ObPureCrc32C { 0x8bdcb4b9L, 0x988c474dL, 0x6ae7c44eL, 0xbe2da0a5L, 0x4c4623a6L, 0x5f16d052L, 0xad7d5351L }; + private static final int[][] SLICING_TABLES = buildSlicingTables(); + /** * Calculate crc32 * @param buf input data buffer @@ -101,10 +105,63 @@ public static long calculate(byte[] buf) { * @return CRC32 value */ public static long calculate(byte[] buf, int offset, int length) { + if (length < SLICING_THRESHOLD) { + return calculateScalar(buf, offset, length); + } + return calculateSlicingBy8(buf, offset, length); + } + + static long calculateScalar(byte[] buf, int offset, int length) { long crc = 0; for (int i = offset; length > 0; length--) { - crc = (CRC32_TAB[(int) ((crc ^ buf[i++]) & 0xff)] ^ (crc >> 8)); + crc = CRC32_TAB[(int) ((crc ^ buf[i++]) & 0xff)] ^ (crc >> 8); + } + return crc; + } + + static long calculateSlicingBy8(byte[] buf, int offset, int length) { + int crc = 0; + int index = offset; + int remaining = length; + while (remaining >= 8) { + int value = crc; + crc = SLICING_TABLES[7][(value ^ buf[index]) & 0xff] + ^ SLICING_TABLES[6][((value >>> 8) ^ buf[index + 1]) & 0xff] + ^ SLICING_TABLES[5][((value >>> 16) ^ buf[index + 2]) & 0xff] + ^ SLICING_TABLES[4][((value >>> 24) ^ buf[index + 3]) & 0xff] + ^ SLICING_TABLES[3][buf[index + 4] & 0xff] + ^ SLICING_TABLES[2][buf[index + 5] & 0xff] + ^ SLICING_TABLES[1][buf[index + 6] & 0xff] + ^ SLICING_TABLES[0][buf[index + 7] & 0xff]; + index += 8; + remaining -= 8; + } + return updateTail(crc, buf, index, remaining) & 0xffffffffL; + } + + private static int updateTail(int crc, byte[] buf, int offset, int length) { + for (int i = offset; length > 0; length--) { + crc = updateByte(crc, buf[i++]); } return crc; } + + private static int updateByte(int crc, byte value) { + return (int) CRC32_TAB[(crc ^ value) & 0xff] ^ (crc >>> 8); + } + + private static int[][] buildSlicingTables() { + int[][] tables = new int[8][256]; + for (int i = 0; i < 256; i++) { + tables[0][i] = (int) CRC32_TAB[i]; + } + for (int slice = 1; slice < tables.length; slice++) { + for (int i = 0; i < 256; i++) { + int previous = tables[slice - 1][i]; + tables[slice][i] = tables[0][previous & 0xff] ^ (previous >>> 8); + } + } + return tables; + } + } diff --git a/src/main/java/com/alipay/oceanbase/rpc/util/Serialization.java b/src/main/java/com/alipay/oceanbase/rpc/util/Serialization.java index c769264e..d13ddc61 100644 --- a/src/main/java/com/alipay/oceanbase/rpc/util/Serialization.java +++ b/src/main/java/com/alipay/oceanbase/rpc/util/Serialization.java @@ -750,16 +750,16 @@ public static byte[] encodeBytesString(ObBytesString str) { str = new ObBytesString(new byte[0]); } byte[] data = str.bytes; - int dataLen = data.length; + int dataLen = str.length(); + int dataOff = str.offset; int strLen = getNeedBytes(dataLen); byte[] ret = new byte[strLen + dataLen + 1]; int index = 0; for (byte b : encodeVi32(dataLen)) { ret[index++] = b; } - for (byte b : data) { - ret[index++] = b; - } + System.arraycopy(data, dataOff, ret, index, dataLen); + index += dataLen; ret[index] = 0; return ret; } @@ -774,8 +774,8 @@ public static void encodeBytesString(ObByteBuf buf, ObBytesString str) { throw new NullPointerException(); int dataLen = (str == null ? 0 : str.length()); encodeVi32(buf, dataLen); - if (str != null) { - buf.writeBytes(str.bytes); + if (str != null && dataLen > 0) { + buf.writeBytes(str.bytes, str.offset, dataLen); } buf.writeByte((byte) 0x00); } @@ -892,6 +892,27 @@ public static ObBytesString decodeBytesString(ByteBuf buf) { return new ObBytesString(content); } + /** + * Decode a binary column directly to byte[] without creating an ObBytesString. + * @param buf input data + * @return decoded binary column + */ + public static byte[] decodeBinaryColumn(ByteBuf buf) { + int dataLen = decodeVi32(buf); + if (dataLen < 0 || dataLen > buf.readableBytes() - 1) { + throw new IllegalArgumentException("invalid binary column length: " + dataLen + + ", readable bytes: " + buf.readableBytes()); + } + + byte[] content = new byte[dataLen]; + buf.readBytes(content); + byte terminator = buf.readByte(); + if (terminator != 0) { + throw new IllegalArgumentException("invalid binary column terminator: " + terminator); + } + return content; + } + /** * Decode bytes * @param buf input data diff --git a/src/test/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemotingChecksumTest.java b/src/test/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemotingChecksumTest.java new file mode 100644 index 00000000..d19e013e --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemotingChecksumTest.java @@ -0,0 +1,69 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2021 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.bolt.transport; + +import com.alipay.oceanbase.rpc.bolt.protocol.ObTablePacket; +import com.alipay.oceanbase.rpc.protocol.packet.ObRpcPacketHeader; +import com.alipay.oceanbase.rpc.protocol.payload.AbstractPayload; +import com.alipay.oceanbase.rpc.protocol.payload.Pcodes; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import org.junit.Assert; +import org.junit.Test; + +public class ObTableRemotingChecksumTest { + + @Test + public void testRequestChecksumRemainsEnabled() { + byte[] payloadContent = new byte[] { 5, 6, 7, 8 }; + FixedPayload payload = new FixedPayload(payloadContent); + ObTablePacket packet = new ObPacketFactory(false).createRequestCommand(payload); + ByteBuf packetContent = Unpooled.wrappedBuffer(packet.getPacketContent()); + try { + ObRpcPacketHeader header = new ObRpcPacketHeader(); + header.decode(packetContent); + + Assert.assertEquals(0x8762fcd6L, header.getChecksum()); + } finally { + packetContent.release(); + } + } + + private static class FixedPayload extends AbstractPayload { + private final byte[] content; + + FixedPayload(byte[] content) { + this.content = content; + } + + @Override + public int getPcode() { + return Pcodes.OB_TABLE_API_LOGIN; + } + + @Override + public byte[] encode() { + return content; + } + + @Override + public long getPayloadContentSize() { + return content.length; + } + } +} diff --git a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjTest.java b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjTest.java index 7bba92f6..27c412de 100644 --- a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjTest.java +++ b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObObjTest.java @@ -25,7 +25,10 @@ import static com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType.CS_TYPE_BINARY; import static com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType.CS_TYPE_UTF8MB4_GENERAL_CI; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; public class ObObjTest { @@ -128,4 +131,39 @@ public void test_text_type() { assertEquals(testBytes, comparable); } + @Test + public void test_text_type_honors_bytes_string_range() { + byte[] expected = "test".getBytes(); + ObBytesString slice = new ObBytesString("__test__".getBytes(), 2, 4); + + assertArrayEquals(expected, ObObjType.ObVarcharType.parseToBytes(slice, CS_TYPE_BINARY)); + assertArrayEquals(expected, + ObObjType.ObVarcharType.parseToBytes(slice, CS_TYPE_UTF8MB4_GENERAL_CI)); + assertEquals("test", + ObObjType.ObVarcharType.parseToComparable(slice, CS_TYPE_UTF8MB4_GENERAL_CI)); + assertSame(slice, ObObjType.ObVarcharType.parseToComparable(slice, CS_TYPE_BINARY)); + } + + @Test + public void test_obj_type_value_lookup() { + ObObjType[] expectedTypes = new ObObjType[128]; + for (ObObjType type : ObObjType.values()) { + assertSame(type, ObObjType.valueOf(type.getValue())); + if (type.getValue() < expectedTypes.length) { + expectedTypes[type.getValue()] = type; + } + } + for (int value = 0; value < expectedTypes.length; value++) { + assertSame(expectedTypes[value], ObObjType.valueOf(value)); + } + + assertNull(ObObjType.valueOf(-1)); + assertNull(ObObjType.valueOf(-128)); + assertNull(ObObjType.valueOf(128)); + assertNull(ObObjType.valueOf(129)); + assertNull(ObObjType.valueOf(255)); + assertNull(ObObjType.valueOf(Integer.MIN_VALUE)); + assertNull(ObObjType.valueOf(Integer.MAX_VALUE)); + } + } diff --git a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableObjTypeTest.java b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableObjTypeTest.java new file mode 100644 index 00000000..a8769184 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObTableObjTypeTest.java @@ -0,0 +1,75 @@ +/*- + * #%L + * com.oceanbase:obkv-table-client + * %% + * Copyright (C) 2021 - 2026 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.protocol.payload.impl; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.Map; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import org.junit.Test; + +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; + +public class ObTableObjTypeTest { + @Test + public void resolvesEveryDeclaredProtocolType() { + for (ObTableObjType type : ObTableObjType.values()) { + assertSame(type, ObTableObjType.valueOf(type.getValue() & 0xFF)); + } + } + + @Test + public void preservesUnknownAndSparseIdBehavior() { + int[] unknown = { Integer.MIN_VALUE, -1, 13, 14, 15, 16, 27, 127, 128, 255, 256, + Integer.MAX_VALUE }; + for (int value : unknown) { + assertNull("value=" + value, ObTableObjType.valueOf(value)); + } + } + + @SuppressWarnings("unchecked") + @Test + public void resolvesIdsOutsideFastLookupArrayThroughFallbackMap() throws Exception { + Method registerLookup = ObTableObjType.class.getDeclaredMethod("registerLookup", int.class, + ObTableObjType.class); + registerLookup.setAccessible(true); + Field overflowLookup = ObTableObjType.class.getDeclaredField("OVERFLOW_VALUE_LOOKUP"); + overflowLookup.setAccessible(true); + Map overflow = (Map) overflowLookup + .get(null); + + int[] overflowIds = { 128, 255, 256 }; + for (int overflowId : overflowIds) { + try { + registerLookup.invoke(null, overflowId, ObTableObjType.ObTableInvalidType); + assertSame(ObTableObjType.ObTableInvalidType, ObTableObjType.valueOf(overflowId)); + if (overflowId <= 0xFF) { + ByteBuf encodedType = Unpooled.wrappedBuffer(new byte[] { (byte) overflowId }); + assertSame(ObTableObjType.ObTableInvalidType, + ObTableSerialUtil.decodeTableObjType(encodedType)); + } + } finally { + overflow.remove(overflowId); + } + assertNull(ObTableObjType.valueOf(overflowId)); + } + } +} diff --git a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbasePutCellCodecTest.java b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbasePutCellCodecTest.java new file mode 100644 index 00000000..1ec9844a --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObHbasePutCellCodecTest.java @@ -0,0 +1,165 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2021 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.protocol.payload.impl.execute; + +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObObj; +import com.alipay.oceanbase.rpc.util.ObByteBuf; +import com.alipay.oceanbase.rpc.util.ObBytesString; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Random; + +public class ObHbasePutCellCodecTest { + + @Test + public void testCompactCellMatchesLegacyAtBoundaries() { + int[] lengths = new int[] { 0, 1, 7, 16, 63, 64, 127, 128, 255, 500 }; + Random random = new Random(20260807L); + for (int length : lengths) { + byte[] qualifier = new byte[length + 9]; + byte[] value = new byte[length + 13]; + random.nextBytes(qualifier); + random.nextBytes(value); + assertCellEquals(qualifier, 4, length, -123456789L, value, 6, length, false, + Long.MAX_VALUE); + assertCellEquals(qualifier, 4, length, Long.MIN_VALUE + length, value, 6, length, true, + 86400000L + length); + } + } + + @Test + public void testCompactCfRowsMatchesLegacyForMultipleRuns() { + ObHbaseCfRows legacy = new ObHbaseCfRows(); + ObHbaseCfRows compact = new ObHbaseCfRows(); + legacy.setRealTableName("test$table$family"); + compact.setRealTableName("test$table$family"); + + appendRun(legacy, compact, 0, Long.MAX_VALUE, 3, 11); + appendRun(legacy, compact, 2, 3600000L, 5, 37); + + Assert.assertEquals(legacy.getPayloadContentSize(), compact.getPayloadContentSize()); + Assert.assertEquals(legacy.getPayloadSize(), compact.getPayloadSize()); + Assert.assertArrayEquals(legacy.encode(), compact.encode()); + } + + @Test + public void testCompactStateValidation() { + ObHbaseCfRows rows = new ObHbaseCfRows(); + rows.setRealTableName("t$f"); + rows.beginCompactKeyCells(0, 1, Long.MAX_VALUE); + try { + rows.getPayloadContentSize(); + Assert.fail("incomplete compact run must fail"); + } catch (IllegalStateException expected) { + Assert.assertTrue(expected.getMessage().contains("incomplete")); + } + + try { + rows.appendCell(new ObHbaseCell(false)); + Assert.fail("compact and legacy cells must not be mixed"); + } catch (IllegalStateException expected) { + Assert.assertTrue(expected.getMessage().contains("cannot be mixed")); + } + } + + @Test + public void testCompactRoutingValuesMatchLegacy() { + byte[] qualifier = new byte[] { 9, 8, 7, 6, 5 }; + byte[] value = new byte[] { 4, 3, 2, 1 }; + long timestamp = -20260807L; + ObHbaseCfRows legacy = new ObHbaseCfRows(); + legacy.beginKeyCells(0, 1); + legacy.appendCell(newLegacyCell(qualifier, 1, 3, timestamp, value, 0, value.length, + Long.MAX_VALUE)); + ObHbaseCfRows compact = new ObHbaseCfRows(); + compact.beginCompactKeyCells(0, 1, Long.MAX_VALUE); + compact.appendCompactCell(qualifier, 1, 3, timestamp, value, 0, value.length); + + Assert.assertEquals(legacy.getFirstCellQualifierValue(), + compact.getFirstCellQualifierValue()); + Assert.assertEquals(legacy.getFirstCellTimestampValue(), + compact.getFirstCellTimestampValue()); + } + + @Test + public void testInvalidSliceRejected() { + ObHbaseCfRows rows = new ObHbaseCfRows(); + rows.beginCompactKeyCells(0, 1, Long.MAX_VALUE); + try { + rows.appendCompactCell(new byte[4], 3, 2, -1, new byte[1], 0, 1); + Assert.fail("invalid qualifier slice must fail"); + } catch (IllegalArgumentException expected) { + Assert.assertTrue(expected.getMessage().contains("qualifier")); + } + } + + private static void appendRun(ObHbaseCfRows legacy, ObHbaseCfRows compact, int keyIndex, + long ttl, int cellCount, int seed) { + legacy.beginKeyCells(keyIndex, cellCount); + compact.reserveAdditionalCompactCells(cellCount); + compact.beginCompactKeyCells(keyIndex, cellCount, ttl); + Random random = new Random(seed); + for (int i = 0; i < cellCount; i++) { + int qualifierLength = i == 0 ? 0 : 7 + i * 31; + int valueLength = 20 + i * 73; + byte[] qualifier = new byte[qualifierLength + 5]; + byte[] value = new byte[valueLength + 9]; + random.nextBytes(qualifier); + random.nextBytes(value); + long timestamp = -(1000L + i); + + ObHbaseCell legacyCell = newLegacyCell(qualifier, 2, qualifierLength, timestamp, value, + 4, valueLength, ttl); + legacy.appendCell(legacyCell); + compact.appendCompactCell(qualifier, 2, qualifierLength, timestamp, value, 4, + valueLength); + } + } + + private static void assertCellEquals(byte[] qualifier, int qualifierOffset, + int qualifierLength, long timestamp, byte[] value, + int valueOffset, int valueLength, boolean hasTtl, long ttl) { + ObHbaseCell legacy = newLegacyCell(qualifier, qualifierOffset, qualifierLength, timestamp, + value, valueOffset, valueLength, hasTtl ? ttl : Long.MAX_VALUE); + long compactSize = ObHbasePutCellCodec.getCellPayloadSize(qualifierLength, timestamp, + valueLength, hasTtl, ttl); + Assert.assertEquals(legacy.getPayloadSize(), compactSize); + + ObByteBuf compact = new ObByteBuf((int) compactSize); + ObHbasePutCellCodec.encodeCell(compact, qualifier, qualifierOffset, qualifierLength, + timestamp, value, valueOffset, valueLength, hasTtl, ttl); + Assert.assertEquals(compact.bytes.length, compact.pos); + Assert.assertArrayEquals(legacy.encode(), compact.bytes); + } + + private static ObHbaseCell newLegacyCell(byte[] qualifier, int qualifierOffset, + int qualifierLength, long timestamp, byte[] value, + int valueOffset, int valueLength, long ttl) { + boolean hasTtl = ttl != Long.MAX_VALUE; + ObHbaseCell cell = new ObHbaseCell(hasTtl); + cell.setQ(ObObj.hbasePutVarchar(new ObBytesString(qualifier, qualifierOffset, + qualifierLength))); + cell.setT(ObObj.hbasePutInt64(timestamp)); + cell.setV(ObObj.hbasePutVarchar(new ObBytesString(value, valueOffset, valueLength))); + if (hasTtl) { + cell.setTTL(ObObj.hbasePutInt64(ttl)); + } + return cell; + } +} diff --git a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpEntityHBaseCompactDecodeTest.java b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpEntityHBaseCompactDecodeTest.java new file mode 100644 index 00000000..e70a0985 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/ObTableSingleOpEntityHBaseCompactDecodeTest.java @@ -0,0 +1,119 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2026 OceanBase + * %% + * OBKV Table Client Framework is licensed under Mulan PSL v2. + * #L% + */ +package com.alipay.oceanbase.rpc.protocol.payload.impl.execute; + +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType; +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.query.ObHBaseCellBatch; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import org.junit.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +public class ObTableSingleOpEntityHBaseCompactDecodeTest { + + @Test + public void decodesKqtvDirectlyIntoCompactBatch() { + ObTableSingleOpEntity decoded = decode(newKqtvEntity(), true); + + ObHBaseCellBatch batch = decoded.getHBaseCellBatch(); + assertNotNull(batch); + assertEquals(1, batch.size()); + assertArrayEquals(bytes("r1"), batch.getRowKey(0)); + assertArrayEquals(bytes("cf\0q1"), batch.getQualifier(0)); + assertEquals(100L, batch.getTimestamp(0)); + assertArrayEquals(bytes("v0"), batch.getValue(0)); + assertEquals(0, decoded.getPropertiesValues().size()); + } + + @Test + public void absentHBaseDecodeContextUsesGenericObObjValues() { + ObTableSingleOpEntity decoded = decode(newKqtvEntity(), false); + + assertNull(decoded.getHBaseCellBatch()); + assertEquals(4, decoded.getPropertiesValues().size()); + assertArrayEquals(bytes("r1"), (byte[]) decoded.getPropertiesValues().get(0).getValue()); + assertEquals(100L, decoded.getPropertiesValues().get(2).getValue()); + } + + @Test + public void nonKqtvSchemaFallsBackToGenericDecode() { + ObTableSingleOpEntity encoded = new ObTableSingleOpEntity(); + addCell(encoded, "VALUE", "r1", "cf\0q1", 100L, "v1"); + prepareForEncode(encoded, "K", "Q", "T", "VALUE"); + ObTableSingleOpEntity decoded = decode(encoded, true, Arrays.asList("K", "Q", "T", "VALUE")); + + assertNull(decoded.getHBaseCellBatch()); + assertEquals(4, decoded.getPropertiesValues().size()); + } + + private static ObTableSingleOpEntity newKqtvEntity() { + ObTableSingleOpEntity entity = new ObTableSingleOpEntity(); + addCell(entity, "V", "r1", "cf\0q1", 100L, "v0"); + prepareForEncode(entity, "K", "Q", "T", "V"); + return entity; + } + + private static void prepareForEncode(ObTableSingleOpEntity entity, String... columns) { + entity.adjustRowkeyColumnName(new LinkedHashMap()); + Map propertyIndexes = new LinkedHashMap(); + for (int i = 0; i < columns.length; i++) { + propertyIndexes.put(columns[i], (long) i); + } + entity.adjustPropertiesColumnName(propertyIndexes); + } + + private static ObTableSingleOpEntity decode(ObTableSingleOpEntity encoded, boolean enabled) { + return decode(encoded, enabled, Arrays.asList("K", "Q", "T", "V")); + } + + private static ObTableSingleOpEntity decode(ObTableSingleOpEntity encoded, boolean enabled, + java.util.List columns) { + ByteBuf buf = Unpooled.wrappedBuffer(encoded.encode()); + try { + ObTableSingleOpEntity decoded = new ObTableSingleOpEntity(); + decoded.setAggPropertiesNames(columns); + decoded.setDecodeHBaseKqtv(enabled); + decoded.decode(buf); + return decoded; + } finally { + buf.release(); + } + } + + private static byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } + + private static ObObj binary(String value) { + ObObjMeta meta = ObObjType.ObVarcharType.getDefaultObjMeta(); + meta.setCsType(ObCollationType.CS_TYPE_BINARY); + return new ObObj(meta, bytes(value)); + } + + private static void addCell(ObTableSingleOpEntity entity, String valueColumn, String row, + String qualifier, long timestamp, String value) { + entity.addPropertyValue("K", binary(row)); + entity.addPropertyValue("Q", binary(qualifier)); + entity.addPropertyValue("T", ObObj.hbasePutInt64(timestamp)); + entity.addPropertyValue(valueColumn, binary(value)); + } +} diff --git a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/CompactHBaseStreamResultTest.java b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/CompactHBaseStreamResultTest.java new file mode 100644 index 00000000..d607cceb --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/CompactHBaseStreamResultTest.java @@ -0,0 +1,214 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2021 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.protocol.payload.impl.execute.query; + +import com.alipay.oceanbase.rpc.location.model.partition.ObPair; +import com.alipay.oceanbase.rpc.protocol.payload.ObPayload; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationLevel; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType; +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.syncquery.ObTableQueryAsyncResult; +import com.alipay.oceanbase.rpc.table.ObTableParam; +import org.junit.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class CompactHBaseStreamResultTest { + + @Test + public void testCompactCellsAdvanceWithoutMaterializingRows() throws Exception { + TestStreamResult streamResult = new TestStreamResult(); + ObHBaseCellBatch batch = batch(); + streamResult.addBatch(batch); + + assertEquals(2, streamResult.getCachedRowCount()); + assertTrue(streamResult.next()); + assertTrue(streamResult.isCurrentHBaseCell()); + assertEquals(0, streamResult.getCurrentHBaseCellIndex()); + assertArrayEquals(bytes("row-1"), streamResult.getCurrentHBaseCellBatch().getRowKey(0)); + assertEquals(1, streamResult.getCachedRowCount()); + + assertTrue(streamResult.next()); + assertTrue(streamResult.isCurrentHBaseCell()); + assertEquals(1, streamResult.getCurrentHBaseCellIndex()); + assertEquals(102L, streamResult.getCurrentHBaseCellBatch().getTimestamp(1)); + assertEquals(0, streamResult.getCachedRowCount()); + assertFalse(streamResult.next()); + } + + @Test + public void testLegacyAccessMaterializesOnlyWhenRequested() throws Exception { + TestStreamResult streamResult = new TestStreamResult(); + streamResult.addBatch(batch()); + + assertTrue(streamResult.next()); + List row = streamResult.getRow(); + assertArrayEquals(bytes("row-1"), (byte[]) row.get(0).getValue()); + assertEquals(101L, row.get(2).getValue()); + + assertEquals(1, streamResult.getCacheRows().size()); + assertFalse(streamResult.isCurrentHBaseCell()); + assertTrue(streamResult.next()); + assertArrayEquals(bytes("row-2"), (byte[]) streamResult.getRow().get(0).getValue()); + } + + @Test + public void testDrainCurrentHBaseRowLeavesNextRowUnread() throws Exception { + TestStreamResult streamResult = new TestStreamResult(); + ObHBaseCellBatch batch = batch(new String[] { "row-1", "row-1", "row-2" }, new String[] { + "q-1", "q-2", "q-3" }); + streamResult.addBatch(batch); + + assertTrue(streamResult.next()); + ObHBaseCellRow row = streamResult.drainCurrentHBaseRow(); + + assertArrayEquals(bytes("row-1"), row.getRowKey()); + assertEquals(2, row.getCellCount()); + assertEquals(1, row.getSliceCount()); + assertEquals(batch, row.getBatch(0)); + assertEquals(0, row.getFromIndex(0)); + assertEquals(2, row.getToIndex(0)); + assertEquals(1, streamResult.getRowIndex()); + assertEquals(1, streamResult.getCachedRowCount()); + + assertTrue(streamResult.next()); + assertEquals(2, streamResult.getCurrentHBaseCellIndex()); + assertArrayEquals(bytes("row-2"), streamResult.getCurrentHBaseCellBatch().getRowKey(2)); + } + + @Test + public void testDrainCurrentHBaseRowAcrossCachedBatches() throws Exception { + TestStreamResult streamResult = new TestStreamResult(); + ObHBaseCellBatch first = batch(new String[] { "row-1", "row-1" }, new String[] { "q-1", + "q-2" }); + ObHBaseCellBatch second = batch(new String[] { "row-1", "row-2" }, new String[] { "q-3", + "q-4" }); + streamResult.addBatch(first); + streamResult.addBatch(second); + + assertTrue(streamResult.next()); + ObHBaseCellRow row = streamResult.drainCurrentHBaseRow(); + + assertEquals(3, row.getCellCount()); + assertEquals(2, row.getSliceCount()); + assertEquals(first, row.getBatch(0)); + assertEquals(0, row.getFromIndex(0)); + assertEquals(2, row.getToIndex(0)); + assertEquals(second, row.getBatch(1)); + assertEquals(0, row.getFromIndex(1)); + assertEquals(1, row.getToIndex(1)); + assertEquals(2, streamResult.getRowIndex()); + assertEquals(1, streamResult.getCachedRowCount()); + + assertTrue(streamResult.next()); + assertEquals(second, streamResult.getCurrentHBaseCellBatch()); + assertEquals(1, streamResult.getCurrentHBaseCellIndex()); + assertArrayEquals(bytes("row-2"), streamResult.getCurrentHBaseCellBatch().getRowKey(1)); + } + + @Test + public void testDrainCurrentHBaseRowDoesNotConsumeDifferentRowInNextBatch() throws Exception { + TestStreamResult streamResult = new TestStreamResult(); + ObHBaseCellBatch first = batch(new String[] { "row-1" }, new String[] { "q-1" }); + ObHBaseCellBatch second = batch(new String[] { "row-2" }, new String[] { "q-2" }); + streamResult.addBatch(first); + streamResult.addBatch(second); + + assertTrue(streamResult.next()); + ObHBaseCellRow row = streamResult.drainCurrentHBaseRow(); + + assertEquals(1, row.getCellCount()); + assertEquals(1, row.getSliceCount()); + assertEquals(1, streamResult.getCachedRowCount()); + assertTrue(streamResult.next()); + assertEquals(second, streamResult.getCurrentHBaseCellBatch()); + assertEquals(0, streamResult.getCurrentHBaseCellIndex()); + } + + private static ObHBaseCellBatch batch() { + return batch(new String[] { "row-1", "row-2" }, new String[] { "q-1", "q-2" }); + } + + private static ObHBaseCellBatch batch(String[] rowKeys, String[] qualifiers) { + assertEquals(rowKeys.length, qualifiers.length); + ObHBaseCellBatch batch = new ObHBaseCellBatch(rowKeys.length); + ObObjMeta binaryMeta = new ObObjMeta(ObObjType.ObVarcharType, + ObCollationLevel.CS_LEVEL_EXPLICIT, ObCollationType.CS_TYPE_BINARY, (byte) 10); + ObObjMeta timestampMeta = new ObObjMeta(ObObjType.ObInt64Type, + ObCollationLevel.CS_LEVEL_NUMERIC, ObCollationType.CS_TYPE_BINARY, (byte) 10); + batch.setMeta(0, binaryMeta); + batch.setMeta(1, binaryMeta); + batch.setMeta(2, timestampMeta); + batch.setMeta(3, binaryMeta); + for (int i = 0; i < rowKeys.length; i++) { + batch + .setCell(i, bytes(rowKeys[i]), bytes(qualifiers[i]), 101L + i, bytes("value-" + i)); + } + return batch; + } + + private static byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } + + private static class TestStreamResult extends AbstractQueryStreamResult { + + TestStreamResult() { + initialized = true; + expectant = new LinkedHashMap>(); + } + + void addBatch(ObHBaseCellBatch batch) { + cacheHBaseCellBatches.addLast(batch); + } + + @Override + protected ObPayload referToNewPartition(ObPair partIdWithObTable) { + return null; + } + + @Override + protected ObTableQueryResult execute(ObPair partIdWithObTable, + ObPayload streamRequest) { + return null; + } + + @Override + protected ObTableQueryAsyncResult executeAsync(ObPair partIdWithObTable, + ObPayload streamRequest) { + return null; + } + + @Override + protected Map> refreshPartition(ObTableQuery tableQuery, + String tableName) { + return Collections.emptyMap(); + } + } +} 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 index 8b6903e8..07254fd0 100644 --- 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 @@ -24,7 +24,9 @@ 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.ObByteBuf; import com.alipay.oceanbase.rpc.util.ObBytesString; +import com.alipay.oceanbase.rpc.util.Serialization; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; import org.junit.Assert; @@ -35,6 +37,8 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; public class ObTableQueryPayloadTest { @@ -57,6 +61,11 @@ public void test_ObHTableFilter() { ObHTableFilter obHTableFilter = getObHTableFilter(); byte[] bytes = obHTableFilter.encode(); + ObByteBuf obByteBuf = new ObByteBuf(bytes.length); + obHTableFilter.encode(obByteBuf); + assertEquals(bytes.length, obByteBuf.pos); + assertArrayEquals(bytes, obByteBuf.bytes); + ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(); buf.writeBytes(bytes); @@ -67,6 +76,16 @@ public void test_ObHTableFilter() { } + @Test + public void test_ObHTableFilterResetPayloadSize() { + ObHTableFilter obHTableFilter = new ObHTableFilter(); + long emptyFilterSize = obHTableFilter.getPayloadContentSize(); + + obHTableFilter.setFilterString("123".getBytes()); + + assertEquals(emptyFilterSize + 3, obHTableFilter.getPayloadContentSize()); + } + @Test public void test_ObTableQuery() { ObTableQuery obTableQuery = getObTableQuery(); @@ -153,6 +172,66 @@ public void test_ObTableQueryResult() { buf.release(); } + @Test + public void test_ObTableQueryResultWithRows() { + ObTableQueryResult obTableQueryResult = getObTableQueryResultWithRows(); + + byte[] bytes = obTableQueryResult.encode(); + ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(); + buf.writeBytes(bytes); + + ObTableQueryResult decodedResult = new ObTableQueryResult(); + decodedResult.decode(buf); + + assertEquals(2, decodedResult.getRowCount()); + assertEquals(obTableQueryResult.getPropertiesNames(), decodedResult.getPropertiesNames()); + assertEquals(2, decodedResult.getPropertiesRows().size()); + assertEquals(3, decodedResult.getPropertiesRows().get(0).size()); + assertEquals(11L, decodedResult.getPropertiesRows().get(0).get(0).getValue()); + assertEquals(13L, decodedResult.getPropertiesRows().get(0).get(2).getValue()); + assertEquals(21L, decodedResult.getPropertiesRows().get(1).get(0).getValue()); + assertEquals(23L, decodedResult.getPropertiesRows().get(1).get(2).getValue()); + assertTrue(decodedResult.getPropertiesNames() instanceof ArrayList); + assertTrue(decodedResult.getPropertiesRows() instanceof ArrayList); + assertTrue(decodedResult.getPropertiesRows().get(0) instanceof ArrayList); + + decodedResult.getPropertiesNames().add("extra"); + assertEquals("extra", decodedResult.getPropertiesNames().remove(3)); + decodedResult.getPropertiesRows().get(0).add(ObObj.getInstance(14L)); + assertEquals(14L, decodedResult.getPropertiesRows().get(0).remove(3).getValue()); + buf.release(); + } + + @Test + public void test_ObTableQueryResultDecodeReplacesPreviousRows() { + ObTableQueryResult decodedResult = new ObTableQueryResult(); + ByteBuf firstBuf = PooledByteBufAllocator.DEFAULT.buffer(); + firstBuf.writeBytes(getObTableQueryResultWithRows().encode()); + decodedResult.decode(firstBuf); + firstBuf.release(); + + ObTableQueryResult emptyResult = new ObTableQueryResult(); + emptyResult.addPropertiesName("only"); + emptyResult.setRowCount(0); + ByteBuf secondBuf = PooledByteBufAllocator.DEFAULT.buffer(); + secondBuf.writeBytes(emptyResult.encode()); + decodedResult.decode(secondBuf); + + assertEquals(0, decodedResult.getRowCount()); + assertEquals(1, decodedResult.getPropertiesNames().size()); + assertEquals("only", decodedResult.getPropertiesNames().get(0)); + assertTrue(decodedResult.getPropertiesRows().isEmpty()); + secondBuf.release(); + } + + @Test + public void test_ObTableQueryResultRejectsInvalidCounts() { + assertInvalidPropertyCount(-1L); + assertInvalidPropertyCount((long) Integer.MAX_VALUE + 1); + assertInvalidRowCount(-1L); + assertInvalidRowCount((long) Integer.MAX_VALUE + 1); + } + @Test public void testFtsParam() { ObFTSParams ftsParams = new ObFTSParams(); @@ -198,6 +277,54 @@ private ObTableQuery getObTableQuery() { return obTableQuery; } + private ObTableQueryResult getObTableQueryResultWithRows() { + ObTableQueryResult result = new ObTableQueryResult(); + result.addPropertiesName("c1"); + result.addPropertiesName("c2"); + result.addPropertiesName("c3"); + result.addPropertiesRow(getObObjRow(11L, 12L, 13L)); + result.addPropertiesRow(getObObjRow(21L, 22L, 23L)); + result.setRowCount(result.getPropertiesRows().size()); + return result; + } + + private List getObObjRow(long first, long second, long third) { + List row = new ArrayList(3); + row.add(ObObj.getInstance(first)); + row.add(ObObj.getInstance(second)); + row.add(ObObj.getInstance(third)); + return row; + } + + private void assertInvalidPropertyCount(long propertyCount) { + ByteBuf buf = newQueryResultBuffer(Serialization.encodeVi64(propertyCount)); + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> new ObTableQueryResult().decode(buf)); + assertTrue(exception.getMessage().contains("property count")); + buf.release(); + } + + private void assertInvalidRowCount(long rowCount) { + byte[] propertyCount = Serialization.encodeVi64(0); + byte[] encodedRowCount = Serialization.encodeVi64(rowCount); + byte[] content = new byte[propertyCount.length + encodedRowCount.length]; + System.arraycopy(propertyCount, 0, content, 0, propertyCount.length); + System.arraycopy(encodedRowCount, 0, content, propertyCount.length, encodedRowCount.length); + + ByteBuf buf = newQueryResultBuffer(content); + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> new ObTableQueryResult().decode(buf)); + assertTrue(exception.getMessage().contains("row count")); + buf.release(); + } + + private ByteBuf newQueryResultBuffer(byte[] content) { + ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(); + buf.writeBytes(Serialization.encodeObUniVersionHeader(1, content.length)); + buf.writeBytes(content); + return buf; + } + private void checkObTableQuery(ObTableQuery obTableQuery, ObTableQuery newObTableQuery) { checkObNewRange(obTableQuery.getKeyRanges().get(0), newObTableQuery.getKeyRanges().get(0)); assertEquals(obTableQuery.getSelectColumns().get(0), newObTableQuery.getSelectColumns() diff --git a/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryResultHBaseDecodeTest.java b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryResultHBaseDecodeTest.java new file mode 100644 index 00000000..f436c6c3 --- /dev/null +++ b/src/test/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/ObTableQueryResultHBaseDecodeTest.java @@ -0,0 +1,319 @@ +/*- + * #%L + * OBKV Table Client Framework + * %% + * Copyright (C) 2021 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.protocol.payload.impl.execute.query; + +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationLevel; +import com.alipay.oceanbase.rpc.protocol.payload.impl.ObCollationType; +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.util.ObBytesString; +import com.alipay.oceanbase.rpc.util.Serialization; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.Unpooled; +import org.junit.Test; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class ObTableQueryResultHBaseDecodeTest { + + @Test + public void testFastDecodeMatchesGenericDecodeAndCachesMeta() { + List> fastRows = new ArrayList>(); + fastRows.add(newHBaseRow(bytes("row-1"), bytes("q-1"), 101L, bytes("value-1"), (byte) 10)); + fastRows.add(newHBaseRow(bytes("row-2"), bytes("q-2"), -102L, bytes("value-2"), (byte) 10)); + + ObTableQueryResult fastResult = decode( + newQueryResult(new String[] { "K", "Q", "T", "V" }, fastRows), true); + ObTableQueryResult genericResult = decode( + newQueryResult(new String[] { "K", "Q", "T", "VALUE" }, copyRows(fastRows)), false); + + assertTrue(fastResult.hasHBaseCellBatch()); + ObHBaseCellBatch batch = fastResult.getHBaseCellBatch(); + assertEquals(2, batch.size()); + assertArrayEquals(bytes("row-1"), batch.getRowKey(0)); + assertArrayEquals(bytes("q-2"), batch.getQualifier(1)); + assertEquals(-102L, batch.getTimestamp(1)); + assertArrayEquals(bytes("value-2"), batch.getValue(1)); + + assertRowsEqual(genericResult.getPropertiesRows(), fastResult.getPropertiesRows()); + assertFalse(fastResult.hasHBaseCellBatch()); + assertTrue(fastResult.getPropertiesRows() instanceof ArrayList); + for (int columnIndex = 0; columnIndex < 4; columnIndex++) { + assertSame(fastResult.getPropertiesRows().get(0).get(columnIndex).getMeta(), fastResult + .getPropertiesRows().get(1).get(columnIndex).getMeta()); + assertNotSame(genericResult.getPropertiesRows().get(0).get(columnIndex).getMeta(), + genericResult.getPropertiesRows().get(1).get(columnIndex).getMeta()); + } + } + + @Test + public void testUnexpectedFirstRowMetaFallsBackToGenericDecode() { + List> rows = new ArrayList>(); + rows.add(newTextKeyRow("row-1", bytes("q-1"), 101L, bytes("value-1"))); + rows.add(newTextKeyRow("row-2", bytes("q-2"), 102L, bytes("value-2"))); + + ObTableQueryResult result = decode( + newQueryResult(new String[] { "K", "Q", "T", "V" }, rows), false); + + assertEquals("row-1", result.getPropertiesRows().get(0).get(0).getValue()); + assertEquals("row-2", result.getPropertiesRows().get(1).get(0).getValue()); + assertNotSame(result.getPropertiesRows().get(0).get(0).getMeta(), result + .getPropertiesRows().get(1).get(0).getMeta()); + } + + @Test + public void testLaterRowMetaMismatchFailsClosed() { + List> rows = new ArrayList>(); + rows.add(newHBaseRow(bytes("row-1"), bytes("q-1"), 101L, bytes("value-1"), (byte) 10)); + rows.add(newHBaseRow(bytes("row-2"), bytes("q-2"), 102L, bytes("value-2"), (byte) 11)); + ObTableQueryResult encodedResult = newQueryResult(new String[] { "K", "Q", "T", "V" }, + rows); + ByteBuf buf = Unpooled.wrappedBuffer(encodedResult.encode()); + try { + IllegalStateException exception = assertThrows(IllegalStateException.class, + () -> new ObTableQueryResult().decode(buf)); + assertTrue(exception.getMessage().contains("meta changed")); + } finally { + buf.release(); + } + } + + @Test + public void testResponseMetaCacheDoesNotCrossResponses() { + List> firstRows = new ArrayList>(); + firstRows.add(newHBaseRow(bytes("row-1"), bytes("q-1"), 101L, bytes("value-1"), (byte) 10)); + List> secondRows = new ArrayList>(); + secondRows + .add(newHBaseRow(bytes("row-2"), bytes("q-2"), 102L, bytes("value-2"), (byte) 11)); + + ObTableQueryResult first = decode( + newQueryResult(new String[] { "K", "Q", "T", "V" }, firstRows), false); + ObTableQueryResult second = decode( + newQueryResult(new String[] { "K", "Q", "T", "V" }, secondRows), false); + + assertEquals(10, first.getPropertiesRows().get(0).get(0).getMeta().getScale()); + assertEquals(11, second.getPropertiesRows().get(0).get(0).getMeta().getScale()); + assertNotSame(first.getPropertiesRows().get(0).get(0).getMeta(), second.getPropertiesRows() + .get(0).get(0).getMeta()); + } + + @Test + public void testCompactBatchCanBeEncodedWithoutMaterializingRows() { + List> rows = new ArrayList>(); + rows.add(newHBaseRow(bytes("row-1"), bytes("q-1"), 101L, bytes("value-1"), + (byte) 10)); + rows.add(newHBaseRow(bytes("row-2"), bytes("q-2"), 102L, bytes("value-2"), + (byte) 10)); + + ObTableQueryResult decoded = decode( + newQueryResult(new String[] { "K", "Q", "T", "V" }, rows), true); + assertTrue(decoded.hasHBaseCellBatch()); + + ObTableQueryResult roundTripped = decode(decoded, false); + assertTrue(decoded.hasHBaseCellBatch()); + assertTrue(roundTripped.hasHBaseCellBatch()); + assertRowsEqual(rows, roundTripped.getPropertiesRows()); + } + + @Test + public void testEmptyKqtvResult() { + ObTableQueryResult result = decode( + newQueryResult(new String[] { "K", "Q", "T", "V" }, new ArrayList>()), + false); + + assertEquals(0, result.getRowCount()); + assertTrue(result.getPropertiesRows().isEmpty()); + } + + @Test + public void testDecodeBinaryColumnFromHeapAndDirectBuffer() { + byte[] expected = bytes("binary-value"); + byte[] encoded = Serialization.encodeBytesString(new ObBytesString(expected)); + + ByteBuf heapBuf = Unpooled.wrappedBuffer(encoded); + try { + assertArrayEquals(expected, Serialization.decodeBinaryColumn(heapBuf)); + assertFalse(heapBuf.isReadable()); + } finally { + heapBuf.release(); + } + + ByteBuf directBuf = PooledByteBufAllocator.DEFAULT.directBuffer(encoded.length); + try { + directBuf.writeBytes(encoded); + assertArrayEquals(expected, Serialization.decodeBinaryColumn(directBuf)); + assertFalse(directBuf.isReadable()); + } finally { + directBuf.release(); + } + } + + @Test + public void testDecodeEmptyBinaryColumn() { + byte[] encoded = Serialization.encodeBytesString(new ObBytesString(new byte[0])); + ByteBuf buf = Unpooled.wrappedBuffer(encoded); + try { + assertArrayEquals(new byte[0], Serialization.decodeBinaryColumn(buf)); + assertFalse(buf.isReadable()); + } finally { + buf.release(); + } + } + + @Test + public void testDecodeBinaryColumnRejectsTruncatedValue() { + ByteBuf buf = Unpooled.buffer(); + try { + buf.writeBytes(Serialization.encodeVi32(2)); + buf.writeByte(1); + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> Serialization.decodeBinaryColumn(buf)); + assertTrue(exception.getMessage().contains("length")); + } finally { + buf.release(); + } + } + + @Test + public void testDecodeBinaryColumnRejectsNegativeLength() { + ByteBuf buf = Unpooled.wrappedBuffer(Serialization.encodeVi32(-1)); + try { + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> Serialization.decodeBinaryColumn(buf)); + assertTrue(exception.getMessage().contains("length")); + } finally { + buf.release(); + } + } + + @Test + public void testDecodeBinaryColumnRejectsInvalidTerminator() { + ByteBuf buf = Unpooled.buffer(); + try { + buf.writeBytes(Serialization.encodeVi32(1)); + buf.writeByte(1); + buf.writeByte(2); + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> Serialization.decodeBinaryColumn(buf)); + assertTrue(exception.getMessage().contains("terminator")); + } finally { + buf.release(); + } + } + + private ObTableQueryResult decode(ObTableQueryResult encodedResult, boolean direct) { + byte[] encoded = encodedResult.encode(); + ByteBuf buf = direct ? PooledByteBufAllocator.DEFAULT.directBuffer(encoded.length) + : Unpooled.buffer(encoded.length); + try { + buf.writeBytes(encoded); + ObTableQueryResult result = new ObTableQueryResult(); + result.decode(buf); + assertFalse(buf.isReadable()); + return result; + } finally { + buf.release(); + } + } + + private ObTableQueryResult newQueryResult(String[] propertyNames, List> rows) { + ObTableQueryResult result = new ObTableQueryResult(); + for (String propertyName : propertyNames) { + result.addPropertiesName(propertyName); + } + result.addAllPropertiesRows(rows); + result.setRowCount(rows.size()); + return result; + } + + private List> copyRows(List> rows) { + List> copiedRows = new ArrayList>(rows.size()); + for (List row : rows) { + copiedRows.add(newHBaseRow((byte[]) row.get(0).getValue(), (byte[]) row.get(1) + .getValue(), (Long) row.get(2).getValue(), (byte[]) row.get(3).getValue(), + row.get(0).getMeta().getScale())); + } + return copiedRows; + } + + private List newHBaseRow(byte[] key, byte[] qualifier, long timestamp, byte[] value, + byte scale) { + List row = new ArrayList(4); + row.add(newBinaryObj(key, scale)); + row.add(newBinaryObj(qualifier, scale)); + row.add(new ObObj(new ObObjMeta(ObObjType.ObInt64Type, ObCollationLevel.CS_LEVEL_NUMERIC, + ObCollationType.CS_TYPE_BINARY, scale), timestamp)); + row.add(newBinaryObj(value, scale)); + return row; + } + + private List newTextKeyRow(String key, byte[] qualifier, long timestamp, byte[] value) { + List row = new ArrayList(4); + row.add(new ObObj(new ObObjMeta(ObObjType.ObVarcharType, + ObCollationLevel.CS_LEVEL_EXPLICIT, ObCollationType.CS_TYPE_UTF8MB4_GENERAL_CI, + (byte) 10), key)); + row.add(newBinaryObj(qualifier, (byte) 10)); + row.add(new ObObj(new ObObjMeta(ObObjType.ObInt64Type, ObCollationLevel.CS_LEVEL_NUMERIC, + ObCollationType.CS_TYPE_BINARY, (byte) 10), timestamp)); + row.add(newBinaryObj(value, (byte) 10)); + return row; + } + + private ObObj newBinaryObj(byte[] value, byte scale) { + return new ObObj(new ObObjMeta(ObObjType.ObVarcharType, ObCollationLevel.CS_LEVEL_EXPLICIT, + ObCollationType.CS_TYPE_BINARY, scale), value); + } + + private void assertRowsEqual(List> expectedRows, List> actualRows) { + assertEquals(expectedRows.size(), actualRows.size()); + for (int rowIndex = 0; rowIndex < expectedRows.size(); rowIndex++) { + List expectedRow = expectedRows.get(rowIndex); + List actualRow = actualRows.get(rowIndex); + assertEquals(expectedRow.size(), actualRow.size()); + for (int columnIndex = 0; columnIndex < expectedRow.size(); columnIndex++) { + ObObj expected = expectedRow.get(columnIndex); + ObObj actual = actualRow.get(columnIndex); + if (columnIndex == 2) { + assertEquals(expected.getValue(), actual.getValue()); + } else { + assertArrayEquals((byte[]) expected.getValue(), (byte[]) actual.getValue()); + } + assertEquals(expected.getMeta().getType(), actual.getMeta().getType()); + assertEquals(expected.getMeta().getCsLevel(), actual.getMeta().getCsLevel()); + assertEquals(expected.getMeta().getCsType(), actual.getMeta().getCsType()); + assertEquals(expected.getMeta().getScale(), actual.getMeta().getScale()); + } + } + } + + private byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } +} diff --git a/src/test/java/com/alipay/oceanbase/rpc/util/ObCrcUtilTest.java b/src/test/java/com/alipay/oceanbase/rpc/util/ObCrcUtilTest.java index fee214ee..c0c76cf7 100644 --- a/src/test/java/com/alipay/oceanbase/rpc/util/ObCrcUtilTest.java +++ b/src/test/java/com/alipay/oceanbase/rpc/util/ObCrcUtilTest.java @@ -21,6 +21,9 @@ import org.junit.Assert; import org.junit.Test; +import java.nio.charset.StandardCharsets; +import java.util.Random; + public class ObCrcUtilTest { @Test public void testObCrc32() { @@ -28,6 +31,41 @@ public void testObCrc32() { Assert.assertEquals(1445960909, ObPureCrc32C.calculate(v1.getBytes())); } + @Test + public void testOceanBaseCrc32cKnownVectors() { + Assert.assertEquals(0L, ObPureCrc32C.calculate(new byte[0])); + Assert.assertEquals(0x6345d352L, + ObPureCrc32C.calculate("hello world".getBytes(StandardCharsets.UTF_8))); + Assert.assertEquals(0x58e3fa20L, + ObPureCrc32C.calculate("123456789".getBytes(StandardCharsets.UTF_8))); + } + + @Test + public void testSlicingBy8MatchesScalar() { + Random random = new Random(20260807L); + int[] lengths = new int[] { 0, 1, 7, 8, 15, 16, 17, 31, 32, 63, 64, 65, 127, 128, 129, 255, + 256, 257, 1023, 1024, 1025, 65535, 1048576 }; + for (int length : lengths) { + byte[] bytes = new byte[length + 11]; + random.nextBytes(bytes); + assertSlicingBy8EqualsScalar(bytes, 5, length); + } + + for (int round = 0; round < 1000; round++) { + int length = random.nextInt(32768); + int prefix = random.nextInt(16); + byte[] bytes = new byte[prefix + length + random.nextInt(16)]; + random.nextBytes(bytes); + assertSlicingBy8EqualsScalar(bytes, prefix, length); + } + } + + private static void assertSlicingBy8EqualsScalar(byte[] bytes, int offset, int length) { + long scalar = ObPureCrc32C.calculateScalar(bytes, offset, length); + Assert.assertEquals(scalar, ObPureCrc32C.calculateSlicingBy8(bytes, offset, length)); + Assert.assertEquals(scalar, ObPureCrc32C.calculate(bytes, offset, length)); + } + @Test public void testCrc64() { CRC64 crc64 = new CRC64(); diff --git a/src/test/java/com/alipay/oceanbase/rpc/util/ObHashUtilTest.java b/src/test/java/com/alipay/oceanbase/rpc/util/ObHashUtilTest.java index add97ada..227ce63c 100644 --- a/src/test/java/com/alipay/oceanbase/rpc/util/ObHashUtilTest.java +++ b/src/test/java/com/alipay/oceanbase/rpc/util/ObHashUtilTest.java @@ -63,6 +63,20 @@ public void testHash() { ObHashUtils.timeStampHash(Timestamp.valueOf("2019-11-12 15:34:28.986"), 47)); } + @Test + public void testVarcharHashHonorsObBytesStringRange() { + byte[] exact = new byte[] { 'A', 'b', 'C' }; + ObBytesString slice = new ObBytesString(new byte[] { 0x01, 'A', 'b', 'C', 0x02 }, 1, 3); + + for (ObCollationType collationType : new ObCollationType[] { + ObCollationType.CS_TYPE_UTF8MB4_GENERAL_CI, ObCollationType.CS_TYPE_UTF8MB4_BIN, + ObCollationType.CS_TYPE_BINARY }) { + Assert.assertEquals( + ObHashUtils.varcharHash(exact, collationType, 47, ObPartFuncType.KEY_V3), + ObHashUtils.varcharHash(slice, collationType, 47, ObPartFuncType.KEY_V3)); + } + } + @Test public void testMurmurHash() { String hello = "HelloWorld";