COLUMN_VALUE_GETTER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case INT32 -> Column::getInt;
+ case INT64 -> Column::getLong;
+ case FLOAT -> Column::getFloat;
+ case DOUBLE -> Column::getDouble;
+ default ->
+ (column, index) -> {
+ throw new UnSupportedDataTypeException(
+ String.format(
+ "Unsupported data type in aggregation AVG : %s", type.getTypeEnum()))
+ .setChecked(false);
+ };
+ };
+
static class SumState implements State {
double sum = 0;
@@ -63,7 +82,7 @@ public void deserialize(byte[] bytes) {
}
}
- private Type dataType;
+ private org.apache.tsfile.read.common.type.Type dataType;
@Override
public void validate(UDFParameterValidator validator) throws UDFException {
@@ -74,7 +93,9 @@ public void validate(UDFParameterValidator validator) throws UDFException {
@Override
public void beforeStart(UDFParameters parameters, UDAFConfigurations configurations) {
- dataType = parameters.getDataType(0);
+ dataType =
+ org.apache.tsfile.read.common.type.Type.fromTsDataType(
+ UDFDataTypeTransformer.transformToTsDataType(parameters.getDataType(0)));
configurations.setOutputDataType(Type.DOUBLE);
}
@@ -86,29 +107,17 @@ public State createState() {
@Override
public void addInput(State state, Column[] columns, BitMap bitMap) {
SumState sumState = (SumState) state;
-
- switch (dataType) {
- case INT32:
- addIntInput(sumState, columns, bitMap);
- return;
- case INT64:
- addLongInput(sumState, columns, bitMap);
- return;
- case FLOAT:
- addFloatInput(sumState, columns, bitMap);
- return;
- case DOUBLE:
- addDoubleInput(sumState, columns, bitMap);
- return;
- case TEXT:
- case STRING:
- case BLOB:
- case BOOLEAN:
- case TIMESTAMP:
- case DATE:
- default:
- throw new UnSupportedDataTypeException(
- String.format("Unsupported data type in aggregation AVG : %s", dataType));
+ final Column column = columns[0];
+ final ColumnValueGetter valueGetter = COLUMN_VALUE_GETTER_SERVICE.call(dataType);
+ final int count = column.getPositionCount();
+ for (int i = 0; i < count; i++) {
+ if (bitMap != null && !bitMap.isMarked(i)) {
+ continue;
+ }
+ if (!column.isNull(i)) {
+ sumState.initResult = true;
+ sumState.sum += valueGetter.getValue(column, i);
+ }
}
}
@@ -140,55 +149,8 @@ public void removeState(State state, State removed) {
sumState.sum -= sumRhs.sum;
}
- private void addIntInput(SumState state, Column[] columns, BitMap bitMap) {
- int count = columns[0].getPositionCount();
- for (int i = 0; i < count; i++) {
- if (bitMap != null && !bitMap.isMarked(i)) {
- continue;
- }
- if (!columns[0].isNull(i)) {
- state.initResult = true;
- state.sum += columns[0].getInt(i);
- }
- }
- }
-
- private void addLongInput(SumState state, Column[] columns, BitMap bitMap) {
- int count = columns[0].getPositionCount();
- for (int i = 0; i < count; i++) {
- if (bitMap != null && !bitMap.isMarked(i)) {
- continue;
- }
- if (!columns[0].isNull(i)) {
- state.initResult = true;
- state.sum += columns[0].getLong(i);
- }
- }
- }
-
- private void addFloatInput(SumState state, Column[] columns, BitMap bitMap) {
- int count = columns[0].getPositionCount();
- for (int i = 0; i < count; i++) {
- if (bitMap != null && !bitMap.isMarked(i)) {
- continue;
- }
- if (!columns[0].isNull(i)) {
- state.initResult = true;
- state.sum += columns[0].getFloat(i);
- }
- }
- }
-
- private void addDoubleInput(SumState state, Column[] columns, BitMap bitMap) {
- int count = columns[0].getPositionCount();
- for (int i = 0; i < count; i++) {
- if (bitMap != null && !bitMap.isMarked(i)) {
- continue;
- }
- if (!columns[0].isNull(i)) {
- state.initResult = true;
- state.sum += columns[0].getDouble(i);
- }
- }
+ @FunctionalInterface
+ private interface ColumnValueGetter {
+ double getValue(Column column, int index);
}
}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/access/Row.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/access/Row.java
index e2dc13a39ab2..c1b466786cdc 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/access/Row.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/access/Row.java
@@ -68,10 +68,11 @@ public interface Row {
float getFloat(int columnIndex) throws IOException;
/**
- * Returns the double value at the specified column in this row.
+ * Returns the numeric value at the specified column in this row as a double.
*
- * Users need to ensure that the data type of the specified column is {@code
- * TSDataType.DOUBLE}.
+ *
Users need to ensure that the data type of the specified column is {@code TSDataType.INT32},
+ * {@code TSDataType.INT64}, {@code TSDataType.FLOAT}, or {@code TSDataType.DOUBLE}, and that the
+ * value is not null. INT64 values may lose precision when converted to double.
*
* @param columnIndex index of the specified column
* @return the double value at the specified column in this row
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/utils/RowImpl.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/utils/RowImpl.java
index a082a1968ce6..4033ff9ac761 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/utils/RowImpl.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/utils/RowImpl.java
@@ -73,7 +73,7 @@ public double getDouble(int columnIndex) {
if (columnIndex >= size()) {
throw new IndexOutOfBoundsException(UdfApiMessages.INDEX_OUT_OF_BOUND);
}
- return (double) rowRecord[columnIndex];
+ return ((Number) rowRecord[columnIndex]).doubleValue();
}
@Override
@@ -129,30 +129,10 @@ private static Type transformToUDFDataType(TSDataType tsDataType) {
if (tsDataType == null) {
return null;
}
- byte type = tsDataType.getType();
- switch (type) {
- case 0:
- return Type.BOOLEAN;
- case 1:
- return Type.INT32;
- case 2:
- return Type.INT64;
- case 3:
- return Type.FLOAT;
- case 4:
- return Type.DOUBLE;
- case 5:
- return Type.TEXT;
- case 8:
- return Type.TIMESTAMP;
- case 9:
- return Type.DATE;
- case 10:
- return Type.BLOB;
- case 11:
- return Type.STRING;
- default:
- throw new IllegalArgumentException(UdfApiMessages.INVALID_INPUT + type);
+ try {
+ return Type.valueOf(tsDataType.getType());
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(UdfApiMessages.INVALID_INPUT + tsDataType.getType(), e);
}
}
}
diff --git a/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java b/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
index 0329812f2d4d..0b0f30d970d9 100644
--- a/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
+++ b/iotdb-client/cli/src/main/i18n/en/org/apache/iotdb/cli/i18n/CliMessages.java
@@ -50,6 +50,10 @@ public final class CliMessages {
public static final String FAILED_TO_WRITE_DATA = "Failed to write data to file: {}";
public static final String FAILED_TO_CREATE_FILE = "Failed to create file: {}";
+ // AbstractCli
+ public static final String EXCEPTION_UNSUPPORTED_DATA_TYPE_ARG_B411C29E =
+ "Unsupported data type: %s";
+
// AbstractDataTool
public static final String USE_HELP_FOR_MORE = "Use -help for more information";
diff --git a/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java b/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
index 61a14d94d588..a260732ab554 100644
--- a/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
+++ b/iotdb-client/cli/src/main/i18n/zh/org/apache/iotdb/cli/i18n/CliMessages.java
@@ -50,6 +50,10 @@ public final class CliMessages {
public static final String FAILED_TO_WRITE_DATA = "向文件写入数据失败:{}";
public static final String FAILED_TO_CREATE_FILE = "创建文件失败:{}";
+ // AbstractCli
+ public static final String EXCEPTION_UNSUPPORTED_DATA_TYPE_ARG_B411C29E =
+ "不支持的数据类型:%s";
+
// AbstractDataTool
public static final String USE_HELP_FOR_MORE = "使用 -help 获取更多信息";
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java
index 088131c25702..aad1074860fb 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/cli/AbstractCli.java
@@ -35,8 +35,11 @@
import org.apache.commons.cli.Options;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.external.commons.lang3.ArrayUtils;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.read.common.type.service.TypeService;
import org.apache.tsfile.utils.BytesUtils;
import org.apache.tsfile.utils.DateUtils;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
import java.io.BufferedReader;
import java.io.InputStreamReader;
@@ -56,6 +59,38 @@
public abstract class AbstractCli {
+ static String timestampPrecision = "ms";
+ static String timeFormat = RpcUtils.DEFAULT_TIME_FORMAT;
+
+ private static final TypeService SQL_VALUE_GETTER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN, INT32, INT64, FLOAT, DOUBLE, TEXT, STRING, OBJECT ->
+ (resultSet, columnIndex, zoneId) -> resultSet.getString(columnIndex);
+ case BLOB ->
+ (resultSet, columnIndex, zoneId) -> {
+ byte[] value = resultSet.getBytes(columnIndex);
+ return value == null ? null : BytesUtils.parseBlobByteArrayToString(value);
+ };
+ case DATE ->
+ (resultSet, columnIndex, zoneId) -> {
+ int value = resultSet.getInt(columnIndex);
+ return resultSet.wasNull() ? null : DateUtils.formatDate(value);
+ };
+ case TIMESTAMP ->
+ (resultSet, columnIndex, zoneId) -> {
+ long value = resultSet.getLong(columnIndex);
+ return resultSet.wasNull()
+ ? null
+ : RpcUtils.formatDatetime(timeFormat, timestampPrecision, value, zoneId);
+ };
+ case ROW, UNKNOWN, VECTOR ->
+ throw new UnSupportedDataTypeException(
+ String.format(
+ CliMessages.EXCEPTION_UNSUPPORTED_DATA_TYPE_ARG_B411C29E,
+ type.getTypeEnum()));
+ };
+
static final String HOST_ARGS = "h";
static final String HOST_NAME = "host";
@@ -119,8 +154,6 @@ public abstract class AbstractCli {
static int maxPrintRowCount = 1000;
private static int fetchSize = 1000;
static int queryTimeout = 0;
- static String timestampPrecision = "ms";
- static String timeFormat = RpcUtils.DEFAULT_TIME_FORMAT;
private static boolean continuePrint = false;
private static int lineCount = 0;
@@ -794,40 +827,14 @@ private static List> cacheResult(
private static String getStringByColumnIndex(
IoTDBJDBCResultSet resultSet, int columnIndex, ZoneId zoneId) throws SQLException {
TSDataType type = resultSet.getColumnTypeByIndex(columnIndex);
- switch (type) {
- case BOOLEAN:
- case INT32:
- case INT64:
- case FLOAT:
- case DOUBLE:
- case TEXT:
- case STRING:
- case OBJECT:
- return resultSet.getString(columnIndex);
- case BLOB:
- byte[] v = resultSet.getBytes(columnIndex);
- if (v == null) {
- return null;
- } else {
- return BytesUtils.parseBlobByteArrayToString(v);
- }
- case DATE:
- int intValue = resultSet.getInt(columnIndex);
- if (resultSet.wasNull()) {
- return null;
- } else {
- return DateUtils.formatDate(intValue);
- }
- case TIMESTAMP:
- long longValue = resultSet.getLong(columnIndex);
- if (resultSet.wasNull()) {
- return null;
- } else {
- return RpcUtils.formatDatetime(timeFormat, timestampPrecision, longValue, zoneId);
- }
- default:
- return null;
- }
+ return SQL_VALUE_GETTER_SERVICE
+ .call(Type.fromTsDataType(type))
+ .get(resultSet, columnIndex, zoneId);
+ }
+
+ @FunctionalInterface
+ private interface SqlValueGetter {
+ String get(ResultSet resultSet, int columnIndex, ZoneId zoneId) throws SQLException;
}
private static List> cacheTracingInfo(ResultSet resultSet, List maxSizeList)
diff --git a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java
index b928fbd98674..06d02a7de834 100644
--- a/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java
+++ b/iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/AbstractDataTool.java
@@ -56,7 +56,10 @@
import org.apache.tsfile.external.commons.lang3.StringUtils;
import org.apache.tsfile.read.common.Field;
import org.apache.tsfile.read.common.RowRecord;
+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.write.UnSupportedDataTypeException;
import org.jline.reader.LineReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -88,6 +91,35 @@
public abstract class AbstractDataTool {
+ private static final TypeService VALUE_PARSER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case TEXT, STRING ->
+ value -> {
+ if (value.startsWith("\"") && value.endsWith("\"")) {
+ return value.substring(1, value.length() - 1);
+ }
+ return value;
+ };
+ case BOOLEAN ->
+ value ->
+ !"true".equalsIgnoreCase(value) && !"false".equalsIgnoreCase(value)
+ ? null
+ : Boolean.parseBoolean(value);
+ case INT32 -> value -> Integer.parseInt(value);
+ case INT64, TIMESTAMP -> value -> Long.parseLong(value);
+ case FLOAT -> value -> Float.parseFloat(value);
+ case DOUBLE -> value -> Double.parseDouble(value);
+ case DATE -> value -> LocalDate.parse(value);
+ case BLOB ->
+ value -> new Binary(parseHexStringToByteArray(value.replaceFirst("0x", "")));
+ case OBJECT, ROW, UNKNOWN, VECTOR ->
+ throw new UnSupportedDataTypeException(
+ String.format(
+ CliMessages.EXCEPTION_UNSUPPORTED_DATA_TYPE_ARG_B411C29E,
+ type.getTypeEnum()));
+ };
+
protected static String host;
protected static String port;
protected static String table;
@@ -467,40 +499,17 @@ private static boolean isConvertFloatPrecisionLack(String s) {
*/
protected static Object typeTrans(String value, TSDataType type) {
try {
- switch (type) {
- case TEXT:
- case STRING:
- if (value.startsWith("\"") && value.endsWith("\"")) {
- return value.substring(1, value.length() - 1);
- }
- return value;
- case BOOLEAN:
- if (!"true".equalsIgnoreCase(value) && !"false".equalsIgnoreCase(value)) {
- return null;
- }
- return Boolean.parseBoolean(value);
- case INT32:
- return Integer.parseInt(value);
- case INT64:
- return Long.parseLong(value);
- case FLOAT:
- return Float.parseFloat(value);
- case DOUBLE:
- return Double.parseDouble(value);
- case TIMESTAMP:
- return Long.parseLong(value);
- case DATE:
- return LocalDate.parse(value);
- case BLOB:
- return new Binary(parseHexStringToByteArray(value.replaceFirst("0x", "")));
- default:
- return null;
- }
+ return VALUE_PARSER_SERVICE.call(Type.fromTsDataType(type)).parse(value);
} catch (NumberFormatException e) {
return null;
}
}
+ @FunctionalInterface
+ private interface ValueParser {
+ Object parse(String value);
+ }
+
private static byte[] parseHexStringToByteArray(String hexString) {
byte[] bytes = new byte[hexString.length() / 2];
for (int i = 0; i < hexString.length(); i += 2) {
diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/SessionDataSet.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/SessionDataSet.java
index a926103222a2..5076afd52d05 100644
--- a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/SessionDataSet.java
+++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/SessionDataSet.java
@@ -30,8 +30,8 @@
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.Field;
import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.common.type.Type;
import org.apache.tsfile.utils.Binary;
-import org.apache.tsfile.write.UnSupportedDataTypeException;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
@@ -161,40 +161,9 @@ private RowRecord constructRowRecordFromValueArray() throws StatementExecutionEx
if (!ioTDBRpcDataSet.isNull(columnIndex)) {
TSDataType dataType = ioTDBRpcDataSet.getDataType(columnIndex);
field = new Field(dataType);
- switch (dataType) {
- case BOOLEAN:
- boolean booleanValue = ioTDBRpcDataSet.getBoolean(columnIndex);
- field.setBoolV(booleanValue);
- break;
- case INT32:
- case DATE:
- int intValue = ioTDBRpcDataSet.getInt(columnIndex);
- field.setIntV(intValue);
- break;
- case INT64:
- case TIMESTAMP:
- long longValue = ioTDBRpcDataSet.getLong(columnIndex);
- field.setLongV(longValue);
- break;
- case FLOAT:
- float floatValue = ioTDBRpcDataSet.getFloat(columnIndex);
- field.setFloatV(floatValue);
- break;
- case DOUBLE:
- double doubleValue = ioTDBRpcDataSet.getDouble(columnIndex);
- field.setDoubleV(doubleValue);
- break;
- case TEXT:
- case BLOB:
- case STRING:
- case OBJECT:
- field.setBinaryV(ioTDBRpcDataSet.getBinary(columnIndex));
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format(
- ISessionMessages.EXCEPTION_DATA_TYPE_ARG_NOT_SUPPORTED_31213160, dataType));
- }
+ TypeServices.FIELD_VALUE_READER_SERVICE
+ .call(Type.fromTsDataType(dataType))
+ .read(ioTDBRpcDataSet, columnIndex, field);
} else {
field = new Field(null);
}
diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/TypeServices.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/TypeServices.java
new file mode 100644
index 000000000000..09b08977d099
--- /dev/null
+++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/TypeServices.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iotdb.isession;
+
+import org.apache.iotdb.isession.i18n.ISessionMessages;
+import org.apache.iotdb.rpc.IoTDBRpcDataSet;
+import org.apache.iotdb.rpc.StatementExecutionException;
+
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.read.common.type.service.TypeService;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+
+/** Type-specific RPC result readers that preserve primitive access without intermediate boxing. */
+final class TypeServices {
+
+ static final TypeService FIELD_VALUE_READER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN ->
+ (dataSet, columnIndex, field) -> field.setBoolV(dataSet.getBoolean(columnIndex));
+ case INT32, DATE ->
+ (dataSet, columnIndex, field) -> field.setIntV(dataSet.getInt(columnIndex));
+ case INT64, TIMESTAMP ->
+ (dataSet, columnIndex, field) -> field.setLongV(dataSet.getLong(columnIndex));
+ case FLOAT ->
+ (dataSet, columnIndex, field) -> field.setFloatV(dataSet.getFloat(columnIndex));
+ case DOUBLE ->
+ (dataSet, columnIndex, field) -> field.setDoubleV(dataSet.getDouble(columnIndex));
+ case TEXT, BLOB, STRING, OBJECT ->
+ (dataSet, columnIndex, field) -> field.setBinaryV(dataSet.getBinary(columnIndex));
+ case ROW, UNKNOWN, VECTOR ->
+ (dataSet, columnIndex, field) -> {
+ throw new UnSupportedDataTypeException(
+ String.format(
+ ISessionMessages.EXCEPTION_DATA_TYPE_ARG_NOT_SUPPORTED_31213160,
+ type.getTypeEnum()));
+ };
+ };
+
+ static {
+ FIELD_VALUE_READER_SERVICE.check();
+ }
+
+ private TypeServices() {}
+
+ @FunctionalInterface
+ interface FieldValueReader {
+
+ void read(IoTDBRpcDataSet dataSet, int columnIndex, Field field)
+ throws StatementExecutionException;
+ }
+}
diff --git a/iotdb-client/jdbc/src/main/i18n/en/org/apache/iotdb/jdbc/i18n/JdbcMessages.java b/iotdb-client/jdbc/src/main/i18n/en/org/apache/iotdb/jdbc/i18n/JdbcMessages.java
index 43e0ef6cb46f..dc26ffc77b58 100644
--- a/iotdb-client/jdbc/src/main/i18n/en/org/apache/iotdb/jdbc/i18n/JdbcMessages.java
+++ b/iotdb-client/jdbc/src/main/i18n/en/org/apache/iotdb/jdbc/i18n/JdbcMessages.java
@@ -91,6 +91,8 @@ public final class JdbcMessages {
// IoTDBAbstractDatabaseMetadata
public static final String NO_DATA_TYPE_MATCHED = "No data type was matched: {}";
+ public static final String EXCEPTION_UNSUPPORTED_DATA_TYPE_ARG_B411C29E =
+ "Unsupported data type: %s";
public static final String GET_READ_ONLY_ERROR = "Get is readOnly error: {}";
public static final String CANNOT_GET_READ_ONLY_MODE = "Can not get the read-only mode";
public static final String GET_SYSTEM_FUNCTIONS_ERROR = "Get system functions error: {}";
diff --git a/iotdb-client/jdbc/src/main/i18n/zh/org/apache/iotdb/jdbc/i18n/JdbcMessages.java b/iotdb-client/jdbc/src/main/i18n/zh/org/apache/iotdb/jdbc/i18n/JdbcMessages.java
index 1aeededfc900..7d5fb9ec2fe0 100644
--- a/iotdb-client/jdbc/src/main/i18n/zh/org/apache/iotdb/jdbc/i18n/JdbcMessages.java
+++ b/iotdb-client/jdbc/src/main/i18n/zh/org/apache/iotdb/jdbc/i18n/JdbcMessages.java
@@ -91,6 +91,7 @@ public final class JdbcMessages {
// IoTDBAbstractDatabaseMetadata
public static final String NO_DATA_TYPE_MATCHED = "没有匹配的数据类型:{}";
+ public static final String EXCEPTION_UNSUPPORTED_DATA_TYPE_ARG_B411C29E = "不支持的数据类型:%s";
public static final String GET_READ_ONLY_ERROR = "获取只读模式错误:{}";
public static final String CANNOT_GET_READ_ONLY_MODE = "无法获取只读模式";
public static final String GET_SYSTEM_FUNCTIONS_ERROR = "获取系统函数错误:{}";
diff --git a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/GroupedLSBWatermarkEncoder.java b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/GroupedLSBWatermarkEncoder.java
index 58579f7d437d..9f56f34e4e82 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/GroupedLSBWatermarkEncoder.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/GroupedLSBWatermarkEncoder.java
@@ -22,9 +22,10 @@
import org.apache.iotdb.jdbc.i18n.JdbcMessages;
import org.apache.thrift.EncodingUtils;
-import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.Field;
import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.common.type.Type;
+import org.apache.tsfile.read.common.type.service.TypeService;
import java.math.BigInteger;
import java.security.MessageDigest;
@@ -32,6 +33,25 @@
import java.util.List;
public class GroupedLSBWatermarkEncoder implements WatermarkEncoder {
+
+ private static final TypeService FIELD_ENCODER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case INT32, DATE ->
+ (encoder, field, timestamp) ->
+ field.setIntV(encoder.encodeInt(field.getIntV(), timestamp));
+ case INT64, TIMESTAMP ->
+ (encoder, field, timestamp) ->
+ field.setLongV(encoder.encodeLong(field.getLongV(), timestamp));
+ case FLOAT ->
+ (encoder, field, timestamp) ->
+ field.setFloatV(encoder.encodeFloat(field.getFloatV(), timestamp));
+ case DOUBLE ->
+ (encoder, field, timestamp) ->
+ field.setDoubleV(encoder.encodeDouble(field.getDoubleV(), timestamp));
+ default -> (encoder, field, timestamp) -> {};
+ };
+
private String secretKey;
private String bitString;
private int markRate = 2;
@@ -116,33 +136,15 @@ public RowRecord encodeRecord(RowRecord rowRecord) {
if (field == null || field.getDataType() == null) {
continue;
}
- TSDataType dataType = field.getDataType();
- switch (dataType) {
- case INT32:
- case DATE:
- int originIntValue = field.getIntV();
- field.setIntV(encodeInt(originIntValue, timestamp));
- break;
- case INT64:
- case TIMESTAMP:
- long originLongValue = field.getLongV();
- field.setLongV(encodeLong(originLongValue, timestamp));
- break;
- case FLOAT:
- float originFloatValue = field.getFloatV();
- field.setFloatV(encodeFloat(originFloatValue, timestamp));
- break;
- case DOUBLE:
- double originDoubleValue = field.getDoubleV();
- field.setDoubleV(encodeDouble(originDoubleValue, timestamp));
- break;
- case BLOB:
- case STRING:
- case BOOLEAN:
- case TEXT:
- default:
- }
+ FIELD_ENCODER_SERVICE
+ .call(Type.fromTsDataType(field.getDataType()))
+ .encode(this, field, timestamp);
}
return rowRecord;
}
+
+ @FunctionalInterface
+ private interface FieldEncoder {
+ void encode(GroupedLSBWatermarkEncoder encoder, Field field, long timestamp);
+ }
}
diff --git a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBAbstractDatabaseMetadata.java b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBAbstractDatabaseMetadata.java
index 0f81bca6069a..0c50512e80b0 100644
--- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBAbstractDatabaseMetadata.java
+++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBAbstractDatabaseMetadata.java
@@ -23,12 +23,16 @@
import org.apache.iotdb.service.rpc.thrift.IClientRPCService;
import org.apache.thrift.TException;
+import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.read.common.block.column.TsBlockSerde;
+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.write.UnSupportedDataTypeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +56,25 @@
public abstract class IoTDBAbstractDatabaseMetadata implements DatabaseMetaData {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBAbstractDatabaseMetadata.class);
+
+ private static final TypeService COLUMN_VALUE_WRITER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case TEXT, STRING, BLOB, OBJECT ->
+ (builder, value) ->
+ builder.writeBinary(new Binary(value.toString(), TSFileConfig.STRING_CHARSET));
+ case FLOAT -> (builder, value) -> builder.writeFloat((float) value);
+ case INT32, DATE -> (builder, value) -> builder.writeInt((int) value);
+ case INT64, TIMESTAMP -> (builder, value) -> builder.writeLong((long) value);
+ case DOUBLE -> (builder, value) -> builder.writeDouble((double) value);
+ case BOOLEAN -> (builder, value) -> builder.writeBoolean((boolean) value);
+ case ROW, UNKNOWN, VECTOR ->
+ throw new UnSupportedDataTypeException(
+ String.format(
+ JdbcMessages.EXCEPTION_UNSUPPORTED_DATA_TYPE_ARG_B411C29E,
+ type.getTypeEnum()));
+ };
+
private static final String METHOD_NOT_SUPPORTED_STRING = JdbcMessages.METHOD_NOT_SUPPORTED;
protected static final String CONVERT_ERROR_MSG = "Convert tsBlock error: {}";
@@ -393,37 +416,9 @@ public static ByteBuffer convertTsBlock(
tsBlockBuilder.getTimeColumnBuilder().writeLong(0);
for (int j = 0; j < tsDataTypeList.size(); j++) {
TSDataType columnType = tsDataTypeList.get(j);
- switch (columnType) {
- case TEXT:
- case STRING:
- case BLOB:
- case OBJECT:
- tsBlockBuilder
- .getColumnBuilder(j)
- .writeBinary(
- new Binary(valuesInRow.get(j).toString(), TSFileConfig.STRING_CHARSET));
- break;
- case FLOAT:
- tsBlockBuilder.getColumnBuilder(j).writeFloat((float) valuesInRow.get(j));
- break;
- case INT32:
- case DATE:
- tsBlockBuilder.getColumnBuilder(j).writeInt((int) valuesInRow.get(j));
- break;
- case INT64:
- case TIMESTAMP:
- tsBlockBuilder.getColumnBuilder(j).writeLong((long) valuesInRow.get(j));
- break;
- case DOUBLE:
- tsBlockBuilder.getColumnBuilder(j).writeDouble((double) valuesInRow.get(j));
- break;
- case BOOLEAN:
- tsBlockBuilder.getColumnBuilder(j).writeBoolean((boolean) valuesInRow.get(j));
- break;
- default:
- LOGGER.error(JdbcMessages.NO_DATA_TYPE_MATCHED, columnType);
- break;
- }
+ COLUMN_VALUE_WRITER_SERVICE
+ .call(Type.fromTsDataType(columnType))
+ .write(tsBlockBuilder.getColumnBuilder(j), valuesInRow.get(j));
}
tsBlockBuilder.declarePosition();
}
@@ -435,6 +430,11 @@ public static ByteBuffer convertTsBlock(
}
}
+ @FunctionalInterface
+ private interface ColumnValueWriter {
+ void write(ColumnBuilder builder, Object value);
+ }
+
protected void close(ResultSet rs, Statement stmt) {
try {
if (rs != null) {
diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBJDBCDataSet.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBJDBCDataSet.java
index 75ff09a882e4..b4516efa7c81 100644
--- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBJDBCDataSet.java
+++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBJDBCDataSet.java
@@ -29,14 +29,12 @@
import org.apache.thrift.TException;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.read.common.type.Type;
import org.apache.tsfile.utils.BytesUtils;
-import org.apache.tsfile.utils.DateUtils;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.apache.tsfile.write.UnSupportedDataTypeException;
import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.ZoneId;
import java.util.ArrayList;
@@ -156,38 +154,7 @@ public IoTDBJDBCDataSet(
time = new byte[Long.BYTES];
currentBitmap = new byte[columnTypeDeduplicatedList.size()];
- values = new byte[columnTypeDeduplicatedList.size()][];
- for (int i = 0; i < values.length; i++) {
- TSDataType dataType = columnTypeDeduplicatedList.get(i);
- switch (dataType) {
- case BOOLEAN:
- values[i] = new byte[1];
- break;
- case INT32:
- case DATE:
- values[i] = new byte[Integer.BYTES];
- break;
- case INT64:
- case TIMESTAMP:
- values[i] = new byte[Long.BYTES];
- break;
- case FLOAT:
- values[i] = new byte[Float.BYTES];
- break;
- case DOUBLE:
- values[i] = new byte[Double.BYTES];
- break;
- case TEXT:
- case BLOB:
- case STRING:
- case OBJECT:
- values[i] = null;
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format(DATA_TYPE_NOT_SUPPORTED, columnTypeDeduplicatedList.get(i)));
- }
- }
+ values = initializeValueBuffers(columnTypeDeduplicatedList);
this.tsQueryDataSet = queryDataSet;
this.emptyResultSet = (queryDataSet == null || !queryDataSet.time.hasRemaining());
}
@@ -275,38 +242,7 @@ public IoTDBJDBCDataSet(
time = new byte[Long.BYTES];
currentBitmap = new byte[columnTypeDeduplicatedList.size()];
- values = new byte[columnTypeDeduplicatedList.size()][];
- for (int i = 0; i < values.length; i++) {
- TSDataType dataType = columnTypeDeduplicatedList.get(i);
- switch (dataType) {
- case BOOLEAN:
- values[i] = new byte[1];
- break;
- case INT32:
- case DATE:
- values[i] = new byte[Integer.BYTES];
- break;
- case INT64:
- case TIMESTAMP:
- values[i] = new byte[Long.BYTES];
- break;
- case FLOAT:
- values[i] = new byte[Float.BYTES];
- break;
- case DOUBLE:
- values[i] = new byte[Double.BYTES];
- break;
- case TEXT:
- case BLOB:
- case STRING:
- case OBJECT:
- values[i] = null;
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format(DATA_TYPE_NOT_SUPPORTED, columnTypeDeduplicatedList.get(i)));
- }
- }
+ values = initializeValueBuffers(columnTypeDeduplicatedList);
this.tsQueryDataSet = queryDataSet;
this.emptyResultSet = (queryDataSet == null || !queryDataSet.time.hasRemaining());
}
@@ -421,26 +357,11 @@ public void constructOneRow() {
if (!isNull(i, rowsIndex)) {
ByteBuffer valueBuffer = tsQueryDataSet.valueList.get(i);
TSDataType dataType = columnTypeDeduplicatedList.get(i);
- switch (dataType) {
- case BOOLEAN:
- case INT32:
- case INT64:
- case FLOAT:
- case DOUBLE:
- case DATE:
- case TIMESTAMP:
- valueBuffer.get(values[i]);
- break;
- case TEXT:
- case BLOB:
- case STRING:
- case OBJECT:
- int length = valueBuffer.getInt();
- values[i] = ReadWriteIOUtils.readBytes(valueBuffer, length);
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format(DATA_TYPE_NOT_SUPPORTED, columnTypeDeduplicatedList.get(i)));
+ if (dataType.isBinary()) {
+ int length = valueBuffer.getInt();
+ values[i] = ReadWriteIOUtils.readBytes(valueBuffer, length);
+ } else {
+ valueBuffer.get(values[i]);
}
}
}
@@ -448,6 +369,34 @@ public void constructOneRow() {
hasCachedRecord = true;
}
+ static byte[][] initializeValueBuffers(List dataTypes) {
+ byte[][] valueBuffers = new byte[dataTypes.size()][];
+ for (int i = 0; i < dataTypes.size(); i++) {
+ TSDataType dataType = dataTypes.get(i);
+ if (dataType == TSDataType.VECTOR || dataType == TSDataType.UNKNOWN) {
+ throw unsupportedDataType(dataType);
+ }
+ if (dataType.isBinary()) {
+ continue;
+ }
+ final int dataTypeSize;
+ try {
+ dataTypeSize = dataType.getDataTypeSize();
+ } catch (UnSupportedDataTypeException e) {
+ throw unsupportedDataType(dataType);
+ }
+ if (dataTypeSize == 0) {
+ throw unsupportedDataType(dataType);
+ }
+ valueBuffers[i] = new byte[dataTypeSize];
+ }
+ return valueBuffers;
+ }
+
+ private static UnSupportedDataTypeException unsupportedDataType(TSDataType dataType) {
+ return new UnSupportedDataTypeException(String.format(DATA_TYPE_NOT_SUPPORTED, dataType));
+ }
+
public boolean isNull(int columnIndex) throws StatementExecutionException {
int index = columnOrdinalMap.get(findColumnNameByIndex(columnIndex)) - START_INDEX;
// time column will never be null
@@ -598,30 +547,9 @@ public String getValueByName(String columnName) throws StatementExecutionExcepti
}
public String getString(int index, TSDataType tsDataType, byte[][] values) {
- switch (tsDataType) {
- case BOOLEAN:
- return String.valueOf(BytesUtils.bytesToBool(values[index]));
- case INT32:
- return String.valueOf(BytesUtils.bytesToInt(values[index]));
- case INT64:
- case TIMESTAMP:
- return String.valueOf(BytesUtils.bytesToLong(values[index]));
- case FLOAT:
- return String.valueOf(BytesUtils.bytesToFloat(values[index]));
- case DOUBLE:
- return String.valueOf(BytesUtils.bytesToDouble(values[index]));
- case TEXT:
- case STRING:
- return new String(values[index], StandardCharsets.UTF_8);
- case OBJECT:
- return BytesUtils.parseObjectByteArrayToString(values[index]);
- case BLOB:
- return BytesUtils.parseBlobByteArrayToString(values[index]);
- case DATE:
- return DateUtils.formatDate(BytesUtils.bytesToInt(values[index]));
- default:
- return null;
- }
+ return TypeServices.JDBC_STRING_READER_SERVICE
+ .call(Type.fromTsDataType(tsDataType))
+ .apply(values[index]);
}
public Object getObjectByName(String columnName) throws StatementExecutionException {
@@ -639,30 +567,9 @@ public Object getObjectByName(String columnName) throws StatementExecutionExcept
}
public Object getObject(int index, TSDataType tsDataType, byte[][] values) {
- switch (tsDataType) {
- case BOOLEAN:
- return BytesUtils.bytesToBool(values[index]);
- case INT32:
- return BytesUtils.bytesToInt(values[index]);
- case INT64:
- return BytesUtils.bytesToLong(values[index]);
- case FLOAT:
- return BytesUtils.bytesToFloat(values[index]);
- case DOUBLE:
- return BytesUtils.bytesToDouble(values[index]);
- case TEXT:
- case STRING:
- return new String(values[index], StandardCharsets.UTF_8);
- case OBJECT:
- case BLOB:
- return new Binary(values[index]);
- case TIMESTAMP:
- return new Timestamp(BytesUtils.bytesToLong(values[index]));
- case DATE:
- return DateUtils.parseIntToDate(BytesUtils.bytesToInt(values[index]));
- default:
- return null;
- }
+ return TypeServices.JDBC_OBJECT_READER_SERVICE
+ .call(Type.fromTsDataType(tsDataType))
+ .apply(values[index]);
}
public String findColumnNameByIndex(int columnIndex) throws StatementExecutionException {
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 56200399b4e6..32f5ea5d45a3 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.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;
@@ -495,37 +495,10 @@ private Object getObjectByTsBlockIndex(int tsBlockColumnIndex)
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 {
@@ -548,47 +521,22 @@ private String getStringByTsBlockColumnIndex(int tsBlockColumnIndex)
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
new file mode 100644
index 000000000000..4b0311243723
--- /dev/null
+++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TypeServices.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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;
+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 {
+
+ static final TypeService> JDBC_STRING_READER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN -> value -> String.valueOf(BytesUtils.bytesToBool(value));
+ case INT32 -> value -> String.valueOf(BytesUtils.bytesToInt(value));
+ case INT64, TIMESTAMP -> value -> String.valueOf(BytesUtils.bytesToLong(value));
+ case FLOAT -> value -> String.valueOf(BytesUtils.bytesToFloat(value));
+ case DOUBLE -> value -> String.valueOf(BytesUtils.bytesToDouble(value));
+ case TEXT, STRING -> value -> new String(value, StandardCharsets.UTF_8);
+ case OBJECT -> BytesUtils::parseObjectByteArrayToString;
+ case BLOB -> BytesUtils::parseBlobByteArrayToString;
+ case DATE -> value -> DateUtils.formatDate(BytesUtils.bytesToInt(value));
+ case ROW, UNKNOWN, VECTOR -> ignored -> null;
+ };
+
+ static final TypeService> JDBC_OBJECT_READER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN -> BytesUtils::bytesToBool;
+ case INT32 -> BytesUtils::bytesToInt;
+ case INT64 -> BytesUtils::bytesToLong;
+ case FLOAT -> BytesUtils::bytesToFloat;
+ case DOUBLE -> BytesUtils::bytesToDouble;
+ case TEXT, STRING -> value -> new String(value, StandardCharsets.UTF_8);
+ case OBJECT, BLOB -> Binary::new;
+ case TIMESTAMP -> value -> new Timestamp(BytesUtils.bytesToLong(value));
+ case DATE -> value -> DateUtils.parseIntToDate(BytesUtils.bytesToInt(value));
+ case ROW, UNKNOWN, VECTOR -> ignored -> null;
+ };
+
+ static final TypeService 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 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 e42710d23e12..6bf9cbba2265 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 @@
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 static List deserialize(ByteBuffer buffer) {
}
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
new file mode 100644
index 000000000000..ccadd0d51d53
--- /dev/null
+++ b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/IoTDBJDBCDataSetTest.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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 {
+
+ @Test
+ public void testInitializeValueBuffersUsesDataTypeInterfaces() {
+ byte[][] buffers =
+ IoTDBJDBCDataSet.initializeValueBuffers(
+ Arrays.asList(
+ TSDataType.BOOLEAN,
+ TSDataType.INT32,
+ TSDataType.INT64,
+ TSDataType.FLOAT,
+ TSDataType.DOUBLE,
+ TSDataType.DATE,
+ TSDataType.TIMESTAMP,
+ TSDataType.TEXT,
+ TSDataType.STRING,
+ TSDataType.BLOB,
+ TSDataType.OBJECT));
+
+ Assert.assertEquals(Byte.BYTES, buffers[0].length);
+ Assert.assertEquals(Integer.BYTES, buffers[1].length);
+ Assert.assertEquals(Long.BYTES, buffers[2].length);
+ Assert.assertEquals(Float.BYTES, buffers[3].length);
+ Assert.assertEquals(Double.BYTES, buffers[4].length);
+ Assert.assertEquals(Integer.BYTES, buffers[5].length);
+ Assert.assertEquals(Long.BYTES, buffers[6].length);
+ Assert.assertNull(buffers[7]);
+ Assert.assertNull(buffers[8]);
+ Assert.assertNull(buffers[9]);
+ Assert.assertNull(buffers[10]);
+ }
+
+ @Test
+ public void testInitializeValueBuffersRejectsInternalTypes() {
+ assertUnsupported(TSDataType.UNKNOWN);
+ assertUnsupported(TSDataType.VECTOR);
+ }
+
+ @Test
+ public void testJdbcValueReadersUseTypeServices() {
+ Assert.assertEquals(
+ "42",
+ TypeServices.JDBC_STRING_READER_SERVICE
+ .call(Type.fromTsDataType(TSDataType.INT32))
+ .apply(BytesUtils.intToBytes(42)));
+ Assert.assertEquals(
+ 42,
+ TypeServices.JDBC_OBJECT_READER_SERVICE
+ .call(Type.fromTsDataType(TSDataType.INT32))
+ .apply(BytesUtils.intToBytes(42)));
+
+ byte[] binary = new byte[] {1, 2, 3};
+ Assert.assertEquals(
+ new Binary(binary),
+ TypeServices.JDBC_OBJECT_READER_SERVICE
+ .call(Type.fromTsDataType(TSDataType.BLOB))
+ .apply(binary));
+
+ Assert.assertNull(
+ TypeServices.JDBC_STRING_READER_SERVICE.call(UnknownType.UNKNOWN).apply(new byte[0]));
+ Assert.assertNull(
+ 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(
+ UnSupportedDataTypeException.class,
+ () -> IoTDBJDBCDataSet.initializeValueBuffers(Arrays.asList(dataType)));
+ Assert.assertTrue(
+ exception
+ .getMessage()
+ .endsWith(String.format(IoTDBJDBCDataSet.DATA_TYPE_NOT_SUPPORTED, dataType)));
+ }
+}
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 cd3929c8224d..20de227def4b 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.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 void testInvalidParameterCount() {
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));
+ }
}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
index 39bb9c06cba5..0a8983070b77 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -70,10 +70,8 @@
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.utils.Pair;
-import org.apache.tsfile.write.UnSupportedDataTypeException;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -83,12 +81,10 @@
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.time.LocalDate;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -108,6 +104,7 @@
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.IntToLongFunction;
import java.util.stream.Collectors;
@SuppressWarnings({"java:S107", "java:S1135"}) // need enough parameters, ignore todos
@@ -2504,11 +2501,7 @@ private TSInsertRecordsOfOneDeviceReq genTSInsertRecordsOfOneDeviceReq(
if (!checkSorted(times)) {
// sort
- Integer[] index = new Integer[times.size()];
- for (int i = 0; i < times.size(); i++) {
- index[i] = i;
- }
- Arrays.sort(index, Comparator.comparingLong(times::get));
+ int[] index = sortedIndex(times);
times.sort(Long::compareTo);
// sort measurementList
measurementsList = sortList(measurementsList, index);
@@ -2561,11 +2554,7 @@ private TSInsertStringRecordsOfOneDeviceReq genTSInsertStringRecordsOfOneDeviceR
}
if (!checkSorted(times)) {
- Integer[] index = new Integer[times.size()];
- for (int i = 0; i < index.length; i++) {
- index[i] = i;
- }
- Arrays.sort(index, Comparator.comparingLong(times::get));
+ int[] index = sortedIndex(times);
times.sort(Long::compareTo);
// sort measurementsList
measurementsList = sortList(measurementsList, index);
@@ -2592,7 +2581,7 @@ private TSInsertStringRecordsOfOneDeviceReq genTSInsertStringRecordsOfOneDeviceR
* @param Input type
* @return ordered list
*/
- private static List sortList(List source, Integer[] index) {
+ private static List sortList(List source, int[] index) {
List sortedList = new ArrayList<>(index.length);
for (int position : index) {
sortedList.add(source.get(position));
@@ -3760,11 +3749,7 @@ public void sortTablet(Tablet tablet) {
long[] timestamps = tablet.getTimestamps();
Object[] values = tablet.getValues();
BitMap[] bitMaps = tablet.getBitMaps();
- Integer[] index = new Integer[tablet.getRowSize()];
- for (int i = 0; i < tablet.getRowSize(); i++) {
- index[i] = i;
- }
- Arrays.sort(index, Comparator.comparingLong(o -> timestamps[o]));
+ int[] index = sortedIndex(timestamps, tablet.getRowSize());
Arrays.sort(timestamps, 0, tablet.getRowSize());
int columnIndex = 0;
for (int i = 0; i < tablet.getSchemas().size(); i++) {
@@ -3798,64 +3783,8 @@ public void sortTablet(Tablet tablet) {
* @param index index
* @return sorted list
*/
- private Object sortList(Object valueList, TSDataType dataType, Integer[] index) {
- switch (dataType) {
- case BOOLEAN:
- boolean[] boolValues = (boolean[]) valueList;
- boolean[] sortedValues = new boolean[boolValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedValues[i] = boolValues[index[i]];
- }
- return sortedValues;
- case INT32:
- int[] intValues = (int[]) valueList;
- int[] sortedIntValues = new int[intValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedIntValues[i] = intValues[index[i]];
- }
- return sortedIntValues;
- case DATE:
- LocalDate[] date = (LocalDate[]) valueList;
- LocalDate[] sortedDateValues = new LocalDate[date.length];
- for (int i = 0; i < index.length; i++) {
- sortedDateValues[i] = date[index[i]];
- }
- return sortedDateValues;
- case INT64:
- case TIMESTAMP:
- long[] longValues = (long[]) valueList;
- long[] sortedLongValues = new long[longValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedLongValues[i] = longValues[index[i]];
- }
- return sortedLongValues;
- case FLOAT:
- float[] floatValues = (float[]) valueList;
- float[] sortedFloatValues = new float[floatValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedFloatValues[i] = floatValues[index[i]];
- }
- return sortedFloatValues;
- case DOUBLE:
- double[] doubleValues = (double[]) valueList;
- double[] sortedDoubleValues = new double[doubleValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedDoubleValues[i] = doubleValues[index[i]];
- }
- return sortedDoubleValues;
- case TEXT:
- case BLOB:
- case STRING:
- case OBJECT:
- Binary[] binaryValues = (Binary[]) valueList;
- Binary[] sortedBinaryValues = new Binary[binaryValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedBinaryValues[i] = binaryValues[index[i]];
- }
- return sortedBinaryValues;
- default:
- throw new UnSupportedDataTypeException(MSG_UNSUPPORTED_DATA_TYPE + dataType);
- }
+ private Object sortList(Object valueList, TSDataType dataType, int[] index) {
+ return SessionUtils.sortValueList(valueList, dataType, index);
}
/**
@@ -3865,7 +3794,7 @@ private Object sortList(Object valueList, TSDataType dataType, Integer[] index)
* @param index index
* @return sorted bitMap
*/
- private BitMap sortBitMap(BitMap bitMap, Integer[] index) {
+ private BitMap sortBitMap(BitMap bitMap, int[] index) {
BitMap sortedBitMap = new BitMap(bitMap.getSize());
for (int i = 0; i < index.length; i++) {
if (bitMap.isMarked(index[i])) {
@@ -3875,6 +3804,52 @@ private BitMap sortBitMap(BitMap bitMap, Integer[] index) {
return sortedBitMap;
}
+ private static int[] sortedIndex(List values) {
+ return sortedIndex(values.size(), index -> values.get(index));
+ }
+
+ private static int[] sortedIndex(long[] values, int size) {
+ return sortedIndex(size, index -> values[index]);
+ }
+
+ private static int[] sortedIndex(int size, IntToLongFunction valueProvider) {
+ int[] index = new int[size];
+ int[] scratch = new int[size];
+ for (int i = 0; i < size; i++) {
+ index[i] = i;
+ }
+ sortIndexes(index, scratch, 0, size, valueProvider);
+ return index;
+ }
+
+ private static void sortIndexes(
+ int[] index, int[] scratch, int from, int to, IntToLongFunction valueProvider) {
+ if (to - from < 2) {
+ return;
+ }
+ int middle = (from + to) >>> 1;
+ sortIndexes(index, scratch, from, middle, valueProvider);
+ sortIndexes(index, scratch, middle, to, valueProvider);
+
+ int left = from;
+ int right = middle;
+ int destination = from;
+ while (left < middle && right < to) {
+ if (valueProvider.applyAsLong(index[left]) <= valueProvider.applyAsLong(index[right])) {
+ scratch[destination++] = index[left++];
+ } else {
+ scratch[destination++] = index[right++];
+ }
+ }
+ while (left < middle) {
+ scratch[destination++] = index[left++];
+ }
+ while (right < to) {
+ scratch[destination++] = index[right++];
+ }
+ System.arraycopy(scratch, from, index, from, to - from);
+ }
+
@Override
public void setSchemaTemplate(String templateName, String prefixPath)
throws IoTDBConnectionException, StatementExecutionException {
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionTypeServices.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionTypeServices.java
new file mode 100644
index 000000000000..24119c3ca464
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionTypeServices.java
@@ -0,0 +1,435 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iotdb.session.util;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.encoding.encoder.Encoder;
+import org.apache.tsfile.enums.TSDataType;
+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;
+import org.apache.tsfile.utils.DateUtils;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.apache.tsfile.write.UnSupportedDataTypeException;
+import org.apache.tsfile.write.record.Tablet;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.time.LocalDate;
+
+/** Type-specific operations used by the Session record-value wire format. */
+final class SessionTypeServices {
+
+ private static final int EMPTY_DATE_INT = 10000101;
+
+ private static final TypeService VALUE_LENGTH_CALCULATOR_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN -> value -> Byte.BYTES;
+ case INT32, DATE -> value -> Integer.BYTES;
+ case INT64, TIMESTAMP -> value -> Long.BYTES;
+ case FLOAT -> value -> Float.BYTES;
+ case DOUBLE -> value -> Double.BYTES;
+ case TEXT, STRING, OBJECT -> value -> Integer.BYTES + getTextBytes(value).length;
+ case BLOB -> value -> Integer.BYTES + ((Binary) value).getValues().length;
+ case ROW, UNKNOWN, VECTOR ->
+ value -> {
+ throw unsupportedDataType(type.getTypeEnum());
+ };
+ };
+
+ private static final TypeService VALUE_WRITER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN -> (value, buffer) -> ReadWriteIOUtils.write((Boolean) value, buffer);
+ case INT32 -> (value, buffer) -> ReadWriteIOUtils.write((Integer) value, buffer);
+ case DATE ->
+ (value, buffer) ->
+ ReadWriteIOUtils.write(
+ DateUtils.parseDateExpressionToInt((LocalDate) value), buffer);
+ case INT64, TIMESTAMP ->
+ (value, buffer) -> ReadWriteIOUtils.write((Long) value, buffer);
+ case FLOAT -> (value, buffer) -> ReadWriteIOUtils.write((Float) value, buffer);
+ case DOUBLE -> (value, buffer) -> ReadWriteIOUtils.write((Double) value, buffer);
+ case TEXT, STRING ->
+ (value, buffer) -> {
+ byte[] bytes = getTextBytes(value);
+ ReadWriteIOUtils.write(bytes.length, buffer);
+ buffer.put(bytes);
+ };
+ case BLOB ->
+ (value, buffer) -> {
+ byte[] bytes = ((Binary) value).getValues();
+ ReadWriteIOUtils.write(bytes.length, buffer);
+ buffer.put(bytes);
+ };
+ // OBJECT was accepted by length calculation historically, but not by value writing.
+ case OBJECT, ROW, UNKNOWN, VECTOR ->
+ (value, buffer) -> {
+ throw unsupportedDataType(type.getTypeEnum());
+ };
+ };
+
+ private static final TypeService
+ TABLET_COLUMN_OCCUPATION_CALCULATOR_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN -> (values, columnIndex, rowSize) -> rowSize;
+ case INT32, FLOAT, DATE ->
+ (values, columnIndex, rowSize) -> rowSize * Integer.BYTES;
+ case INT64, DOUBLE, TIMESTAMP ->
+ (values, columnIndex, rowSize) -> rowSize * Long.BYTES;
+ case TEXT, BLOB, STRING, OBJECT ->
+ (values, columnIndex, rowSize) -> {
+ int occupation = rowSize * Integer.BYTES;
+ Binary[] binaries = (Binary[]) values[columnIndex];
+ for (int rowIndex = 0; rowIndex < rowSize; rowIndex++) {
+ occupation +=
+ binaries[rowIndex] != null
+ ? binaries[rowIndex].getLength()
+ : Binary.EMPTY_VALUE.getLength();
+ }
+ return occupation;
+ };
+ case ROW, UNKNOWN, VECTOR ->
+ (values, columnIndex, rowSize) -> {
+ throw unsupportedTabletDataType(type.getTypeEnum());
+ };
+ };
+
+ private static final TypeService TABLET_VALUE_WRITER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case INT32 ->
+ (tablet, columnIndex, valueBuffer) -> {
+ int[] values = (int[]) tablet.getValues()[columnIndex];
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ valueBuffer.putInt(
+ tablet.isNull(index, columnIndex) ? Integer.MIN_VALUE : values[index]);
+ }
+ };
+ case INT64, TIMESTAMP ->
+ (tablet, columnIndex, valueBuffer) -> {
+ long[] values = (long[]) tablet.getValues()[columnIndex];
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ valueBuffer.putLong(
+ tablet.isNull(index, columnIndex) ? Long.MIN_VALUE : values[index]);
+ }
+ };
+ case FLOAT ->
+ (tablet, columnIndex, valueBuffer) -> {
+ float[] values = (float[]) tablet.getValues()[columnIndex];
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ valueBuffer.putFloat(
+ tablet.isNull(index, columnIndex) ? Float.MIN_VALUE : values[index]);
+ }
+ };
+ case DOUBLE ->
+ (tablet, columnIndex, valueBuffer) -> {
+ double[] values = (double[]) tablet.getValues()[columnIndex];
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ valueBuffer.putDouble(
+ tablet.isNull(index, columnIndex) ? Double.MIN_VALUE : values[index]);
+ }
+ };
+ case BOOLEAN ->
+ (tablet, columnIndex, valueBuffer) -> {
+ boolean[] values = (boolean[]) tablet.getValues()[columnIndex];
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ valueBuffer.put(
+ BytesUtils.boolToByte(!tablet.isNull(index, columnIndex) && values[index]));
+ }
+ };
+ case TEXT, STRING, BLOB, OBJECT ->
+ (tablet, columnIndex, valueBuffer) -> {
+ Binary[] values = (Binary[]) tablet.getValues()[columnIndex];
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ Binary value =
+ !tablet.isNull(index, columnIndex) && values[index] != null
+ ? values[index]
+ : Binary.EMPTY_VALUE;
+ valueBuffer.putInt(value.getLength());
+ valueBuffer.put(value.getValues());
+ }
+ };
+ case DATE ->
+ (tablet, columnIndex, valueBuffer) -> {
+ LocalDate[] values = (LocalDate[]) tablet.getValues()[columnIndex];
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ valueBuffer.putInt(
+ !tablet.isNull(index, columnIndex) && values[index] != null
+ ? DateUtils.parseDateExpressionToInt(values[index])
+ : EMPTY_DATE_INT);
+ }
+ };
+ case ROW, UNKNOWN, VECTOR ->
+ (tablet, columnIndex, valueBuffer) -> {
+ throw unsupportedTabletDataType(type.getTypeEnum());
+ };
+ };
+
+ private static final TypeService TABLET_VALUE_ENCODER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case INT32 ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ int[] values = (int[]) tablet.getValues()[columnIndex];
+ int lastNonNullValue = 0;
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ if (!tablet.isNull(index, columnIndex)) {
+ lastNonNullValue = values[index];
+ }
+ encoder.encode(lastNonNullValue, outputStream);
+ }
+ };
+ case INT64, TIMESTAMP ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ long[] values = (long[]) tablet.getValues()[columnIndex];
+ long lastNonNullValue = 0;
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ if (!tablet.isNull(index, columnIndex)) {
+ lastNonNullValue = values[index];
+ }
+ encoder.encode(lastNonNullValue, outputStream);
+ }
+ };
+ case FLOAT ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ float[] values = (float[]) tablet.getValues()[columnIndex];
+ float lastNonNullValue = 0.0f;
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ if (!tablet.isNull(index, columnIndex)) {
+ lastNonNullValue = values[index];
+ }
+ encoder.encode(lastNonNullValue, outputStream);
+ }
+ };
+ case DOUBLE ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ double[] values = (double[]) tablet.getValues()[columnIndex];
+ double lastNonNullValue = 0.0;
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ if (!tablet.isNull(index, columnIndex)) {
+ lastNonNullValue = values[index];
+ }
+ encoder.encode(lastNonNullValue, outputStream);
+ }
+ };
+ case BOOLEAN ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ boolean[] values = (boolean[]) tablet.getValues()[columnIndex];
+ boolean lastNonNullValue = false;
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ if (!tablet.isNull(index, columnIndex)) {
+ lastNonNullValue = values[index];
+ }
+ encoder.encode(lastNonNullValue, outputStream);
+ }
+ };
+ case TEXT, STRING, BLOB ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ Binary[] values = (Binary[]) tablet.getValues()[columnIndex];
+ Binary lastNonNullValue = Binary.EMPTY_VALUE;
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ if (!tablet.isNull(index, columnIndex) && values[index] != null) {
+ lastNonNullValue = values[index];
+ }
+ encoder.encode(lastNonNullValue, outputStream);
+ }
+ };
+ case DATE ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ LocalDate[] values = (LocalDate[]) tablet.getValues()[columnIndex];
+ int lastNonNullValue = EMPTY_DATE_INT;
+ for (int index = 0; index < tablet.getRowSize(); index++) {
+ if (!tablet.isNull(index, columnIndex)) {
+ lastNonNullValue = DateUtils.parseDateExpressionToInt(values[index]);
+ }
+ // Previous values make null runs more compressible without changing the bitmap.
+ encoder.encode(lastNonNullValue, outputStream);
+ }
+ };
+ case OBJECT, ROW, UNKNOWN, VECTOR ->
+ (tablet, columnIndex, encoder, outputStream) -> {
+ throw unsupportedTabletDataType(type.getTypeEnum());
+ };
+ };
+
+ private static final TypeService VALUE_LIST_SORTER_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case BOOLEAN ->
+ (valueList, index) -> {
+ boolean[] values = (boolean[]) valueList;
+ boolean[] sortedValues = new boolean[values.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[i] = values[index[i]];
+ }
+ return sortedValues;
+ };
+ case INT32 ->
+ (valueList, index) -> {
+ int[] values = (int[]) valueList;
+ int[] sortedValues = new int[values.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[i] = values[index[i]];
+ }
+ return sortedValues;
+ };
+ case DATE ->
+ (valueList, index) -> {
+ LocalDate[] values = (LocalDate[]) valueList;
+ LocalDate[] sortedValues = new LocalDate[values.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[i] = values[index[i]];
+ }
+ return sortedValues;
+ };
+ case INT64, TIMESTAMP ->
+ (valueList, index) -> {
+ long[] values = (long[]) valueList;
+ long[] sortedValues = new long[values.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[i] = values[index[i]];
+ }
+ return sortedValues;
+ };
+ case FLOAT ->
+ (valueList, index) -> {
+ float[] values = (float[]) valueList;
+ float[] sortedValues = new float[values.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[i] = values[index[i]];
+ }
+ return sortedValues;
+ };
+ case DOUBLE ->
+ (valueList, index) -> {
+ double[] values = (double[]) valueList;
+ double[] sortedValues = new double[values.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[i] = values[index[i]];
+ }
+ return sortedValues;
+ };
+ case TEXT, BLOB, STRING, OBJECT ->
+ (valueList, index) -> {
+ Binary[] values = (Binary[]) valueList;
+ Binary[] sortedValues = new Binary[values.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[i] = values[index[i]];
+ }
+ return sortedValues;
+ };
+ case ROW, UNKNOWN, VECTOR ->
+ (valueList, index) -> {
+ throw unsupportedValueListDataType(type.getTypeEnum());
+ };
+ };
+
+ static {
+ VALUE_LENGTH_CALCULATOR_SERVICE.check();
+ VALUE_WRITER_SERVICE.check();
+ TABLET_COLUMN_OCCUPATION_CALCULATOR_SERVICE.check();
+ TABLET_VALUE_WRITER_SERVICE.check();
+ TABLET_VALUE_ENCODER_SERVICE.check();
+ VALUE_LIST_SORTER_SERVICE.check();
+ }
+
+ private SessionTypeServices() {}
+
+ static ValueLengthCalculator valueLengthCalculator(TSDataType dataType) {
+ return VALUE_LENGTH_CALCULATOR_SERVICE.call(Type.fromTsDataType(dataType));
+ }
+
+ static ValueWriter valueWriter(TSDataType dataType) {
+ return VALUE_WRITER_SERVICE.call(Type.fromTsDataType(dataType));
+ }
+
+ static TabletColumnOccupationCalculator tabletColumnOccupationCalculator(TSDataType dataType) {
+ return TABLET_COLUMN_OCCUPATION_CALCULATOR_SERVICE.call(Type.fromTsDataType(dataType));
+ }
+
+ static TabletValueWriter tabletValueWriter(TSDataType dataType) {
+ return TABLET_VALUE_WRITER_SERVICE.call(Type.fromTsDataType(dataType));
+ }
+
+ static TabletValueEncoder tabletValueEncoder(TSDataType dataType) {
+ return TABLET_VALUE_ENCODER_SERVICE.call(Type.fromTsDataType(dataType));
+ }
+
+ static ValueListSorter valueListSorter(TSDataType dataType) {
+ return VALUE_LIST_SORTER_SERVICE.call(Type.fromTsDataType(dataType));
+ }
+
+ private static byte[] getTextBytes(Object value) {
+ if (value instanceof Binary binary) {
+ return binary.getValues();
+ }
+ return ((String) value).getBytes(TSFileConfig.STRING_CHARSET);
+ }
+
+ private static IoTDBConnectionException unsupportedDataType(Object dataType) {
+ return new IoTDBConnectionException(Session.MSG_UNSUPPORTED_DATA_TYPE + dataType);
+ }
+
+ private static UnSupportedDataTypeException unsupportedTabletDataType(Object dataType) {
+ return new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.", dataType));
+ }
+
+ private static UnSupportedDataTypeException unsupportedValueListDataType(Object dataType) {
+ return new UnSupportedDataTypeException(Session.MSG_UNSUPPORTED_DATA_TYPE + dataType);
+ }
+
+ @FunctionalInterface
+ interface ValueLengthCalculator {
+ int calculate(Object value) throws IoTDBConnectionException;
+ }
+
+ @FunctionalInterface
+ interface ValueWriter {
+ void write(Object value, ByteBuffer buffer) throws IoTDBConnectionException;
+ }
+
+ @FunctionalInterface
+ interface TabletColumnOccupationCalculator {
+ int calculate(Object[] values, int columnIndex, int rowSize);
+ }
+
+ @FunctionalInterface
+ interface TabletValueWriter {
+ void write(Tablet tablet, int columnIndex, ByteBuffer valueBuffer);
+ }
+
+ @FunctionalInterface
+ interface TabletValueEncoder {
+ void encode(
+ Tablet tablet, int columnIndex, Encoder encoder, ByteArrayOutputStream outputStream);
+ }
+
+ @FunctionalInterface
+ interface ValueListSorter {
+ Object sort(Object valueList, int[] index);
+ }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
index b3b29f0be932..d6ff4d735ae5 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
@@ -24,17 +24,13 @@
import org.apache.iotdb.rpc.UrlUtils;
import org.apache.iotdb.session.i18n.SessionMessages;
-import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.encoding.encoder.Encoder;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
-import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.utils.BytesUtils;
-import org.apache.tsfile.utils.DateUtils;
import org.apache.tsfile.utils.ReadWriteIOUtils;
-import org.apache.tsfile.write.UnSupportedDataTypeException;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.slf4j.Logger;
@@ -43,17 +39,13 @@
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
-import static org.apache.iotdb.session.Session.MSG_UNSUPPORTED_DATA_TYPE;
-
public class SessionUtils {
private static final Logger LOGGER = LoggerFactory.getLogger(SessionUtils.class);
private static final byte TYPE_NULL = -2;
- private static final int EMPTY_DATE_INT = 10000101;
public static ByteBuffer getTimeBuffer(Tablet tablet) {
ByteBuffer timeBuffer = ByteBuffer.allocate(getTimeBytesSize(tablet));
@@ -119,40 +111,8 @@ private static int getTotalValueOccupation(Tablet tablet) {
private static int calOccupationOfOneColumn(
TSDataType dataType, Object[] values, int columnIndex, int rowSize) {
- int valueOccupation = 0;
- switch (dataType) {
- case BOOLEAN:
- valueOccupation += rowSize;
- break;
- case INT32:
- case FLOAT:
- case DATE:
- valueOccupation += rowSize * 4;
- break;
- case INT64:
- case DOUBLE:
- case TIMESTAMP:
- valueOccupation += rowSize * 8;
- break;
- case TEXT:
- case BLOB:
- case STRING:
- case OBJECT:
- valueOccupation += rowSize * 4;
- Binary[] binaries = (Binary[]) values[columnIndex];
- for (int rowIndex = 0; rowIndex < rowSize; rowIndex++) {
- valueOccupation +=
- binaries[rowIndex] != null
- ? binaries[rowIndex].getLength()
- : Binary.EMPTY_VALUE.getLength();
- }
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format(
- SessionMessages.EXCEPTION_DATA_TYPE_ARG_NOT_SUPPORTED_31213160, dataType));
- }
- return valueOccupation;
+ return SessionTypeServices.tabletColumnOccupationCalculator(dataType)
+ .calculate(values, columnIndex, rowSize);
}
public static ByteBuffer getValueBuffer(
@@ -163,47 +123,17 @@ public static ByteBuffer getValueBuffer(
return buffer;
}
+ public static Object sortValueList(Object valueList, TSDataType dataType, int[] index) {
+ return SessionTypeServices.valueListSorter(dataType).sort(valueList, index);
+ }
+
public static int calculateLength(List types, List extends Object> values)
throws IoTDBConnectionException {
int res = 0;
for (int i = 0; i < types.size(); i++) {
// types
res += Byte.BYTES;
- switch (types.get(i)) {
- case BOOLEAN:
- res += 1;
- break;
- case INT32:
- case DATE:
- res += Integer.BYTES;
- break;
- case INT64:
- case TIMESTAMP:
- res += Long.BYTES;
- break;
- case FLOAT:
- res += Float.BYTES;
- break;
- case DOUBLE:
- res += Double.BYTES;
- break;
- case TEXT:
- case STRING:
- case OBJECT:
- res += Integer.BYTES;
- if (values.get(i) instanceof Binary) {
- res += ((Binary) values.get(i)).getValues().length;
- } else {
- res += ((String) values.get(i)).getBytes(TSFileConfig.STRING_CHARSET).length;
- }
- break;
- case BLOB:
- res += Integer.BYTES;
- res += ((Binary) values.get(i)).getValues().length;
- break;
- default:
- throw new IoTDBConnectionException(MSG_UNSUPPORTED_DATA_TYPE + types.get(i));
- }
+ res += SessionTypeServices.valueLengthCalculator(types.get(i)).calculate(values.get(i));
}
return res;
}
@@ -228,47 +158,9 @@ public static void putValues(
ReadWriteIOUtils.write(TYPE_NULL, buffer);
continue;
}
- ReadWriteIOUtils.write(types.get(i), buffer);
- switch (types.get(i)) {
- case BOOLEAN:
- ReadWriteIOUtils.write((Boolean) values.get(i), buffer);
- break;
- case INT32:
- ReadWriteIOUtils.write((Integer) values.get(i), buffer);
- break;
- case DATE:
- ReadWriteIOUtils.write(
- DateUtils.parseDateExpressionToInt((LocalDate) values.get(i)), buffer);
- break;
- case INT64:
- case TIMESTAMP:
- ReadWriteIOUtils.write((Long) values.get(i), buffer);
- break;
- case FLOAT:
- ReadWriteIOUtils.write((Float) values.get(i), buffer);
- break;
- case DOUBLE:
- ReadWriteIOUtils.write((Double) values.get(i), buffer);
- break;
- case TEXT:
- case STRING:
- byte[] bytes;
- if (values.get(i) instanceof Binary) {
- bytes = ((Binary) values.get(i)).getValues();
- } else {
- bytes = ((String) values.get(i)).getBytes(TSFileConfig.STRING_CHARSET);
- }
- ReadWriteIOUtils.write(bytes.length, buffer);
- buffer.put(bytes);
- break;
- case BLOB:
- bytes = ((Binary) values.get(i)).getValues();
- ReadWriteIOUtils.write(bytes.length, buffer);
- buffer.put(bytes);
- break;
- default:
- throw new IoTDBConnectionException(MSG_UNSUPPORTED_DATA_TYPE + types.get(i));
- }
+ TSDataType type = types.get(i);
+ ReadWriteIOUtils.write(type, buffer);
+ SessionTypeServices.valueWriter(type).write(values.get(i), buffer);
} catch (Throwable e) {
LOGGER.error(
SessionMessages.LOG_CANNOT_PUT_VALUES_MEASUREMENT_ARG_TYPE_ARG_27AFC67B,
@@ -281,99 +173,11 @@ public static void putValues(
buffer.flip();
}
- @SuppressWarnings({
- "squid:S6541",
- "squid:S3776"
- }) /// ignore Cognitive Complexity of methods should not be too high
- // ignore Methods should not perform too many tasks (aka Brain method)
private static void getValueBufferOfDataType(
TSDataType dataType, Tablet tablet, int i, ByteBuffer valueBuffer) {
-
- switch (dataType) {
- case INT32:
- int[] intValues = (int[]) tablet.getValues()[i];
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- valueBuffer.putInt(intValues[index]);
- } else {
- valueBuffer.putInt(Integer.MIN_VALUE);
- }
- }
- break;
- case INT64:
- case TIMESTAMP:
- long[] longValues = (long[]) tablet.getValues()[i];
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- valueBuffer.putLong(longValues[index]);
- } else {
- valueBuffer.putLong(Long.MIN_VALUE);
- }
- }
- break;
- case FLOAT:
- float[] floatValues = (float[]) tablet.getValues()[i];
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- valueBuffer.putFloat(floatValues[index]);
- } else {
- valueBuffer.putFloat(Float.MIN_VALUE);
- }
- }
- break;
- case DOUBLE:
- double[] doubleValues = (double[]) tablet.getValues()[i];
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- valueBuffer.putDouble(doubleValues[index]);
- } else {
- valueBuffer.putDouble(Double.MIN_VALUE);
- }
- }
- break;
- case BOOLEAN:
- boolean[] boolValues = (boolean[]) tablet.getValues()[i];
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- valueBuffer.put(BytesUtils.boolToByte(boolValues[index]));
- } else {
- valueBuffer.put(BytesUtils.boolToByte(false));
- }
- }
- break;
- case TEXT:
- case STRING:
- case BLOB:
- case OBJECT:
- Binary[] binaryValues = (Binary[]) tablet.getValues()[i];
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i) && binaryValues[index] != null) {
- valueBuffer.putInt(binaryValues[index].getLength());
- valueBuffer.put(binaryValues[index].getValues());
- } else {
- valueBuffer.putInt(Binary.EMPTY_VALUE.getLength());
- valueBuffer.put(Binary.EMPTY_VALUE.getValues());
- }
- }
- break;
- case DATE:
- LocalDate[] dateValues = (LocalDate[]) tablet.getValues()[i];
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i) && dateValues[index] != null) {
- valueBuffer.putInt(DateUtils.parseDateExpressionToInt(dateValues[index]));
- } else {
- valueBuffer.putInt(EMPTY_DATE_INT);
- }
- }
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format(
- SessionMessages.EXCEPTION_DATA_TYPE_ARG_NOT_SUPPORTED_31213160, dataType));
- }
+ SessionTypeServices.tabletValueWriter(dataType).write(tablet, i, valueBuffer);
}
- @SuppressWarnings({"java:S3776", "java:S6541"})
public static void encodeValue(
TSDataType dataType,
Tablet tablet,
@@ -381,86 +185,7 @@ public static void encodeValue(
Encoder encoder,
ByteArrayOutputStream outputStream) {
- switch (dataType) {
- case INT32:
- int[] intValues = (int[]) tablet.getValues()[i];
- int lastNonNullIntValue = 0;
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- lastNonNullIntValue = intValues[index];
- }
- encoder.encode(lastNonNullIntValue, outputStream);
- }
- break;
- case INT64:
- case TIMESTAMP:
- long[] longValues = (long[]) tablet.getValues()[i];
- long lastNonNullLongValue = 0;
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- lastNonNullLongValue = longValues[index];
- }
- encoder.encode(lastNonNullLongValue, outputStream);
- }
- break;
- case FLOAT:
- float[] floatValues = (float[]) tablet.getValues()[i];
- float lastNonNullFloatValue = 0.0f;
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- lastNonNullFloatValue = floatValues[index];
- }
- encoder.encode(lastNonNullFloatValue, outputStream);
- }
- break;
- case DOUBLE:
- double[] doubleValues = (double[]) tablet.getValues()[i];
- double lastNonNullDoubleValue = 0.0;
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- lastNonNullDoubleValue = doubleValues[index];
- }
- encoder.encode(lastNonNullDoubleValue, outputStream);
- }
- break;
- case BOOLEAN:
- boolean[] boolValues = (boolean[]) tablet.getValues()[i];
- boolean lastNonNullBooleanValue = false;
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- lastNonNullBooleanValue = boolValues[index];
- }
- encoder.encode(lastNonNullBooleanValue, outputStream);
- }
- break;
- case TEXT:
- case STRING:
- case BLOB:
- Binary[] binaryValues = (Binary[]) tablet.getValues()[i];
- Binary lastNonNullBinaryValue = Binary.EMPTY_VALUE;
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i) && binaryValues[index] != null) {
- lastNonNullBinaryValue = binaryValues[index];
- }
- encoder.encode(lastNonNullBinaryValue, outputStream);
- }
- break;
- case DATE:
- LocalDate[] dateValues = (LocalDate[]) tablet.getValues()[i];
- int lastNonNullDateValue = EMPTY_DATE_INT;
- for (int index = 0; index < tablet.getRowSize(); index++) {
- if (!tablet.isNull(index, i)) {
- lastNonNullDateValue = DateUtils.parseDateExpressionToInt(dateValues[index]);
- }
- // use the previous value as the placeholder of nulls to increase encoding performance
- encoder.encode(lastNonNullDateValue, outputStream);
- }
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format(
- SessionMessages.EXCEPTION_DATA_TYPE_ARG_NOT_SUPPORTED_31213160, dataType));
- }
+ SessionTypeServices.tabletValueEncoder(dataType).encode(tablet, i, encoder, outputStream);
try {
encoder.flush(outputStream);
} catch (IOException e) {
diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java
index b8524d4b3112..88b102642620 100644
--- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java
+++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java
@@ -147,6 +147,27 @@ public void testGetValueBuffer2() throws IoTDBConnectionException {
}
}
+ // Covers DATE/TIMESTAMP conversion and STRING/BLOB payloads, including mixed fixed/variable
+ // widths; the encoded buffer must account for each type marker, length prefix, and payload.
+ @Test
+ public void testRecordValueTypeServices() throws IoTDBConnectionException {
+ List types =
+ Arrays.asList(TSDataType.DATE, TSDataType.TIMESTAMP, TSDataType.STRING, TSDataType.BLOB);
+ List