You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/08/25 12:37:33 UTC
(iotdb) branch remove_swtich_type updated (b239e8ae577 -> 616b34b343d)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a change to branch remove_swtich_type
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from b239e8ae577 multiple refactors
new bf032def159 multiple refactors
new d8b157e540e multiple refactors
new 616b34b343d multiple refactors
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../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 +
.../process/window/partition/frame/RangeFrame.java | 294 ++++----------
.../org/apache/iotdb/calc/utils/TypeServices.java | 230 +++++++++++
.../aggregation/VarianceAccumulatorTest.java | 3 +-
.../apache/iotdb/calc/utils/TypeServicesTest.java | 112 ++++++
.../apache/iotdb/db/utils/SerializeUtilsTest.java | 95 +++++
.../apache/iotdb/commons/utils/SerializeUtils.java | 382 +-----------------
10 files changed, 1050 insertions(+), 951 deletions(-)
create mode 100644 iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionTypeServices.java
(iotdb) 02/03: multiple refactors
Posted by ji...@apache.org.
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 <jt...@163.com>
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<>();
(iotdb) 01/03: multiple refactors
Posted by ji...@apache.org.
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 bf032def159e537be39faa80ccbb842ba3bdc1fc
Author: Tian Jiang <jt...@163.com>
AuthorDate: Mon Aug 24 16:46:04 2026 +0800
multiple refactors
---
.../process/window/partition/frame/RangeFrame.java | 294 +++++----------------
.../org/apache/iotdb/calc/utils/TypeServices.java | 230 ++++++++++++++++
.../aggregation/VarianceAccumulatorTest.java | 3 +-
.../apache/iotdb/calc/utils/TypeServicesTest.java | 112 ++++++++
4 files changed, 412 insertions(+), 227 deletions(-)
diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/window/partition/frame/RangeFrame.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/window/partition/frame/RangeFrame.java
index cf59324bdd9..6e53ce6549d 100644
--- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/window/partition/frame/RangeFrame.java
+++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/window/partition/frame/RangeFrame.java
@@ -24,10 +24,11 @@ import org.apache.iotdb.calc.execution.operator.process.window.utils.ColumnList;
import org.apache.iotdb.calc.execution.operator.process.window.utils.Range;
import org.apache.iotdb.calc.execution.operator.process.window.utils.RowComparator;
import org.apache.iotdb.calc.i18n.CalcMessages;
+import org.apache.iotdb.calc.utils.TypeServices;
import org.apache.iotdb.commons.exception.SemanticException;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.write.UnSupportedDataTypeException;
+import org.apache.tsfile.read.common.type.Type;
import java.util.List;
@@ -44,6 +45,7 @@ public class RangeFrame implements Frame {
private List<ColumnList> allSortedColumns;
private ColumnList column;
private TSDataType dataType;
+ private TypeServices.RangeFrameComparator rangeFrameComparator;
private int partitionSize;
private RowComparator peerGroupComparator;
@@ -69,6 +71,8 @@ public class RangeFrame implements Frame {
this.allSortedColumns = sortedColumns;
this.column = sortedColumns.get(0);
this.dataType = column.getDataType();
+ this.rangeFrameComparator =
+ TypeServices.RANGE_FRAME_COMPARATOR_SERVICE.call(Type.fromTsDataType(dataType));
this.peerGroupComparator = comparator;
this.recentRange = new Range(0, 0);
}
@@ -285,34 +289,14 @@ public class RangeFrame implements Frame {
private boolean compareInAscFrameStartFollowing(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int followInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return followInt >= currentInt + deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long followLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return followLong >= currentLong + deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float followFloat = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return followFloat >= currentFloat + deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double followDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return followDouble >= currentDouble + deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.GREATER_THAN_OR_EQUAL);
}
// Find first row which satisfy:
@@ -331,34 +315,14 @@ public class RangeFrame implements Frame {
private boolean compareInAscFrameEndFollowing(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int followInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return followInt > currentInt + deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long followLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return followLong > currentLong + deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float followFloat = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return followFloat > currentFloat + deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double followDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return followDouble > currentDouble + deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.GREATER_THAN);
}
// Find first row which satisfy:
@@ -377,34 +341,14 @@ public class RangeFrame implements Frame {
private boolean compareInAscFrameStartPreceding(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int precedeInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return precedeInt >= currentInt - deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long precedeLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return precedeLong >= currentLong - deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float precedeFollow = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return precedeFollow >= currentFloat - deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double precedeDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return precedeDouble >= currentDouble - deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.SUBTRACT,
+ TypeServices.RangeFrameComparison.GREATER_THAN_OR_EQUAL);
}
// Find first row which satisfy:
@@ -423,34 +367,14 @@ public class RangeFrame implements Frame {
private boolean compareInAscFrameEndPreceding(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int precedeInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return precedeInt > currentInt - deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long precedeLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return precedeLong > currentLong - deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float precedeFollow = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return precedeFollow > currentFloat - deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double precedeDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return precedeDouble > currentDouble - deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.SUBTRACT,
+ TypeServices.RangeFrameComparison.GREATER_THAN);
}
// Find first row which satisfy:
@@ -469,34 +393,14 @@ public class RangeFrame implements Frame {
private boolean compareInDescFrameStartFollowing(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int followInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return followInt <= currentInt - deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long followLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return followLong <= currentLong - deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float followFloat = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return followFloat <= currentFloat - deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double followDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return followDouble <= currentDouble - deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.SUBTRACT,
+ TypeServices.RangeFrameComparison.LESS_THAN_OR_EQUAL);
}
// Find first row which satisfy:
@@ -515,34 +419,14 @@ public class RangeFrame implements Frame {
private boolean compareInDescFrameEndFollowing(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int followInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return followInt < currentInt - deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long followLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return followLong < currentLong - deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float followFloat = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return followFloat < currentFloat - deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double followDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return followDouble < currentDouble - deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.SUBTRACT,
+ TypeServices.RangeFrameComparison.LESS_THAN);
}
// Find first row which satisfy:
@@ -561,34 +445,14 @@ public class RangeFrame implements Frame {
private boolean compareInDescFrameStartPreceding(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int precedeInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return precedeInt <= currentInt + deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long precedeLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return precedeLong <= currentLong + deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float precedeFollow = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return precedeFollow <= currentFloat + deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double precedeDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return precedeDouble <= currentDouble + deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.LESS_THAN_OR_EQUAL);
}
// Find first row which satisfy:
@@ -607,33 +471,13 @@ public class RangeFrame implements Frame {
private boolean compareInDescFrameEndPreceding(int currentIndex, int recentIndex, int channel) {
checkArgument(!partition.isNull(channel, currentIndex));
- switch (column.getDataType()) {
- case INT32:
- case DATE:
- int currentInt = column.getInt(currentIndex);
- int precedeInt = column.getInt(recentIndex);
- int deltaInt = partition.getInt(channel, currentIndex);
- return precedeInt < currentInt + deltaInt;
- case INT64:
- case TIMESTAMP:
- long currentLong = column.getLong(currentIndex);
- long precedeLong = column.getLong(recentIndex);
- long deltaLong = partition.getLong(channel, currentIndex);
- return precedeLong < currentLong + deltaLong;
- case FLOAT:
- float currentFloat = column.getFloat(currentIndex);
- float precedeFollow = column.getFloat(recentIndex);
- float deltaFloat = partition.getFloat(channel, currentIndex);
- return precedeFollow < currentFloat + deltaFloat;
- case DOUBLE:
- double currentDouble = column.getDouble(currentIndex);
- double precedeDouble = column.getDouble(recentIndex);
- double deltaDouble = partition.getDouble(channel, currentIndex);
- return precedeDouble < currentDouble + deltaDouble;
- default:
- // Unreachable
- throw new UnSupportedDataTypeException(
- CalcMessages.UNSUPPORTED_DATA_TYPE + column.getDataType());
- }
+ return rangeFrameComparator.compare(
+ column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.LESS_THAN);
}
}
diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/TypeServices.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/TypeServices.java
index a68dae23632..23ab14974df 100644
--- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/TypeServices.java
+++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/TypeServices.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.calc.utils;
import org.apache.iotdb.calc.execution.operator.process.window.partition.Partition;
+import org.apache.iotdb.calc.execution.operator.process.window.utils.ColumnList;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedMaxMinByBaseAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.BinaryBigArray;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.BooleanBigArray;
@@ -153,6 +154,74 @@ public class TypeServices {
};
};
+ // RANGE frame offsets must retain each primitive type's native overflow and precision rules.
+ public static final TypeService<RangeFrameComparator> RANGE_FRAME_COMPARATOR_SERVICE =
+ type ->
+ switch (type.getTypeEnum()) {
+ case INT32, DATE ->
+ (column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ offsetOperation,
+ comparison) -> {
+ int current = column.getInt(currentIndex);
+ int offset = partition.getInt(channel, currentIndex);
+ int boundary = offsetOperation.apply(current, offset);
+ return comparison.compare(column.getInt(recentIndex), boundary);
+ };
+ case INT64, TIMESTAMP ->
+ (column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ offsetOperation,
+ comparison) -> {
+ long current = column.getLong(currentIndex);
+ long offset = partition.getLong(channel, currentIndex);
+ long boundary = offsetOperation.apply(current, offset);
+ return comparison.compare(column.getLong(recentIndex), boundary);
+ };
+ case FLOAT ->
+ (column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ offsetOperation,
+ comparison) -> {
+ float current = column.getFloat(currentIndex);
+ float offset = partition.getFloat(channel, currentIndex);
+ float boundary = offsetOperation.apply(current, offset);
+ return comparison.compare(column.getFloat(recentIndex), boundary);
+ };
+ case DOUBLE ->
+ (column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ offsetOperation,
+ comparison) -> {
+ double current = column.getDouble(currentIndex);
+ double offset = partition.getDouble(channel, currentIndex);
+ double boundary = offsetOperation.apply(current, offset);
+ return comparison.compare(column.getDouble(recentIndex), boundary);
+ };
+ case BOOLEAN, TEXT, BLOB, STRING, OBJECT, ROW, UNKNOWN, VECTOR ->
+ (column,
+ partition,
+ currentIndex,
+ recentIndex,
+ channel,
+ offsetOperation,
+ comparison) -> {
+ throw new UnSupportedDataTypeException(CalcMessages.UNSUPPORTED_DATA_TYPE + type);
+ };
+ };
+
public static final TypeService<Function<DefaultEncodingProvider, TSEncoding>>
DEFAULT_ENCODING_BY_TYPE_SERVICE =
type ->
@@ -290,6 +359,7 @@ public class TypeServices {
MEMORY_USAGE_OF_ONE_SERIALIZABLE_ROW_FIELD_SERVICE.check();
PRIMITIVE_TYPE_VALUE_EXTRACTOR_SERVICE.check();
NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE.check();
+ RANGE_FRAME_COMPARATOR_SERVICE.check();
DEFAULT_ENCODING_BY_TYPE_SERVICE.check();
DEFAULT_VALUE_WRITER_SERVICE.check();
INTERMEDIATE_VALUE_WRITER_SERVICE.check();
@@ -324,6 +394,166 @@ public class TypeServices {
ColumnToDoubleConverter create(Supplier<? extends RuntimeException> exceptionSupplier);
}
+ @FunctionalInterface
+ public interface RangeFrameComparator {
+ boolean compare(
+ ColumnList column,
+ Partition partition,
+ int currentIndex,
+ int recentIndex,
+ int channel,
+ RangeFrameOffsetOperation offsetOperation,
+ RangeFrameComparison comparison);
+ }
+
+ public enum RangeFrameOffsetOperation {
+ ADD {
+ @Override
+ int apply(int value, int offset) {
+ return value + offset;
+ }
+
+ @Override
+ long apply(long value, long offset) {
+ return value + offset;
+ }
+
+ @Override
+ float apply(float value, float offset) {
+ return value + offset;
+ }
+
+ @Override
+ double apply(double value, double offset) {
+ return value + offset;
+ }
+ },
+ SUBTRACT {
+ @Override
+ int apply(int value, int offset) {
+ return value - offset;
+ }
+
+ @Override
+ long apply(long value, long offset) {
+ return value - offset;
+ }
+
+ @Override
+ float apply(float value, float offset) {
+ return value - offset;
+ }
+
+ @Override
+ double apply(double value, double offset) {
+ return value - offset;
+ }
+ };
+
+ abstract int apply(int value, int offset);
+
+ abstract long apply(long value, long offset);
+
+ abstract float apply(float value, float offset);
+
+ abstract double apply(double value, double offset);
+ }
+
+ public enum RangeFrameComparison {
+ GREATER_THAN_OR_EQUAL {
+ @Override
+ boolean compare(int left, int right) {
+ return left >= right;
+ }
+
+ @Override
+ boolean compare(long left, long right) {
+ return left >= right;
+ }
+
+ @Override
+ boolean compare(float left, float right) {
+ return left >= right;
+ }
+
+ @Override
+ boolean compare(double left, double right) {
+ return left >= right;
+ }
+ },
+ GREATER_THAN {
+ @Override
+ boolean compare(int left, int right) {
+ return left > right;
+ }
+
+ @Override
+ boolean compare(long left, long right) {
+ return left > right;
+ }
+
+ @Override
+ boolean compare(float left, float right) {
+ return left > right;
+ }
+
+ @Override
+ boolean compare(double left, double right) {
+ return left > right;
+ }
+ },
+ LESS_THAN_OR_EQUAL {
+ @Override
+ boolean compare(int left, int right) {
+ return left <= right;
+ }
+
+ @Override
+ boolean compare(long left, long right) {
+ return left <= right;
+ }
+
+ @Override
+ boolean compare(float left, float right) {
+ return left <= right;
+ }
+
+ @Override
+ boolean compare(double left, double right) {
+ return left <= right;
+ }
+ },
+ LESS_THAN {
+ @Override
+ boolean compare(int left, int right) {
+ return left < right;
+ }
+
+ @Override
+ boolean compare(long left, long right) {
+ return left < right;
+ }
+
+ @Override
+ boolean compare(float left, float right) {
+ return left < right;
+ }
+
+ @Override
+ boolean compare(double left, double right) {
+ return left < right;
+ }
+ };
+
+ abstract boolean compare(int left, int right);
+
+ abstract boolean compare(long left, long right);
+
+ abstract boolean compare(float left, float right);
+
+ abstract boolean compare(double left, double right);
+ }
+
@FunctionalInterface
public interface DefaultValueWriter {
void write(Partition partition, int channel, int index, ColumnBuilder builder);
diff --git a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/VarianceAccumulatorTest.java b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/VarianceAccumulatorTest.java
index 872eeaa2a33..a11bad4f5e1 100644
--- a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/VarianceAccumulatorTest.java
+++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/VarianceAccumulatorTest.java
@@ -22,10 +22,9 @@ import org.apache.iotdb.calc.execution.aggregation.VarianceAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;
import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.common.block.column.DoubleColumnBuilder;
import org.apache.tsfile.read.common.block.column.DoubleColumn;
+import org.apache.tsfile.read.common.block.column.DoubleColumnBuilder;
import org.apache.tsfile.read.common.block.column.FloatColumn;
import org.apache.tsfile.read.common.block.column.IntColumn;
import org.apache.tsfile.read.common.block.column.LongColumn;
diff --git a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/utils/TypeServicesTest.java b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/utils/TypeServicesTest.java
index 983d969387b..f11222c26fe 100644
--- a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/utils/TypeServicesTest.java
+++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/utils/TypeServicesTest.java
@@ -18,20 +18,28 @@
package org.apache.iotdb.calc.utils;
+import org.apache.iotdb.calc.execution.operator.process.window.partition.Partition;
+import org.apache.iotdb.calc.execution.operator.process.window.utils.ColumnList;
+
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.DoubleColumn;
import org.apache.tsfile.read.common.block.column.FloatColumn;
import org.apache.tsfile.read.common.block.column.IntColumn;
import org.apache.tsfile.read.common.block.column.LongColumn;
+import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.type.Type;
import org.junit.Assert;
import org.junit.Test;
+import java.util.Collections;
import java.util.Optional;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
public class TypeServicesTest {
@@ -86,6 +94,51 @@ public class TypeServicesTest {
.apply(0));
}
+ // Covers every supported RANGE-frame type and guards native integer overflow and long precision.
+ @Test
+ public void testRangeFrameComparatorPreservesNativeArithmetic() {
+ assertRangeFrameComparator(
+ TSDataType.INT32,
+ new IntColumn(3, Optional.empty(), new int[] {10, 12, 8}),
+ new IntColumn(3, Optional.empty(), new int[] {2, 0, 0}));
+ assertRangeFrameComparator(
+ TSDataType.DATE,
+ new IntColumn(3, Optional.empty(), new int[] {10, 12, 8}, TSDataType.DATE),
+ new IntColumn(3, Optional.empty(), new int[] {2, 0, 0}));
+ assertRangeFrameComparator(
+ TSDataType.INT64,
+ new LongColumn(3, Optional.empty(), new long[] {10, 12, 8}),
+ new LongColumn(3, Optional.empty(), new long[] {2, 0, 0}));
+ assertRangeFrameComparator(
+ TSDataType.TIMESTAMP,
+ new LongColumn(3, Optional.empty(), new long[] {10, 12, 8}),
+ new LongColumn(3, Optional.empty(), new long[] {2, 0, 0}));
+ assertRangeFrameComparator(
+ TSDataType.FLOAT,
+ new FloatColumn(3, Optional.empty(), new float[] {10, 12, 8}),
+ new FloatColumn(3, Optional.empty(), new float[] {2, 0, 0}));
+ assertRangeFrameComparator(
+ TSDataType.DOUBLE,
+ new DoubleColumn(3, Optional.empty(), new double[] {10, 12, 8}),
+ new DoubleColumn(3, Optional.empty(), new double[] {2, 0, 0}));
+
+ assertTrue(
+ compareRangeFrameValues(
+ TSDataType.INT32,
+ new IntColumn(2, Optional.empty(), new int[] {Integer.MAX_VALUE, Integer.MIN_VALUE}),
+ new IntColumn(2, Optional.empty(), new int[] {1, 0}),
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.GREATER_THAN_OR_EQUAL));
+ assertFalse(
+ compareRangeFrameValues(
+ TSDataType.INT64,
+ new LongColumn(
+ 2, Optional.empty(), new long[] {9_007_199_254_740_992L, 9_007_199_254_740_992L}),
+ new LongColumn(2, Optional.empty(), new long[] {1, 0}),
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.GREATER_THAN_OR_EQUAL));
+ }
+
private static double convert(final TSDataType dataType, final Column column) {
return TypeServices.NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE
.call(Type.fromTsDataType(dataType))
@@ -95,4 +148,63 @@ public class TypeServicesTest {
"supported data type should not use the exception factory"))
.convert(column, 0);
}
+
+ private static void assertRangeFrameComparator(
+ final TSDataType dataType, final Column valueColumn, final Column offsetColumn) {
+ assertTrue(
+ compareRangeFrameValues(
+ dataType,
+ valueColumn,
+ offsetColumn,
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.GREATER_THAN_OR_EQUAL));
+ assertFalse(
+ compareRangeFrameValues(
+ dataType,
+ valueColumn,
+ offsetColumn,
+ TypeServices.RangeFrameOffsetOperation.ADD,
+ TypeServices.RangeFrameComparison.GREATER_THAN));
+ assertTrue(
+ compareRangeFrameValues(
+ dataType,
+ valueColumn,
+ offsetColumn,
+ TypeServices.RangeFrameOffsetOperation.SUBTRACT,
+ TypeServices.RangeFrameComparison.LESS_THAN_OR_EQUAL));
+ assertFalse(
+ compareRangeFrameValues(
+ dataType,
+ valueColumn,
+ offsetColumn,
+ TypeServices.RangeFrameOffsetOperation.SUBTRACT,
+ TypeServices.RangeFrameComparison.LESS_THAN));
+ }
+
+ private static boolean compareRangeFrameValues(
+ final TSDataType dataType,
+ final Column valueColumn,
+ final Column offsetColumn,
+ final TypeServices.RangeFrameOffsetOperation offsetOperation,
+ final TypeServices.RangeFrameComparison comparison) {
+ final long[] times = new long[valueColumn.getPositionCount()];
+ final TsBlock block =
+ new TsBlock(
+ new TimeColumn(valueColumn.getPositionCount(), times), valueColumn, offsetColumn);
+ final Partition partition =
+ new Partition(Collections.singletonList(block), 0, valueColumn.getPositionCount());
+ final ColumnList columnList = new ColumnList(Collections.singletonList(valueColumn));
+ return TypeServices.RANGE_FRAME_COMPARATOR_SERVICE
+ .call(Type.fromTsDataType(dataType))
+ .compare(
+ columnList,
+ partition,
+ 0,
+ offsetOperation == TypeServices.RangeFrameOffsetOperation.ADD
+ ? 1
+ : valueColumn.getPositionCount() - 1,
+ 1,
+ offsetOperation,
+ comparison);
+ }
}
(iotdb) 03/03: multiple refactors
Posted by ji...@apache.org.
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 616b34b343d42cb9b65053cca0a0d95cc4e04e4e
Author: Tian Jiang <jt...@163.com>
AuthorDate: Tue Aug 25 20:31:45 2026 +0800
multiple refactors
---
.../apache/iotdb/db/utils/SerializeUtilsTest.java | 95 +++++
.../apache/iotdb/commons/utils/SerializeUtils.java | 382 ++-------------------
2 files changed, 116 insertions(+), 361 deletions(-)
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java
index 3a36e4d87ab..38cc89f2ea0 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java
@@ -206,6 +206,33 @@ public class SerializeUtilsTest {
}
}
+ /** DATE and TIMESTAMP use the INT32 and INT64 wire encodings while retaining their type tags. */
+ @Test
+ public void serdesDateAndTimestampBatchDataTest() {
+ BatchData dateData = new BatchData(TSDataType.DATE);
+ BatchData timestampData = new BatchData(TSDataType.TIMESTAMP);
+ for (int i = 0; i < 3; i++) {
+ dateData.putInt(i, 2024 + i);
+ timestampData.putLong(i, 1_000L + i);
+ }
+
+ ByteArrayOutputStream dateBytes = new ByteArrayOutputStream();
+ SerializeUtils.serializeBatchData(dateData, new DataOutputStream(dateBytes));
+ BatchData deserializedDate =
+ SerializeUtils.deserializeBatchData(ByteBuffer.wrap(dateBytes.toByteArray()));
+ ByteArrayOutputStream timestampBytes = new ByteArrayOutputStream();
+ SerializeUtils.serializeBatchData(timestampData, new DataOutputStream(timestampBytes));
+ BatchData deserializedTimestamp =
+ SerializeUtils.deserializeBatchData(ByteBuffer.wrap(timestampBytes.toByteArray()));
+
+ Assert.assertEquals(TSDataType.DATE, deserializedDate.getDataType());
+ Assert.assertEquals(TSDataType.TIMESTAMP, deserializedTimestamp.getDataType());
+ for (int i = 0; i < 3; i++) {
+ Assert.assertEquals(2024 + i, deserializedDate.getIntByIndex(i));
+ Assert.assertEquals(1_000L + i, deserializedTimestamp.getLongByIndex(i));
+ }
+ }
+
/** This method tests SerializeUtils.serializeTVPair() and SerializeUtils.deserializeTVPair() */
@Test
public void serdesTVPairTest() {
@@ -240,6 +267,44 @@ public class SerializeUtilsTest {
}
}
+ /**
+ * Covers the DATE/TIMESTAMP aliases and binary type aliases in TVPair serialization. The
+ * Long.MIN_VALUE timestamp is the wire-format null-value marker and must not consume a value.
+ */
+ @Test
+ public void serdesAdditionalTVPairTypesTest() {
+ List<TimeValuePair> pairs =
+ Arrays.asList(
+ new TimeValuePair(1, Type.fromTsDataType(TSDataType.DATE).getTsPrimitiveType(2024)),
+ new TimeValuePair(2, Type.fromTsDataType(TSDataType.TIMESTAMP).getTsPrimitiveType(2L)),
+ new TimeValuePair(
+ 3,
+ Type.fromTsDataType(TSDataType.BLOB)
+ .getTsPrimitiveType(BytesUtils.valueOf("blob"))),
+ new TimeValuePair(
+ 4,
+ Type.fromTsDataType(TSDataType.STRING)
+ .getTsPrimitiveType(BytesUtils.valueOf("string"))));
+
+ for (TimeValuePair pair : pairs) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ SerializeUtils.serializeTVPair(pair, new DataOutputStream(baos));
+ Assert.assertEquals(
+ pair, SerializeUtils.deserializeTVPair(ByteBuffer.wrap(baos.toByteArray())));
+ }
+
+ // A sentinel timestamp represents a null value; serialization writes no value bytes after it.
+ TimeValuePair nullValue =
+ new TimeValuePair(
+ Long.MIN_VALUE, Type.fromTsDataType(TSDataType.INT32).getTsPrimitiveType(42));
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ SerializeUtils.serializeTVPair(nullValue, new DataOutputStream(baos));
+ TimeValuePair deserialized =
+ SerializeUtils.deserializeTVPair(ByteBuffer.wrap(baos.toByteArray()));
+ Assert.assertEquals(Long.MIN_VALUE, deserialized.getTimestamp());
+ Assert.assertNull(deserialized.getValue());
+ }
+
/** This method tests SerializeUtils.serializeTVPairs() and SerializeUtils.deserializeTVPairs() */
@Test
public void serdesTVPairsTest() {
@@ -274,6 +339,36 @@ public class SerializeUtilsTest {
}
}
+ /** Ensures the batch TVPair dispatcher uses the same primitive encodings as single pairs. */
+ @Test
+ public void serdesAdditionalTVPairsTypesTest() {
+ List<List<TimeValuePair>> pairs =
+ Arrays.asList(
+ Collections.singletonList(
+ new TimeValuePair(
+ 1, Type.fromTsDataType(TSDataType.DATE).getTsPrimitiveType(2024))),
+ Collections.singletonList(
+ new TimeValuePair(
+ 2, Type.fromTsDataType(TSDataType.TIMESTAMP).getTsPrimitiveType(2L))),
+ Collections.singletonList(
+ new TimeValuePair(
+ 3,
+ Type.fromTsDataType(TSDataType.BLOB)
+ .getTsPrimitiveType(BytesUtils.valueOf("blob")))),
+ Collections.singletonList(
+ new TimeValuePair(
+ 4,
+ Type.fromTsDataType(TSDataType.STRING)
+ .getTsPrimitiveType(BytesUtils.valueOf("string")))));
+
+ for (List<TimeValuePair> pair : pairs) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ SerializeUtils.serializeTVPairs(pair, new DataOutputStream(baos));
+ Assert.assertEquals(
+ pair, SerializeUtils.deserializeTVPairs(ByteBuffer.wrap(baos.toByteArray())));
+ }
+ }
+
/** This method tests SerializeUtils.serializeObject() and SerializeUtils.deserializeObject() */
@Test
public void serdesObjectTest() {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/SerializeUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/SerializeUtils.java
index 1b0b360b809..6e1ded9e40b 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/SerializeUtils.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/SerializeUtils.java
@@ -29,9 +29,7 @@ import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.common.BatchData;
import org.apache.tsfile.read.common.BatchData.BatchDataType;
import org.apache.tsfile.read.common.type.Type;
-import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.ReadWriteIOUtils;
-import org.apache.tsfile.utils.TsPrimitiveType;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
@@ -161,187 +159,20 @@ public class SerializeUtils {
int length = buffer.getInt();
TSDataType dataType = TSDataType.values()[buffer.get()];
BatchData batchData = BatchDataType.deserialize(buffer.get(), dataType);
- switch (dataType) {
- case DATE:
- case INT32:
- for (int i = 0; i < length; i++) {
- batchData.putInt(buffer.getLong(), buffer.getInt());
- }
- break;
- case TIMESTAMP:
- case INT64:
- for (int i = 0; i < length; i++) {
- batchData.putLong(buffer.getLong(), buffer.getLong());
- }
- break;
- case BLOB:
- case OBJECT:
- case STRING:
- case TEXT:
- for (int i = 0; i < length; i++) {
- long time = buffer.getLong();
- int len = buffer.getInt();
- byte[] bytes = new byte[len];
- buffer.get(bytes);
- batchData.putBinary(time, new Binary(bytes));
- }
- break;
- case FLOAT:
- for (int i = 0; i < length; i++) {
- batchData.putFloat(buffer.getLong(), buffer.getFloat());
- }
- break;
- case DOUBLE:
- for (int i = 0; i < length; i++) {
- batchData.putDouble(buffer.getLong(), buffer.getDouble());
- }
- break;
- case BOOLEAN:
- for (int i = 0; i < length; i++) {
- batchData.putBoolean(buffer.getLong(), buffer.get() == 1);
- }
- break;
- case VECTOR:
- for (int i = 0; i < length; i++) {
- long time = buffer.getLong();
- int valuesLength = buffer.getInt();
- TsPrimitiveType[] values = new TsPrimitiveType[valuesLength];
- for (int j = 0; j < valuesLength; j++) {
- boolean notNull = (buffer.get() == 1);
- if (notNull) {
- switch (TSDataType.values()[buffer.get()]) {
- case BOOLEAN:
- values[j] = new TsPrimitiveType.TsBoolean(buffer.get() == 1);
- break;
- case DOUBLE:
- values[j] = new TsPrimitiveType.TsDouble(buffer.getDouble());
- break;
- case FLOAT:
- values[j] = new TsPrimitiveType.TsFloat(buffer.getFloat());
- break;
- case BLOB:
- case OBJECT:
- case STRING:
- case TEXT:
- int len = buffer.getInt();
- byte[] bytes = new byte[len];
- buffer.get(bytes);
- values[j] = new TsPrimitiveType.TsBinary(new Binary(bytes));
- break;
- case TIMESTAMP:
- case INT64:
- values[j] = new TsPrimitiveType.TsLong(buffer.getLong());
- break;
- case DATE:
- case INT32:
- values[j] = new TsPrimitiveType.TsInt(buffer.getInt());
- break;
- default:
- break;
- }
- }
- }
- batchData.putVector(time, values);
- }
- break;
- }
+ Type.fromTsDataType(dataType).deserialize(buffer, batchData, length);
batchData.resetBatchData();
return batchData;
}
- private static void serializeTextTVPairs(
- List<TimeValuePair> timeValuePairs, DataOutputStream dataOutputStream) throws IOException {
- for (TimeValuePair timeValuePair : timeValuePairs) {
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeInt(timeValuePair.getValue().getBinary().getLength());
- dataOutputStream.write(timeValuePair.getValue().getBinary().getValues());
- }
- }
- }
-
- private static void serializeBooleanTVPairs(
- List<TimeValuePair> timeValuePairs, DataOutputStream dataOutputStream) throws IOException {
- for (TimeValuePair timeValuePair : timeValuePairs) {
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeBoolean(timeValuePair.getValue().getBoolean());
- }
- }
- }
-
- private static void serializeIntTVPairs(
- List<TimeValuePair> timeValuePairs, DataOutputStream dataOutputStream) throws IOException {
- for (TimeValuePair timeValuePair : timeValuePairs) {
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeInt(timeValuePair.getValue().getInt());
- }
- }
- }
-
- private static void serializeLongTVPairs(
- List<TimeValuePair> timeValuePairs, DataOutputStream dataOutputStream) throws IOException {
- for (TimeValuePair timeValuePair : timeValuePairs) {
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeLong(timeValuePair.getValue().getLong());
- }
- }
- }
-
- private static void serializeFloatTVPairs(
- List<TimeValuePair> timeValuePairs, DataOutputStream dataOutputStream) throws IOException {
- for (TimeValuePair timeValuePair : timeValuePairs) {
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeFloat(timeValuePair.getValue().getFloat());
- }
- }
- }
-
- private static void serializeDoubleTVPairs(
- List<TimeValuePair> timeValuePairs, DataOutputStream dataOutputStream) throws IOException {
- for (TimeValuePair timeValuePair : timeValuePairs) {
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeDouble(timeValuePair.getValue().getDouble());
- }
- }
- }
-
public static void serializeTVPairs(
List<TimeValuePair> timeValuePairs, DataOutputStream dataOutputStream) {
try {
TSDataType dataType = timeValuePairs.get(0).getValue().getDataType();
dataOutputStream.write(dataType.ordinal());
dataOutputStream.writeInt(timeValuePairs.size());
- switch (timeValuePairs.get(0).getValue().getDataType()) {
- case BLOB:
- case OBJECT:
- case STRING:
- case TEXT:
- serializeTextTVPairs(timeValuePairs, dataOutputStream);
- break;
- case BOOLEAN:
- serializeBooleanTVPairs(timeValuePairs, dataOutputStream);
- break;
- case TIMESTAMP:
- case INT64:
- serializeLongTVPairs(timeValuePairs, dataOutputStream);
- break;
- case DATE:
- case INT32:
- serializeIntTVPairs(timeValuePairs, dataOutputStream);
- break;
- case FLOAT:
- serializeFloatTVPairs(timeValuePairs, dataOutputStream);
- break;
- case DOUBLE:
- serializeDoubleTVPairs(timeValuePairs, dataOutputStream);
- break;
- default:
- break;
+ Type type = Type.fromTsDataType(dataType);
+ for (TimeValuePair timeValuePair : timeValuePairs) {
+ serializeTVPair(type, timeValuePair, dataOutputStream);
}
} catch (IOException ignored) {
// unreachable
@@ -356,138 +187,19 @@ public class SerializeUtils {
TSDataType dataType = timeValuePair.getValue().getDataType();
try {
dataOutputStream.write(dataType.ordinal());
- switch (dataType) {
- case STRING:
- case BLOB:
- case OBJECT:
- case TEXT:
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeInt(timeValuePair.getValue().getBinary().getLength());
- dataOutputStream.write(timeValuePair.getValue().getBinary().getValues());
- }
- break;
- case BOOLEAN:
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeBoolean(timeValuePair.getValue().getBoolean());
- }
- break;
- case TIMESTAMP:
- case INT64:
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeLong(timeValuePair.getValue().getLong());
- }
- break;
- case DATE:
- case INT32:
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeInt(timeValuePair.getValue().getInt());
- }
- break;
- case FLOAT:
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeFloat(timeValuePair.getValue().getFloat());
- }
- break;
- case DOUBLE:
- dataOutputStream.writeLong(timeValuePair.getTimestamp());
- if (timeValuePair.getTimestamp() != Long.MIN_VALUE) {
- dataOutputStream.writeDouble(timeValuePair.getValue().getDouble());
- }
- break;
- default:
- break;
- }
+ serializeTVPair(Type.fromTsDataType(dataType), timeValuePair, dataOutputStream);
} catch (IOException e) {
// unreachable
}
}
- private static void deserializeDoubleTVPairs(
- ByteBuffer buffer, List<TimeValuePair> ret, int size, TSDataType dataType) {
- for (int i = 0; i < size; i++) {
- long time = buffer.getLong();
- TimeValuePair pair =
- time != Long.MIN_VALUE
- ? new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getDouble()))
- : new TimeValuePair(time, null);
- ret.add(pair);
- }
- }
-
- private static void deserializeFloatTVPairs(
- ByteBuffer buffer, List<TimeValuePair> ret, int size, TSDataType dataType) {
- for (int i = 0; i < size; i++) {
- long time = buffer.getLong();
- TimeValuePair pair =
- time != Long.MIN_VALUE
- ? new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getFloat()))
- : new TimeValuePair(time, null);
- ret.add(pair);
- }
- }
-
- private static void deserializeIntTVPairs(
- ByteBuffer buffer, List<TimeValuePair> ret, int size, TSDataType dataType) {
- for (int i = 0; i < size; i++) {
- long time = buffer.getLong();
- TimeValuePair pair =
- time != Long.MIN_VALUE
- ? new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getInt()))
- : new TimeValuePair(time, null);
- ret.add(pair);
- }
- }
-
- private static void deserializeLongTVPairs(
- ByteBuffer buffer, List<TimeValuePair> ret, int size, TSDataType dataType) {
- for (int i = 0; i < size; i++) {
- long time = buffer.getLong();
- TimeValuePair pair =
- time != Long.MIN_VALUE
- ? new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getLong()))
- : new TimeValuePair(time, null);
- ret.add(pair);
- }
- }
-
- private static void deserializeBooleanTVPairs(
- ByteBuffer buffer, List<TimeValuePair> ret, int size, TSDataType dataType) {
- for (int i = 0; i < size; i++) {
- long time = buffer.getLong();
- TimeValuePair pair =
- time != Long.MIN_VALUE
- ? new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.get() == 1))
- : new TimeValuePair(time, null);
- ret.add(pair);
- }
- }
-
- private static void deserializeTextTVPairs(
- ByteBuffer buffer, List<TimeValuePair> ret, int size, TSDataType dataType) {
- for (int i = 0; i < size; i++) {
- long time = buffer.getLong();
- TimeValuePair pair;
- if (time != Long.MIN_VALUE) {
- int bytesLen = buffer.getInt();
- byte[] bytes = new byte[bytesLen];
- buffer.get(bytes);
- TsPrimitiveType primitiveType =
- Type.fromTsDataType(dataType).getTsPrimitiveType(new Binary(bytes));
- pair = new TimeValuePair(time, primitiveType);
- } else {
- pair = new TimeValuePair(time, null);
- }
- ret.add(pair);
+ private static void serializeTVPair(
+ Type type, TimeValuePair timeValuePair, DataOutputStream dataOutputStream)
+ throws IOException {
+ if (timeValuePair.getTimestamp() == Long.MIN_VALUE) {
+ dataOutputStream.writeLong(Long.MIN_VALUE);
+ } else {
+ type.serialize(timeValuePair, dataOutputStream);
}
}
@@ -498,32 +210,9 @@ public class SerializeUtils {
TSDataType dataType = TSDataType.values()[buffer.get()];
int size = buffer.getInt();
List<TimeValuePair> ret = new ArrayList<>(size);
- switch (dataType) {
- case DOUBLE:
- deserializeDoubleTVPairs(buffer, ret, size, dataType);
- break;
- case FLOAT:
- deserializeFloatTVPairs(buffer, ret, size, dataType);
- break;
- case DATE:
- case INT32:
- deserializeIntTVPairs(buffer, ret, size, dataType);
- break;
- case TIMESTAMP:
- case INT64:
- deserializeLongTVPairs(buffer, ret, size, dataType);
- break;
- case BOOLEAN:
- deserializeBooleanTVPairs(buffer, ret, size, dataType);
- break;
- case BLOB:
- case OBJECT:
- case STRING:
- case TEXT:
- deserializeTextTVPairs(buffer, ret, size, dataType);
- break;
- default:
- break;
+ Type type = Type.fromTsDataType(dataType);
+ for (int i = 0; i < size; i++) {
+ ret.add(deserializeTVPair(type, buffer));
}
return ret;
}
@@ -532,42 +221,13 @@ public class SerializeUtils {
if (buffer == null || (buffer.limit() - buffer.position() == 0)) {
return null;
}
- TSDataType dataType = TSDataType.values()[buffer.get()];
+ Type type = Type.fromTsDataType(TSDataType.values()[buffer.get()]);
+ return deserializeTVPair(type, buffer);
+ }
+
+ private static TimeValuePair deserializeTVPair(Type type, ByteBuffer buffer) {
long time = buffer.getLong();
- if (time == Long.MIN_VALUE) {
- return new TimeValuePair(time, null);
- }
- switch (dataType) {
- case DOUBLE:
- return new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getDouble()));
- case FLOAT:
- return new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getFloat()));
- case DATE:
- case INT32:
- return new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getInt()));
- case TIMESTAMP:
- case INT64:
- return new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.getLong()));
- case BOOLEAN:
- return new TimeValuePair(
- time, Type.fromTsDataType(dataType).getTsPrimitiveType(buffer.get() == 1));
- case BLOB:
- case OBJECT:
- case STRING:
- case TEXT:
- int bytesLen = buffer.getInt();
- byte[] bytes = new byte[bytesLen];
- buffer.get(bytes);
- TsPrimitiveType primitiveType =
- Type.fromTsDataType(dataType).getTsPrimitiveType(new Binary(bytes));
- return new TimeValuePair(time, primitiveType);
- default:
- return null;
- }
+ return new TimeValuePair(time, time == Long.MIN_VALUE ? null : type.deserialize(buffer));
}
public static void serializeObject(Object object, DataOutputStream dataOutputStream) {