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 d8b157e540e7ecac3f0c9746cc84193c2d757f19 Author: Tian Jiang <[email protected]> AuthorDate: Mon Aug 24 18:36:45 2026 +0800 multiple refactors --- .../java/org/apache/iotdb/session/Session.java | 133 +++---- .../iotdb/session/util/SessionTypeServices.java | 435 +++++++++++++++++++++ .../apache/iotdb/session/util/SessionUtils.java | 296 +------------- .../iotdb/session/util/SessionUtilsTest.java | 21 + 4 files changed, 522 insertions(+), 363 deletions(-) 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 648a63845b4..a04226bf338 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 @@ -69,10 +69,8 @@ import org.apache.tsfile.enums.TSDataType; 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; @@ -82,12 +80,10 @@ import org.slf4j.LoggerFactory; 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; @@ -107,6 +103,7 @@ import java.util.concurrent.ThreadLocalRandom; 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 @@ -2492,11 +2489,7 @@ public class Session implements ISession { 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); @@ -2549,11 +2542,7 @@ public class Session implements ISession { } 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); @@ -2580,7 +2569,7 @@ public class Session implements ISession { * @param <T> Input type * @return ordered list */ - private static <T> List<T> sortList(List<T> source, Integer[] index) { + private static <T> List<T> sortList(List<T> source, int[] index) { List<T> sortedList = new ArrayList<>(index.length); for (int position : index) { sortedList.add(source.get(position)); @@ -3669,11 +3658,7 @@ public class Session implements ISession { 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++) { @@ -3707,64 +3692,8 @@ public class Session implements ISession { * @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); } /** @@ -3774,7 +3703,7 @@ public class Session implements ISession { * @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])) { @@ -3784,6 +3713,52 @@ public class Session implements ISession { return sortedBitMap; } + private static int[] sortedIndex(List<Long> 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 00000000000..24119c3ca46 --- /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<ValueLengthCalculator> 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<ValueWriter> 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<TabletColumnOccupationCalculator> + 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<TabletValueWriter> 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<TabletValueEncoder> 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<ValueListSorter> 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 da15b5a0731..aa3bd32b1f7 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,16 +24,12 @@ import org.apache.iotdb.rpc.IoTDBConnectionException; 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.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; @@ -42,17 +38,13 @@ import org.slf4j.LoggerFactory; 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)); @@ -118,39 +110,8 @@ public class SessionUtils { 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("Data type %s is not supported.", dataType)); - } - return valueOccupation; + return SessionTypeServices.tabletColumnOccupationCalculator(dataType) + .calculate(values, columnIndex, rowSize); } public static ByteBuffer getValueBuffer( @@ -161,47 +122,17 @@ public class SessionUtils { return buffer; } + public static Object sortValueList(Object valueList, TSDataType dataType, int[] index) { + return SessionTypeServices.valueListSorter(dataType).sort(valueList, index); + } + public static int calculateLength(List<TSDataType> 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; } @@ -226,47 +157,9 @@ public class SessionUtils { 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( "Cannot put values for measurement {}, type={}", measurements.get(i), types.get(i), e); @@ -276,98 +169,11 @@ public class SessionUtils { 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("Data type %s is not supported.", dataType)); - } + SessionTypeServices.tabletValueWriter(dataType).write(tablet, i, valueBuffer); } - @SuppressWarnings({"java:S3776", "java:S6541"}) public static void encodeValue( TSDataType dataType, Tablet tablet, @@ -375,85 +181,7 @@ public class SessionUtils { 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("Data type %s is not supported.", 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 ffdc5835dfc..e2c1df282d6 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 @@ -146,6 +146,27 @@ public class SessionUtilsTest { } } + // 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<TSDataType> types = + Arrays.asList(TSDataType.DATE, TSDataType.TIMESTAMP, TSDataType.STRING, TSDataType.BLOB); + List<Object> values = + Arrays.asList( + LocalDate.of(2024, 4, 1), + 123L, + new Binary(new byte[] {1, 2}), + new Binary(new byte[] {3, 4, 5})); + List<String> measurements = Arrays.asList("date", "timestamp", "string", "blob"); + + ByteBuffer buffer = SessionUtils.getValueBuffer(types, values, measurements); + + Assert.assertEquals( + 1 + Integer.BYTES + 1 + Long.BYTES + 1 + Integer.BYTES + 2 + 1 + Integer.BYTES + 3, + buffer.limit()); + } + @Test public void testGetValueBuffer3() { List<IMeasurementSchema> schemas = new ArrayList<>();
