This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch remove_swtich_type in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 286f049c4339f2012c94fba3207cb5558f77e627 Author: Tian Jiang <[email protected]> AuthorDate: Thu Aug 27 17:00:27 2026 +0800 multiple refactors --- .../org/apache/iotdb/isession/SessionDataSet.java | 38 +------ .../org/apache/iotdb/isession/TypeServices.java | 65 ++++++++++++ .../payload/SubscriptionRecordHandler.java | 74 ++----------- .../session/subscription/payload/TypeServices.java | 115 +++++++++++++++++++++ .../subscription/payload/TypeServicesTest.java | 69 +++++++++++++ 5 files changed, 260 insertions(+), 101 deletions(-) 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 46881ce5893..5076afd52d0 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.thrift.TException; 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,39 +161,9 @@ public class SessionDataSet implements ISessionDataSet { 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("Data type %s is not supported.", 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 00000000000..3a74ddbd7b4 --- /dev/null +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/TypeServices.java @@ -0,0 +1,65 @@ +/* + * 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.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<FieldValueReader> 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("Data type %s is not supported.", 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/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java index 69bd24d7687..bd51bfa87d6 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/SubscriptionRecordHandler.java @@ -26,18 +26,15 @@ import org.apache.thrift.annotation.Nullable; 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.query.dataset.AbstractResultSet; import org.apache.tsfile.read.query.dataset.ResultSet; -import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; -import org.apache.tsfile.utils.DateUtils; -import org.apache.tsfile.write.UnSupportedDataTypeException; import org.apache.tsfile.write.record.TSRecord; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; import java.io.IOException; -import java.time.LocalDate; import java.util.ArrayList; import java.util.Collections; import java.util.Iterator; @@ -343,39 +340,9 @@ public class SubscriptionRecordHandler implements Iterable<ResultSet>, Subscript final String measurement = currentTablet.getSchemas().get(columnIndex).getMeasurementName(); final TSDataType dataType = currentTablet.getSchemas().get(columnIndex).getType(); final Object value = currentTablet.getValues()[columnIndex]; - switch (dataType) { - case BOOLEAN: - record.addPoint(measurement, ((boolean[]) value)[currentRowIndex]); - break; - case INT32: - record.addPoint(measurement, ((int[]) value)[currentRowIndex]); - break; - case DATE: - record.addPoint(measurement, ((LocalDate[]) value)[currentRowIndex]); - break; - case INT64: - case TIMESTAMP: - record.addPoint(measurement, ((long[]) value)[currentRowIndex]); - break; - case FLOAT: - record.addPoint(measurement, ((float[]) value)[currentRowIndex]); - break; - case DOUBLE: - record.addPoint(measurement, ((double[]) value)[currentRowIndex]); - break; - case TEXT: - case STRING: - case BLOB: - case OBJECT: - final Binary binary = ((Binary[]) value)[currentRowIndex]; - if (Objects.nonNull(binary)) { - record.addPoint(measurement, binary.getValues()); - } - break; - default: - throw new UnSupportedDataTypeException( - String.format("Data type %s is not supported.", dataType)); - } + TypeServices.TS_RECORD_VALUE_APPENDER_SERVICE + .call(Type.fromTsDataType(dataType)) + .append(record, measurement, value, currentRowIndex); } return record; } @@ -383,36 +350,9 @@ public class SubscriptionRecordHandler implements Iterable<ResultSet>, Subscript private static Field generateFieldFromTabletValue( final TSDataType dataType, final Object value, final int index) { final Field field = new Field(dataType); - switch (dataType) { - case BOOLEAN: - field.setBoolV(((boolean[]) value)[index]); - break; - case INT32: - field.setIntV(((int[]) value)[index]); - break; - case DATE: - field.setIntV(DateUtils.parseDateExpressionToInt(((LocalDate[]) value)[index])); - break; - case INT64: - case TIMESTAMP: - field.setLongV(((long[]) value)[index]); - break; - case FLOAT: - field.setFloatV(((float[]) value)[index]); - break; - case DOUBLE: - field.setDoubleV(((double[]) value)[index]); - break; - case TEXT: - case STRING: - case BLOB: - case OBJECT: - field.setBinaryV(new Binary((((Binary[]) value)[index]).getValues())); - break; - default: - throw new UnSupportedDataTypeException( - String.format("Data type %s is not supported.", dataType)); - } + TypeServices.FIELD_VALUE_READER_SERVICE + .call(Type.fromTsDataType(dataType)) + .read(field, value, index); return field; } diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/TypeServices.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/TypeServices.java new file mode 100644 index 00000000000..813d500ab70 --- /dev/null +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/payload/TypeServices.java @@ -0,0 +1,115 @@ +/* + * 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.subscription.payload; + +import org.apache.tsfile.read.common.Field; +import org.apache.tsfile.read.common.type.service.TypeService; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.DateUtils; +import org.apache.tsfile.write.UnSupportedDataTypeException; +import org.apache.tsfile.write.record.TSRecord; + +import java.time.LocalDate; +import java.util.Objects; + +/** + * Type-specific Tablet readers that preserve primitive-array access without intermediate boxing. + */ +final class TypeServices { + + static final TypeService<TSRecordValueAppender> TS_RECORD_VALUE_APPENDER_SERVICE = + type -> + switch (type.getTypeEnum()) { + case BOOLEAN -> + (record, measurement, values, rowIndex) -> + record.addPoint(measurement, ((boolean[]) values)[rowIndex]); + case INT32 -> + (record, measurement, values, rowIndex) -> + record.addPoint(measurement, ((int[]) values)[rowIndex]); + case DATE -> + (record, measurement, values, rowIndex) -> + record.addPoint(measurement, ((LocalDate[]) values)[rowIndex]); + case INT64, TIMESTAMP -> + (record, measurement, values, rowIndex) -> + record.addPoint(measurement, ((long[]) values)[rowIndex]); + case FLOAT -> + (record, measurement, values, rowIndex) -> + record.addPoint(measurement, ((float[]) values)[rowIndex]); + case DOUBLE -> + (record, measurement, values, rowIndex) -> + record.addPoint(measurement, ((double[]) values)[rowIndex]); + case TEXT, STRING, BLOB, OBJECT -> + (record, measurement, values, rowIndex) -> { + Binary binary = ((Binary[]) values)[rowIndex]; + if (Objects.nonNull(binary)) { + record.addPoint(measurement, binary.getValues()); + } + }; + case ROW, UNKNOWN, VECTOR -> + (record, measurement, values, rowIndex) -> { + throw unsupportedDataType(type.getTypeEnum()); + }; + }; + + static final TypeService<FieldValueReader> FIELD_VALUE_READER_SERVICE = + type -> + switch (type.getTypeEnum()) { + case BOOLEAN -> (field, values, index) -> field.setBoolV(((boolean[]) values)[index]); + case INT32 -> (field, values, index) -> field.setIntV(((int[]) values)[index]); + case DATE -> + (field, values, index) -> + field.setIntV( + DateUtils.parseDateExpressionToInt(((LocalDate[]) values)[index])); + case INT64, TIMESTAMP -> + (field, values, index) -> field.setLongV(((long[]) values)[index]); + case FLOAT -> (field, values, index) -> field.setFloatV(((float[]) values)[index]); + case DOUBLE -> (field, values, index) -> field.setDoubleV(((double[]) values)[index]); + case TEXT, STRING, BLOB, OBJECT -> + (field, values, index) -> + field.setBinaryV(new Binary(((Binary[]) values)[index].getValues())); + case ROW, UNKNOWN, VECTOR -> + (field, values, index) -> { + throw unsupportedDataType(type.getTypeEnum()); + }; + }; + + static { + TS_RECORD_VALUE_APPENDER_SERVICE.check(); + FIELD_VALUE_READER_SERVICE.check(); + } + + private TypeServices() {} + + private static UnSupportedDataTypeException unsupportedDataType(Object dataType) { + return new UnSupportedDataTypeException( + String.format("Data type %s is not supported.", dataType)); + } + + @FunctionalInterface + interface TSRecordValueAppender { + + void append(TSRecord record, String measurement, Object values, int rowIndex); + } + + @FunctionalInterface + interface FieldValueReader { + + void read(Field field, Object values, int index); + } +} diff --git a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/payload/TypeServicesTest.java b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/payload/TypeServicesTest.java new file mode 100644 index 00000000000..718e4b65339 --- /dev/null +++ b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/payload/TypeServicesTest.java @@ -0,0 +1,69 @@ +/* + * 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.subscription.payload; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.Field; +import org.apache.tsfile.read.common.type.Type; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.DateUtils; +import org.apache.tsfile.write.record.TSRecord; +import org.junit.Assert; +import org.junit.Test; + +import java.time.LocalDate; + +public class TypeServicesTest { + + @Test + public void testAppendTabletValuesToTsRecord() { + TSRecord record = new TSRecord("root.sg.d", 1); + + TypeServices.TS_RECORD_VALUE_APPENDER_SERVICE + .call(Type.fromTsDataType(TSDataType.INT32)) + .append(record, "int", new int[] {42}, 0); + TypeServices.TS_RECORD_VALUE_APPENDER_SERVICE + .call(Type.fromTsDataType(TSDataType.DATE)) + .append(record, "date", new LocalDate[] {LocalDate.of(2024, 8, 1)}, 0); + TypeServices.TS_RECORD_VALUE_APPENDER_SERVICE + .call(Type.fromTsDataType(TSDataType.BLOB)) + .append(record, "nullBlob", new Binary[] {null}, 0); + + Assert.assertEquals(2, record.dataPointList.size()); + Assert.assertEquals(42, record.dataPointList.get(0).getValue()); + Assert.assertEquals(20240801, record.dataPointList.get(1).getValue()); + } + + @Test + public void testReadTabletValuesIntoField() { + Field dateField = new Field(TSDataType.DATE); + TypeServices.FIELD_VALUE_READER_SERVICE + .call(Type.fromTsDataType(TSDataType.DATE)) + .read(dateField, new LocalDate[] {LocalDate.of(2024, 8, 1)}, 0); + Assert.assertEquals( + DateUtils.parseDateExpressionToInt(LocalDate.of(2024, 8, 1)).intValue(), + dateField.getIntV()); + + Field blobField = new Field(TSDataType.BLOB); + TypeServices.FIELD_VALUE_READER_SERVICE + .call(Type.fromTsDataType(TSDataType.BLOB)) + .read(blobField, new Binary[] {new Binary(new byte[] {1, 2, 3})}, 0); + Assert.assertArrayEquals(new byte[] {1, 2, 3}, blobField.getBinaryV().getValues()); + } +}
