This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch remove_swtich_type in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 814cb3c90e7843126e932bec101dc463e3e2a87e Author: Tian Jiang <[email protected]> AuthorDate: Thu Aug 27 14:53:34 2026 +0800 multiple refactors --- .../java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java | 96 +++++----------------- .../java/org/apache/iotdb/rpc/TypeServices.java | 94 +++++++++++++++++++++ .../iotdb/rpc/stmt/PreparedParameterSerde.java | 41 ++++----- .../org/apache/iotdb/rpc/IoTDBJDBCDataSetTest.java | 70 ++++++++++++++++ .../iotdb/rpc/stmt/PreparedParameterSerdeTest.java | 12 +++ 5 files changed, 219 insertions(+), 94 deletions(-) diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java index 42bf974a3eb..786847b8eac 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java @@ -27,12 +27,12 @@ import org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq; import org.apache.iotdb.service.rpc.thrift.TSFetchResultsResp; import org.apache.thrift.TException; -import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.block.column.Column; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.common.block.column.TsBlockSerde; +import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.utils.Binary; -import org.apache.tsfile.utils.BytesUtils; import org.apache.tsfile.utils.DateUtils; import java.nio.ByteBuffer; @@ -473,37 +473,10 @@ public class IoTDBRpcDataSet { return null; } lastReadWasNull = false; - TSDataType tsDataType = getDataTypeByTsBlockColumnIndex(tsBlockColumnIndex); - switch (tsDataType) { - case BOOLEAN: - case INT32: - case INT64: - case FLOAT: - case DOUBLE: - return curTsBlock.getColumn(tsBlockColumnIndex).getObject(tsBlockIndex); - case TIMESTAMP: - long timestamp = - (tsBlockColumnIndex == -1 - ? curTsBlock.getTimeByIndex(tsBlockIndex) - : curTsBlock.getColumn(tsBlockColumnIndex).getLong(tsBlockIndex)); - return convertToTimestamp(timestamp, timeFactor); - case TEXT: - case STRING: - return curTsBlock - .getColumn(tsBlockColumnIndex) - .getBinary(tsBlockIndex) - .getStringValue(TSFileConfig.STRING_CHARSET); - case OBJECT: - return BytesUtils.parseObjectByteArrayToString( - curTsBlock.getColumn(tsBlockColumnIndex).getBinary(tsBlockIndex).getValues()); - case BLOB: - return BytesUtils.parseBlobByteArrayToString( - curTsBlock.getColumn(tsBlockColumnIndex).getBinary(tsBlockIndex).getValues()); - case DATE: - return DateUtils.formatDate(curTsBlock.getColumn(tsBlockColumnIndex).getInt(tsBlockIndex)); - default: - return null; - } + Type type = Type.fromTsDataType(getDataTypeByTsBlockColumnIndex(tsBlockColumnIndex)); + return TypeServices.RPC_OBJECT_READER_SERVICE + .call(type) + .read(type, getColumnByTsBlockColumnIndex(tsBlockColumnIndex), tsBlockIndex, timeFactor); } public String getString(int columnIndex) throws StatementExecutionException { @@ -526,47 +499,22 @@ public class IoTDBRpcDataSet { return null; } lastReadWasNull = false; - return getString(tsBlockColumnIndex, getDataTypeByTsBlockColumnIndex(tsBlockColumnIndex)); - } - - private String getString(int index, TSDataType tsDataType) { - switch (tsDataType) { - case BOOLEAN: - return String.valueOf(curTsBlock.getColumn(index).getBoolean(tsBlockIndex)); - case INT32: - return String.valueOf(curTsBlock.getColumn(index).getInt(tsBlockIndex)); - case INT64: - return String.valueOf( - (index == -1 - ? curTsBlock.getTimeByIndex(tsBlockIndex) - : curTsBlock.getColumn(index).getLong(tsBlockIndex))); - case TIMESTAMP: - long timestamp = - (index == -1 - ? curTsBlock.getTimeByIndex(tsBlockIndex) - : curTsBlock.getColumn(index).getLong(tsBlockIndex)); - return RpcUtils.formatDatetime(timeFormat, timePrecision, timestamp, zoneId); - case FLOAT: - return String.valueOf(curTsBlock.getColumn(index).getFloat(tsBlockIndex)); - case DOUBLE: - return String.valueOf(curTsBlock.getColumn(index).getDouble(tsBlockIndex)); - case TEXT: - case STRING: - return curTsBlock - .getColumn(index) - .getBinary(tsBlockIndex) - .getStringValue(TSFileConfig.STRING_CHARSET); - case OBJECT: - return BytesUtils.parseObjectByteArrayToString( - curTsBlock.getColumn(index).getBinary(tsBlockIndex).getValues()); - case BLOB: - return BytesUtils.parseBlobByteArrayToString( - curTsBlock.getColumn(index).getBinary(tsBlockIndex).getValues()); - case DATE: - return DateUtils.formatDate(curTsBlock.getColumn(index).getInt(tsBlockIndex)); - default: - return null; - } + Type type = Type.fromTsDataType(getDataTypeByTsBlockColumnIndex(tsBlockColumnIndex)); + return TypeServices.RPC_STRING_READER_SERVICE + .call(type) + .read( + type, + getColumnByTsBlockColumnIndex(tsBlockColumnIndex), + tsBlockIndex, + timeFormat, + timePrecision, + zoneId); + } + + private Column getColumnByTsBlockColumnIndex(int tsBlockColumnIndex) { + return tsBlockColumnIndex == -1 + ? curTsBlock.getTimeColumn() + : curTsBlock.getColumn(tsBlockColumnIndex); } public Timestamp getTimestamp(int columnIndex) throws StatementExecutionException { diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TypeServices.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TypeServices.java index 4e1cec95c6f..4b031124372 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TypeServices.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TypeServices.java @@ -18,6 +18,9 @@ package org.apache.iotdb.rpc; +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.read.common.type.service.TypeService; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BytesUtils; @@ -25,6 +28,7 @@ import org.apache.tsfile.utils.DateUtils; import java.nio.charset.StandardCharsets; import java.sql.Timestamp; +import java.time.ZoneId; import java.util.function.Function; final class TypeServices { @@ -59,10 +63,100 @@ final class TypeServices { case ROW, UNKNOWN, VECTOR -> ignored -> null; }; + static final TypeService<RpcObjectReader> RPC_OBJECT_READER_SERVICE = + type -> + switch (type.getTypeEnum()) { + case BOOLEAN, INT32, INT64, FLOAT, DOUBLE -> + (actualType, column, position, timeFactor) -> + actualType.getObject(column, position); + case TIMESTAMP -> + (actualType, column, position, timeFactor) -> + RpcUtils.convertToTimestamp(actualType.getLong(column, position), timeFactor); + case TEXT, STRING -> + (actualType, column, position, timeFactor) -> + actualType + .getBinary(column, position) + .getStringValue(TSFileConfig.STRING_CHARSET); + case OBJECT -> + (actualType, column, position, timeFactor) -> + BytesUtils.parseObjectByteArrayToString( + actualType.getBinary(column, position).getValues()); + case BLOB -> + (actualType, column, position, timeFactor) -> + BytesUtils.parseBlobByteArrayToString( + actualType.getBinary(column, position).getValues()); + case DATE -> + (actualType, column, position, timeFactor) -> + DateUtils.formatDate(actualType.getInt(column, position)); + case ROW, UNKNOWN, VECTOR -> (actualType, column, position, timeFactor) -> null; + }; + + static final TypeService<RpcStringReader> RPC_STRING_READER_SERVICE = + type -> + switch (type.getTypeEnum()) { + case BOOLEAN -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + String.valueOf(actualType.getBoolean(column, position)); + case INT32 -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + String.valueOf(actualType.getInt(column, position)); + case INT64 -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + String.valueOf(actualType.getLong(column, position)); + case FLOAT -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + String.valueOf(actualType.getFloat(column, position)); + case DOUBLE -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + String.valueOf(actualType.getDouble(column, position)); + case TIMESTAMP -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + RpcUtils.formatDatetime( + timeFormat, timePrecision, actualType.getLong(column, position), zoneId); + case TEXT, STRING -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + actualType + .getBinary(column, position) + .getStringValue(TSFileConfig.STRING_CHARSET); + case OBJECT -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + BytesUtils.parseObjectByteArrayToString( + actualType.getBinary(column, position).getValues()); + case BLOB -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + BytesUtils.parseBlobByteArrayToString( + actualType.getBinary(column, position).getValues()); + case DATE -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> + DateUtils.formatDate(actualType.getInt(column, position)); + case ROW, UNKNOWN, VECTOR -> + (actualType, column, position, timeFormat, timePrecision, zoneId) -> null; + }; + static { JDBC_STRING_READER_SERVICE.check(); JDBC_OBJECT_READER_SERVICE.check(); + RPC_OBJECT_READER_SERVICE.check(); + RPC_STRING_READER_SERVICE.check(); } private TypeServices() {} + + @FunctionalInterface + interface RpcObjectReader { + + Object read(Type type, Column column, int position, int timeFactor); + } + + @FunctionalInterface + interface RpcStringReader { + + String read( + Type type, + Column column, + int position, + String timeFormat, + String timePrecision, + ZoneId zoneId); + } } diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerde.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerde.java index e42710d23e1..6bf9cbba226 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerde.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerde.java @@ -21,10 +21,13 @@ package org.apache.iotdb.rpc.stmt; import org.apache.iotdb.rpc.i18n.RpcMessages; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.PublicBAOS; import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.apache.tsfile.utils.TsPrimitiveType; import java.io.IOException; import java.io.OutputStream; @@ -145,27 +148,25 @@ public class PreparedParameterSerde { } private static Object deserializeValue(ByteBuffer buffer, TSDataType type) { - switch (type) { - case UNKNOWN: - return null; - case BOOLEAN: - return ReadWriteIOUtils.readBool(buffer); - case INT32: - return ReadWriteIOUtils.readInt(buffer); - case INT64: - return ReadWriteIOUtils.readLong(buffer); - case FLOAT: - return ReadWriteIOUtils.readFloat(buffer); - case DOUBLE: - return ReadWriteIOUtils.readDouble(buffer); - case TEXT: - case STRING: - return ReadWriteIOUtils.readString(buffer); - case BLOB: - return ReadWriteIOUtils.readBinary(buffer).getValues(); - default: - throw new IllegalArgumentException(RpcMessages.UNSUPPORTED_TYPE + type); + if (type == TSDataType.UNKNOWN) { + return null; + } + if (!type.isNumeric() + && type != TSDataType.BOOLEAN + && type != TSDataType.TEXT + && type != TSDataType.STRING + && type != TSDataType.BLOB) { + throw new IllegalArgumentException(RpcMessages.UNSUPPORTED_TYPE + type); + } + + TsPrimitiveType value = Type.fromTsDataType(type).deserialize(buffer); + if (type == TSDataType.BLOB) { + return value.getBinary().getValues(); + } + if (type == TSDataType.TEXT || type == TSDataType.STRING) { + return value.getBinary().getStringValue(TSFileConfig.STRING_CHARSET); } + return value.getValue(); } /** Convert byte array to hexadecimal string representation. */ diff --git a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/IoTDBJDBCDataSetTest.java b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/IoTDBJDBCDataSetTest.java index af98f2a60ec..ccadd0d51d5 100644 --- a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/IoTDBJDBCDataSetTest.java +++ b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/IoTDBJDBCDataSetTest.java @@ -18,15 +18,20 @@ package org.apache.iotdb.rpc; +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.read.common.type.UnknownType; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BytesUtils; +import org.apache.tsfile.utils.DateUtils; import org.apache.tsfile.write.UnSupportedDataTypeException; import org.junit.Assert; import org.junit.Test; +import java.nio.ByteBuffer; +import java.time.ZoneId; import java.util.Arrays; public class IoTDBJDBCDataSetTest { @@ -93,6 +98,71 @@ public class IoTDBJDBCDataSetTest { TypeServices.JDBC_OBJECT_READER_SERVICE.call(UnknownType.UNKNOWN).apply(new byte[0])); } + @Test + public void testRpcValueReadersUseTypeServices() { + Type intType = Type.fromTsDataType(TSDataType.INT32); + Column intColumn = intType.createColumnBuilder(1).writeInt(42).build(); + Assert.assertEquals(42, readRpcObject(intType, intColumn, 1_000)); + Assert.assertEquals("42", readRpcString(intType, intColumn)); + + Type textType = Type.fromTsDataType(TSDataType.TEXT); + Column textColumn = + textType + .createColumnBuilder(1) + .writeBinary(new Binary("text", TSFileConfig.STRING_CHARSET)) + .build(); + Assert.assertEquals("text", readRpcObject(textType, textColumn, 1_000)); + Assert.assertEquals("text", readRpcString(textType, textColumn)); + + Type timestampType = Type.fromTsDataType(TSDataType.TIMESTAMP); + Column timestampColumn = timestampType.createColumnBuilder(1).writeLong(1_234_567).build(); + Assert.assertEquals( + RpcUtils.convertToTimestamp(1_234_567, 1_000_000), + readRpcObject(timestampType, timestampColumn, 1_000_000)); + Assert.assertEquals("1234567", readRpcString(timestampType, timestampColumn)); + + Type dateType = Type.fromTsDataType(TSDataType.DATE); + Column dateColumn = dateType.createColumnBuilder(1).writeInt(20240801).build(); + Assert.assertEquals(DateUtils.formatDate(20240801), readRpcObject(dateType, dateColumn, 1_000)); + Assert.assertEquals(DateUtils.formatDate(20240801), readRpcString(dateType, dateColumn)); + + byte[] blob = new byte[] {1, 2, 3}; + Type blobType = Type.fromTsDataType(TSDataType.BLOB); + Column blobColumn = blobType.createColumnBuilder(1).writeBinary(new Binary(blob)).build(); + Assert.assertEquals( + BytesUtils.parseBlobByteArrayToString(blob), readRpcObject(blobType, blobColumn, 1_000)); + Assert.assertEquals( + BytesUtils.parseBlobByteArrayToString(blob), readRpcString(blobType, blobColumn)); + + byte[] object = ByteBuffer.allocate(16).putLong(0).putLong(42).array(); + Type objectType = Type.fromTsDataType(TSDataType.OBJECT); + Column objectColumn = objectType.createColumnBuilder(1).writeBinary(new Binary(object)).build(); + Assert.assertEquals( + BytesUtils.parseObjectByteArrayToString(object), + readRpcObject(objectType, objectColumn, 1_000)); + Assert.assertEquals( + BytesUtils.parseObjectByteArrayToString(object), readRpcString(objectType, objectColumn)); + + Assert.assertNull( + TypeServices.RPC_OBJECT_READER_SERVICE + .call(UnknownType.UNKNOWN) + .read(UnknownType.UNKNOWN, null, 0, 1_000)); + Assert.assertNull( + TypeServices.RPC_STRING_READER_SERVICE + .call(UnknownType.UNKNOWN) + .read(UnknownType.UNKNOWN, null, 0, "long", "ms", ZoneId.of("UTC"))); + } + + private static Object readRpcObject(Type type, Column column, int timeFactor) { + return TypeServices.RPC_OBJECT_READER_SERVICE.call(type).read(type, column, 0, timeFactor); + } + + private static String readRpcString(Type type, Column column) { + return TypeServices.RPC_STRING_READER_SERVICE + .call(type) + .read(type, column, 0, "long", "ms", ZoneId.of("UTC")); + } + private static void assertUnsupported(TSDataType dataType) { UnSupportedDataTypeException exception = Assert.assertThrows( diff --git a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerdeTest.java b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerdeTest.java index cd3929c8224..20de227def4 100644 --- a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerdeTest.java +++ b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerdeTest.java @@ -33,6 +33,7 @@ import static org.apache.iotdb.rpc.stmt.PreparedParameterSerde.serialize; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; /** Unit tests for {@link PreparedParameterSerde}. */ @@ -124,4 +125,15 @@ public class PreparedParameterSerdeTest { buffer.flip(); deserialize(buffer); } + + @Test + public void testUnsupportedType() { + ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES + Byte.BYTES + Integer.BYTES); + buffer.putInt(1); + TSDataType.DATE.serializeTo(buffer); + buffer.putInt(20240801); + buffer.flip(); + + assertThrows(IllegalArgumentException.class, () -> deserialize(buffer)); + } }
