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 9e58ae029a2c10e4ace4ccd12e1559c49c932488 Author: Tian Jiang <[email protected]> AuthorDate: Thu Aug 13 18:36:21 2026 +0800 multiple refactors --- .../org/apache/iotdb/calc/i18n/CalcMessages.java | 2 + .../org/apache/iotdb/calc/i18n/CalcMessages.java | 2 + .../aggregation/CentralMomentAccumulator.java | 36 +++---- .../aggregation/TableCentralMomentAccumulator.java | 38 +++---- .../grouped/GroupedCentralMomentAccumulator.java | 38 +++---- .../org/apache/iotdb/calc/utils/TypeServices.java | 36 ++++++- .../apache/iotdb/calc/utils/TypeServicesTest.java | 98 ++++++++++++++++++ .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 ++ .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 ++ .../iotdb/db/pipe/event/common/row/PipeRow.java | 28 +---- .../org/apache/iotdb/db/utils/TypeServices.java | 30 +++++- .../db/pipe/event/common/row/PipeRowTest.java | 115 +++++++++++++++++++++ 12 files changed, 340 insertions(+), 95 deletions(-) diff --git a/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java b/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java index 3a8a2afb8f2..c5c339d6df5 100644 --- a/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java +++ b/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java @@ -133,6 +133,8 @@ public final class CalcMessages { public static final String UNSUPPORTED_COMPARISON_OPERATOR = "Unsupported comparison operator: "; public static final String UNSUPPORTED_DATA_TYPE = "Unsupported data type: "; + public static final String UNSUPPORTED_DATA_TYPE_IN_CENTRAL_MOMENT_AGGREGATION = + "Unsupported data type in CentralMoment Aggregation: %s"; public static final String UNSUPPORTED_DEFAULT_VALUE_DATA_TYPE_IN_LAG = "Unsupported default value's data type in Lag: "; public static final String UNSUPPORTED_DATA_TYPE_LOWER = "unsupported data type: "; diff --git a/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java b/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java index f3c3aa34f87..05a8e667fbf 100644 --- a/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java +++ b/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java @@ -126,6 +126,8 @@ public final class CalcMessages { public static final String UNSUPPORTED_COLUMN_TRANSFORMER = "不支持的 ColumnTransformer"; public static final String UNSUPPORTED_COMPARISON_OPERATOR = "不支持的比较运算符："; public static final String UNSUPPORTED_DATA_TYPE = "不支持的数据类型："; + public static final String UNSUPPORTED_DATA_TYPE_IN_CENTRAL_MOMENT_AGGREGATION = + "CentralMoment 聚合中不支持的数据类型：%s"; public static final String UNSUPPORTED_DEFAULT_VALUE_DATA_TYPE_IN_LAG = "Lag 中不支持的默认值数据类型："; public static final String UNSUPPORTED_DATA_TYPE_LOWER = "不支持的数据类型："; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/aggregation/CentralMomentAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/aggregation/CentralMomentAccumulator.java index f3ae6ed6e69..ddd381cfc9a 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/aggregation/CentralMomentAccumulator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/aggregation/CentralMomentAccumulator.java @@ -14,10 +14,14 @@ package org.apache.iotdb.calc.execution.aggregation; +import org.apache.iotdb.calc.i18n.CalcMessages; +import org.apache.iotdb.calc.utils.TypeServices; + import org.apache.tsfile.block.column.Column; import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.statistics.Statistics; +import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; @@ -35,7 +39,7 @@ public class CentralMomentAccumulator implements Accumulator { KURTOSIS } - private final TSDataType seriesDataType; + private final TypeServices.ColumnToDoubleConverter doubleValueConverter; private final MomentType momentType; private long count; @@ -45,7 +49,15 @@ public class CentralMomentAccumulator implements Accumulator { private double m4; public CentralMomentAccumulator(TSDataType seriesDataType, MomentType momentType) { - this.seriesDataType = seriesDataType; + this.doubleValueConverter = + TypeServices.NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE + .call(Type.fromTsDataType(seriesDataType)) + .create( + () -> + new UnsupportedOperationException( + String.format( + CalcMessages.UNSUPPORTED_DATA_TYPE_IN_CENTRAL_MOMENT_AGGREGATION, + seriesDataType))); this.momentType = momentType; } @@ -60,25 +72,7 @@ public class CentralMomentAccumulator implements Accumulator { if (columns[1].isNull(i)) { continue; } - update(getDoubleValue(columns[1], i)); - } - } - - private double getDoubleValue(Column column, int position) { - switch (seriesDataType) { - case INT32: - case DATE: - return column.getInt(position); - case INT64: - case TIMESTAMP: - return column.getLong(position); - case FLOAT: - return column.getFloat(position); - case DOUBLE: - return column.getDouble(position); - default: - throw new UnsupportedOperationException( - "Unsupported data type in CentralMoment Aggregation: " + seriesDataType); + update(doubleValueConverter.convert(columns[1], i)); } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableCentralMomentAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableCentralMomentAccumulator.java index e3a7b629dcb..97a095c1838 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableCentralMomentAccumulator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/TableCentralMomentAccumulator.java @@ -15,6 +15,8 @@ package org.apache.iotdb.calc.execution.operator.source.relational.aggregation; import org.apache.iotdb.calc.execution.aggregation.CentralMomentAccumulator; +import org.apache.iotdb.calc.i18n.CalcMessages; +import org.apache.iotdb.calc.utils.TypeServices; import org.apache.tsfile.block.column.Column; import org.apache.tsfile.block.column.ColumnBuilder; @@ -23,6 +25,7 @@ import org.apache.tsfile.file.metadata.statistics.Statistics; import org.apache.tsfile.read.common.block.column.BinaryColumn; import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder; import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn; +import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.RamUsageEstimator; import org.apache.tsfile.write.UnSupportedDataTypeException; @@ -39,6 +42,7 @@ public class TableCentralMomentAccumulator implements TableAccumulator { private static final double EPSILON = 1e-12; private final TSDataType seriesDataType; + private final TypeServices.ColumnToDoubleConverter doubleValueConverter; private final CentralMomentAccumulator.MomentType momentType; private long count; @@ -50,6 +54,15 @@ public class TableCentralMomentAccumulator implements TableAccumulator { public TableCentralMomentAccumulator( TSDataType seriesDataType, CentralMomentAccumulator.MomentType momentType) { this.seriesDataType = seriesDataType; + this.doubleValueConverter = + TypeServices.NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE + .call(Type.fromTsDataType(seriesDataType)) + .create( + () -> + new UnSupportedDataTypeException( + String.format( + CalcMessages.UNSUPPORTED_DATA_TYPE_IN_CENTRAL_MOMENT_AGGREGATION, + seriesDataType))); this.momentType = momentType; } @@ -59,7 +72,7 @@ public class TableCentralMomentAccumulator implements TableAccumulator { if (mask.isSelectAll()) { for (int i = 0; i < positionCount; i++) { if (!arguments[0].isNull(i)) { - update(getDoubleValue(arguments[0], i)); + update(doubleValueConverter.convert(arguments[0], i)); } } } else { @@ -67,31 +80,12 @@ public class TableCentralMomentAccumulator implements TableAccumulator { for (int i = 0; i < positionCount; i++) { int position = selectedPositions[i]; if (!arguments[0].isNull(position)) { - update(getDoubleValue(arguments[0], position)); + update(doubleValueConverter.convert(arguments[0], position)); } } } } - private double getDoubleValue(Column column, int position) { - switch (seriesDataType) { - case INT32: - case DATE: - return column.getInt(position); - case INT64: - case TIMESTAMP: - return column.getLong(position); - case FLOAT: - return column.getFloat(position); - case DOUBLE: - return column.getDouble(position); - default: - throw new UnSupportedDataTypeException( - String.format( - "Unsupported data type in CentralMoment Aggregation: %s", seriesDataType)); - } - } - private void update(double value) { long n1 = count; long n = n1 + 1; @@ -236,7 +230,7 @@ public class TableCentralMomentAccumulator implements TableAccumulator { return; } - double value = getDoubleValue(arguments[0], 0); + double value = doubleValueConverter.convert(arguments[0], 0); if (count == 1) { reset(); return; diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/GroupedCentralMomentAccumulator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/GroupedCentralMomentAccumulator.java index 246edc6bcd9..0d4c89d104b 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/GroupedCentralMomentAccumulator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/GroupedCentralMomentAccumulator.java @@ -23,6 +23,8 @@ import org.apache.iotdb.calc.execution.aggregation.CentralMomentAccumulator; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.DoubleBigArray; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.LongBigArray; +import org.apache.iotdb.calc.i18n.CalcMessages; +import org.apache.iotdb.calc.utils.TypeServices; import org.apache.tsfile.block.column.Column; import org.apache.tsfile.block.column.ColumnBuilder; @@ -30,6 +32,7 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.block.column.BinaryColumn; import org.apache.tsfile.read.common.block.column.BinaryColumnBuilder; import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn; +import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.RamUsageEstimator; import org.apache.tsfile.write.UnSupportedDataTypeException; @@ -45,7 +48,7 @@ public class GroupedCentralMomentAccumulator implements GroupedAccumulator { private static final int INTERMEDIATE_SIZE = Long.BYTES + 4 * Double.BYTES; private static final double EPSILON = 1e-12; - private final TSDataType seriesDataType; + private final TypeServices.ColumnToDoubleConverter doubleValueConverter; private final CentralMomentAccumulator.MomentType momentType; private final LongBigArray counts = new LongBigArray(); @@ -56,7 +59,15 @@ public class GroupedCentralMomentAccumulator implements GroupedAccumulator { public GroupedCentralMomentAccumulator( TSDataType seriesDataType, CentralMomentAccumulator.MomentType momentType) { - this.seriesDataType = seriesDataType; + this.doubleValueConverter = + TypeServices.NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE + .call(Type.fromTsDataType(seriesDataType)) + .create( + () -> + new UnSupportedDataTypeException( + String.format( + CalcMessages.UNSUPPORTED_DATA_TYPE_IN_CENTRAL_MOMENT_AGGREGATION, + seriesDataType))); this.momentType = momentType; } @@ -85,7 +96,7 @@ public class GroupedCentralMomentAccumulator implements GroupedAccumulator { if (mask.isSelectAll()) { for (int i = 0; i < positionCount; i++) { if (!arguments[0].isNull(i)) { - update(groupIds[i], getDoubleValue(arguments[0], i)); + update(groupIds[i], doubleValueConverter.convert(arguments[0], i)); } } } else { @@ -93,31 +104,12 @@ public class GroupedCentralMomentAccumulator implements GroupedAccumulator { for (int i = 0; i < positionCount; i++) { int position = selectedPositions[i]; if (!arguments[0].isNull(position)) { - update(groupIds[position], getDoubleValue(arguments[0], position)); + update(groupIds[position], doubleValueConverter.convert(arguments[0], position)); } } } } - private double getDoubleValue(Column column, int position) { - switch (seriesDataType) { - case INT32: - case DATE: - return column.getInt(position); - case INT64: - case TIMESTAMP: - return column.getLong(position); - case FLOAT: - return column.getFloat(position); - case DOUBLE: - return column.getDouble(position); - default: - throw new UnSupportedDataTypeException( - String.format( - "Unsupported data type in CentralMoment Aggregation: %s", seriesDataType)); - } - } - private void update(int groupId, double value) { long n1 = counts.get(groupId); long n = n1 + 1; 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 a70021361f4..a68dae23632 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 @@ -30,9 +30,9 @@ import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.gr import org.apache.iotdb.calc.i18n.CalcMessages; import org.apache.iotdb.calc.utils.datastructure.SortKey; +import org.apache.tsfile.block.column.Column; import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.file.metadata.enums.TSEncoding; -import org.apache.tsfile.read.common.type.service.IntTypeService; import org.apache.tsfile.read.common.type.service.TypeService; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.DateUtils; @@ -44,6 +44,7 @@ import java.util.Comparator; import java.util.function.Function; import java.util.function.IntFunction; import java.util.function.IntUnaryOperator; +import java.util.function.Supplier; import static org.apache.iotdb.calc.transformation.datastructure.util.BinaryUtils.MIN_ARRAY_HEADER_SIZE; import static org.apache.iotdb.calc.transformation.datastructure.util.BinaryUtils.MIN_OBJECT_HEADER_SIZE; @@ -82,12 +83,15 @@ public class TypeServices { Comparator.comparing( sortKey -> type.getBoolean(sortKey.tsBlock.getColumn(index), sortKey.rowIndex)); + // TypeService.check() must be able to build a strategy for every enum value. case ROW, UNKNOWN, VECTOR -> - throw new IllegalArgumentException( - String.format(CalcMessages.DATA_TYPE_CANNOT_BE_ORDERED, type)); + index -> { + throw new IllegalArgumentException( + String.format(CalcMessages.DATA_TYPE_CANNOT_BE_ORDERED, type)); + }; }; - public static final IntTypeService MEMORY_USAGE_OF_ONE_MERGE_SORT_KEY_SERVICE = + public static final TypeService<Integer> MEMORY_USAGE_OF_ONE_MERGE_SORT_KEY_SERVICE = type -> switch (type.getTypeEnum()) { case BOOLEAN -> 1; @@ -136,6 +140,19 @@ public class TypeServices { }; }; + // The caller supplies the exception so shared conversion keeps each aggregation API's contract. + public static final TypeService<ColumnToDoubleConverterFactory> + NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE = + type -> + switch (type.getTypeEnum()) { + case INT32, DATE, INT64, TIMESTAMP, FLOAT, DOUBLE -> ignored -> type::getDouble; + case BOOLEAN, TEXT, BLOB, STRING, OBJECT, ROW, UNKNOWN, VECTOR -> + exceptionSupplier -> + (column, position) -> { + throw exceptionSupplier.get(); + }; + }; + public static final TypeService<Function<DefaultEncodingProvider, TSEncoding>> DEFAULT_ENCODING_BY_TYPE_SERVICE = type -> @@ -272,6 +289,7 @@ public class TypeServices { MEMORY_USAGE_OF_ONE_MERGE_SORT_KEY_SERVICE.check(); MEMORY_USAGE_OF_ONE_SERIALIZABLE_ROW_FIELD_SERVICE.check(); PRIMITIVE_TYPE_VALUE_EXTRACTOR_SERVICE.check(); + NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE.check(); DEFAULT_ENCODING_BY_TYPE_SERVICE.check(); DEFAULT_VALUE_WRITER_SERVICE.check(); INTERMEDIATE_VALUE_WRITER_SERVICE.check(); @@ -296,6 +314,16 @@ public class TypeServices { TSEncoding getDefaultTextEncoding(); } + @FunctionalInterface + public interface ColumnToDoubleConverter { + double convert(Column column, int position); + } + + @FunctionalInterface + public interface ColumnToDoubleConverterFactory { + ColumnToDoubleConverter create(Supplier<? extends RuntimeException> exceptionSupplier); + } + @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/utils/TypeServicesTest.java b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/utils/TypeServicesTest.java new file mode 100644 index 00000000000..983d969387b --- /dev/null +++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/utils/TypeServicesTest.java @@ -0,0 +1,98 @@ +/* + * 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.calc.utils; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.enums.TSDataType; +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.type.Type; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Optional; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +public class TypeServicesTest { + + @Test + public void testNumericColumnToDoubleConversion() { + assertEquals( + 1.0, convert(TSDataType.INT32, new IntColumn(1, Optional.empty(), new int[] {1})), 0.0); + assertEquals( + 2.0, + convert( + TSDataType.DATE, new IntColumn(1, Optional.empty(), new int[] {2}, TSDataType.DATE)), + 0.0); + assertEquals( + 3.0, convert(TSDataType.INT64, new LongColumn(1, Optional.empty(), new long[] {3L})), 0.0); + assertEquals( + 4.0, + convert(TSDataType.TIMESTAMP, new LongColumn(1, Optional.empty(), new long[] {4L})), + 0.0); + assertEquals( + 5.5, + convert(TSDataType.FLOAT, new FloatColumn(1, Optional.empty(), new float[] {5.5F})), + 0.0); + assertEquals( + 6.5, + convert(TSDataType.DOUBLE, new DoubleColumn(1, Optional.empty(), new double[] {6.5D})), + 0.0); + } + + @Test + public void testNumericColumnToDoubleConversionUsesCallerException() { + final UnsupportedOperationException expected = new UnsupportedOperationException("expected"); + final TypeServices.ColumnToDoubleConverter converter = + TypeServices.NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE + .call(Type.fromTsDataType(TSDataType.BOOLEAN)) + .create(() -> expected); + + final UnsupportedOperationException actual = + Assert.assertThrows( + UnsupportedOperationException.class, + () -> converter.convert(new IntColumn(1, Optional.empty(), new int[] {1}), 0)); + + assertSame(expected, actual); + } + + @Test + public void testUnsupportedMergeSortComparatorFailsOnUse() { + Assert.assertThrows( + IllegalArgumentException.class, + () -> + TypeServices.MERGE_SORT_COMPARATOR_SERVICE + .call(Type.fromTsDataType(TSDataType.VECTOR)) + .apply(0)); + } + + private static double convert(final TSDataType dataType, final Column column) { + return TypeServices.NUMERIC_COLUMN_TO_DOUBLE_CONVERTER_SERVICE + .call(Type.fromTsDataType(dataType)) + .create( + () -> + new IllegalStateException( + "supported data type should not use the exception factory")) + .convert(column, 0); + } +} diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 13369fc5ee3..a0087258f6b 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -1429,6 +1429,12 @@ public final class DataNodePipeMessages { public static final String FAILED_TO_UNBIND_FROM_PIPE_TSFILE_TO = "Failed to unbind from pipe tsfile to tablets metrics, pipe map is not empty, pipe: {}"; + // --------------------------------------------------------------------------- + // pipe – PipeRow + // --------------------------------------------------------------------------- + public static final String UNSUPPORTED_DATA_TYPE_FOR_COLUMN_FMT = + "unsupported data type %s for column %s"; + // --------------------------------------------------------------------------- // pipe – AbstractSameTypeNumericOperator // --------------------------------------------------------------------------- diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 49ee222c3e9..e768903bff2 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -1376,6 +1376,12 @@ public final class DataNodePipeMessages { public static final String FAILED_TO_UNBIND_FROM_PIPE_TSFILE_TO = "解绑 from pipe tsfile to tablets metrics, pipe map is not empty, pipe: {} 失败"; + // --------------------------------------------------------------------------- + // pipe – PipeRow + // --------------------------------------------------------------------------- + public static final String UNSUPPORTED_DATA_TYPE_FOR_COLUMN_FMT = + "不支持的数据类型 %s，列名为 %s"; + // --------------------------------------------------------------------------- // pipe – AbstractSameTypeNumericOperator // --------------------------------------------------------------------------- diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRow.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRow.java index b0897ed396a..e32d51adf95 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRow.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/row/PipeRow.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.pipe.event.common.row; +import org.apache.iotdb.db.utils.TypeServices; import org.apache.iotdb.pipe.api.access.Row; import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; import org.apache.iotdb.pipe.api.type.Type; @@ -119,30 +120,9 @@ public class PipeRow implements Row { @Override public Object getObject(final int columnIndex) { - switch (getDataType(columnIndex)) { - case INT32: - return getInt(columnIndex); - case DATE: - return getDate(columnIndex); - case INT64: - case TIMESTAMP: - return getLong(columnIndex); - case FLOAT: - return getFloat(columnIndex); - case DOUBLE: - return getDouble(columnIndex); - case BOOLEAN: - return getBoolean(columnIndex); - case TEXT: - case BLOB: - case STRING: - return getBinary(columnIndex); - default: - throw new UnsupportedOperationException( - String.format( - "unsupported data type %s for column %s", - getDataType(columnIndex), columnNameStringList[columnIndex])); - } + return TypeServices.Pipe.PIPE_ROW_OBJECT_GETTER_SERVICE + .call(org.apache.tsfile.read.common.type.Type.fromTsDataType(valueColumnTypes[columnIndex])) + .get(this, columnIndex); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java index c81d0d19ebc..295ebb30939 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/TypeServices.java @@ -86,6 +86,7 @@ import org.apache.iotdb.db.utils.datastructure.LongTVList; import org.apache.iotdb.db.utils.datastructure.TVList; import org.apache.iotdb.db.utils.windowing.window.EvictableBatchList; import org.apache.iotdb.db.utils.windowing.window.WindowImpl; +import org.apache.iotdb.pipe.api.access.Row; import org.apache.iotdb.rpc.RpcUtils; import com.google.common.io.BaseEncoding; @@ -3253,6 +3254,11 @@ public class TypeServices { Object get(Object column, int rowIndex); } + @FunctionalInterface + public interface PipeRowObjectGetter { + Object get(Row row, int columnIndex); + } + @FunctionalInterface public interface AggregateTabletColumnValueWriter { void write(Object column, int rowIndex, Object value); @@ -3312,6 +3318,26 @@ public class TypeServices { DataNodePipeMessages.INVALID_INPUT + type.getTypeEnum()); }; + public static final TypeService<PipeRowObjectGetter> PIPE_ROW_OBJECT_GETTER_SERVICE = + type -> + switch (type.getTypeEnum()) { + case BOOLEAN -> Row::getBoolean; + case INT32 -> Row::getInt; + case DATE -> Row::getDate; + case INT64, TIMESTAMP -> Row::getLong; + case FLOAT -> Row::getFloat; + case DOUBLE -> Row::getDouble; + case TEXT, BLOB, STRING -> Row::getBinary; + case OBJECT, ROW, UNKNOWN, VECTOR -> + (row, columnIndex) -> { + throw new UnsupportedOperationException( + String.format( + DataNodePipeMessages.UNSUPPORTED_DATA_TYPE_FOR_COLUMN_FMT, + row.getDataType(columnIndex), + row.getColumnName(columnIndex))); + }; + }; + public static final TypeService<TsPrimitiveTabletValueWriter> PIPE_TS_PRIMITIVE_TABLET_VALUE_WRITER_SERVICE = type -> @@ -3492,7 +3518,9 @@ public class TypeServices { OPC_UA_TABLET_OBJECT_VALUE_GETTER_SERVICE.check(); OPC_UA_DATA_TYPE_SERVICE.check(); PIPE_INSERT_EVENT_VALUE_LIST_TYPE_SERVICE.check(); - PIPE_DATA_TYPE_TRANSFORMER_SERVICE.check(); + // PIPE_DATA_TYPE_TRANSFORMER_SERVICE returns a value directly and intentionally rejects + // internal TsFile types, so the generic service check cannot be applied to it. + PIPE_ROW_OBJECT_GETTER_SERVICE.check(); PIPE_TS_PRIMITIVE_TABLET_VALUE_WRITER_SERVICE.check(); PIPE_BATCH_DATA_TABLET_VALUE_WRITER_SERVICE.check(); PIPE_TABLET_VALUE_COLUMN_FILTER_SERVICE.check(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/row/PipeRowTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/row/PipeRowTest.java new file mode 100644 index 00000000000..3699693f1ed --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/row/PipeRowTest.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.iotdb.db.pipe.event.common.row; + +import org.apache.iotdb.pipe.api.type.Binary; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.BytesUtils; +import org.junit.Assert; +import org.junit.Test; + +import java.time.LocalDate; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; + +public class PipeRowTest { + + @Test + public void testGetObjectForSupportedTypes() { + final LocalDate date = LocalDate.of(2026, 8, 12); + final PipeRow row = + new PipeRow( + 0, + "root.test.device", + false, + null, + new long[] {1L}, + new TSDataType[] { + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.DATE, + TSDataType.INT64, + TSDataType.TIMESTAMP, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.TEXT, + TSDataType.BLOB, + TSDataType.STRING + }, + new Object[] { + new boolean[] {true}, + new int[] {2}, + new LocalDate[] {date}, + new long[] {3L}, + new long[] {4L}, + new float[] {5.5F}, + new double[] {6.5D}, + new org.apache.tsfile.utils.Binary[] {BytesUtils.valueOf("text")}, + new org.apache.tsfile.utils.Binary[] {BytesUtils.valueOf("blob")}, + new org.apache.tsfile.utils.Binary[] {BytesUtils.valueOf("string")} + }, + null, + new String[] { + "boolean", + "int32", + "date", + "int64", + "timestamp", + "float", + "double", + "text", + "blob", + "string" + }); + + assertEquals(Boolean.TRUE, row.getObject(0)); + assertEquals(2, row.getObject(1)); + assertEquals(date, row.getObject(2)); + assertEquals(3L, row.getObject(3)); + assertEquals(4L, row.getObject(4)); + assertEquals(5.5F, row.getObject(5)); + assertEquals(6.5D, row.getObject(6)); + assertBinaryEquals("text", row.getObject(7)); + assertBinaryEquals("blob", row.getObject(8)); + assertBinaryEquals("string", row.getObject(9)); + } + + @Test + public void testGetObjectForUnsupportedType() { + final PipeRow row = + new PipeRow( + 0, + "root.test.device", + false, + null, + new long[] {1L}, + new TSDataType[] {TSDataType.OBJECT}, + new Object[] {new Object[] {new Object()}}, + null, + new String[] {"object"}); + + Assert.assertThrows(IllegalArgumentException.class, () -> row.getObject(0)); + } + + private static void assertBinaryEquals(final String expected, final Object actual) { + assertArrayEquals(Binary.stringToBytes(expected), ((Binary) actual).getValues()); + } +}
