You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/12/24 15:00:24 UTC
(iotdb) branch master updated: Pipe: Receiver write table model data type conversion test (#14499)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new d2f95b87e65 Pipe: Receiver write table model data type conversion test (#14499)
d2f95b87e65 is described below
commit d2f95b87e65c1f169aedc3acfac9e86e329328ae
Author: Zhenyu Luo <lu...@gmail.com>
AuthorDate: Tue Dec 24 23:00:18 2024 +0800
Pipe: Receiver write table model data type conversion test (#14499)
---
.../IoTDBPipeTypeConversionISessionIT.java | 497 ++++++++++++++++
.../it/tablemodel/IoTDBPipeTypeConversionIT.java | 632 +++++++++++++++++++++
.../transform/converter/ValueConverter.java | 4 +-
.../statement/PipeConvertedInsertRowStatement.java | 7 +
.../PipeConvertedInsertTabletStatement.java | 7 +
.../relational/sql/ast/WrappedInsertStatement.java | 3 +-
.../plan/statement/crud/InsertBaseStatement.java | 11 +-
7 files changed, 1157 insertions(+), 4 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeTypeConversionISessionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeTypeConversionISessionIT.java
new file mode 100644
index 00000000000..f5ba1f99364
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeTypeConversionISessionIT.java
@@ -0,0 +1,497 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.commons.utils.function.CheckedTriConsumer;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.db.pipe.receiver.transform.converter.ValueConverter;
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.DateUtils;
+import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeTypeConversionISessionIT extends AbstractPipeTableModelTestIT {
+ private static final int generateDataSize = 100;
+
+ @Test
+ public void insertTablet() {
+ prepareTypeConversionTest(
+ (ITableSession senderSession, ITableSession receiverSession, Tablet tablet) -> {
+ senderSession.insert(tablet);
+ },
+ false);
+ }
+
+ @Test
+ public void insertTabletReceiveByTsFile() {
+ prepareTypeConversionTest(
+ (ITableSession senderSession, ITableSession receiverSession, Tablet tablet) -> {
+ senderSession.insert(tablet);
+ },
+ true);
+ }
+
+ private SessionDataSet query(
+ ITableSession session, List<IMeasurementSchema> measurementSchemas, String tableName)
+ throws IoTDBConnectionException, StatementExecutionException {
+ String sql = "select ";
+ StringBuilder param = new StringBuilder();
+ for (IMeasurementSchema schema : measurementSchemas) {
+ param.append(schema.getMeasurementName());
+ param.append(',');
+ }
+ sql = sql + param.substring(0, param.length() - 1);
+ sql = sql + " from " + tableName + " ORDER BY time ASC";
+ session.executeNonQueryStatement("use test");
+ return session.executeQueryStatement(sql);
+ }
+
+ private void prepareTypeConversionTest(
+ CheckedTriConsumer<ITableSession, ITableSession, Tablet, Exception> executeDataWriteOperation,
+ boolean isTsFile) {
+ List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas =
+ generateMeasurementSchemas();
+ Tablet tablet = generateTabletAndMeasurementSchema(measurementSchemas, "test");
+ createDatabaseAndTable(measurementSchemas, true, tablet.getColumnTypes(), senderEnv);
+ createDatabaseAndTable(measurementSchemas, false, tablet.getColumnTypes(), receiverEnv);
+ try (ITableSession senderSession = senderEnv.getTableSessionConnection();
+ ITableSession receiverSession = receiverEnv.getTableSessionConnection()) {
+
+ if (isTsFile) {
+ // Send TsFile data to receiver
+ executeDataWriteOperation.accept(senderSession, receiverSession, tablet);
+ senderSession.executeNonQueryStatement("flush");
+ createDataPipe(true);
+ } else {
+ // Send Tablet data to receiver
+ createDataPipe(false);
+ // The actual implementation logic of inserting data
+ executeDataWriteOperation.accept(senderSession, receiverSession, tablet);
+ senderSession.executeNonQueryStatement("flush");
+ }
+
+ // Verify receiver data
+ long timeoutSeconds = 30;
+ List<List<Object>> expectedValues =
+ generateTabletResultSetForTable(tablet, measurementSchemas);
+ await()
+ .pollInSameThread()
+ .pollDelay(1L, TimeUnit.SECONDS)
+ .pollInterval(1L, TimeUnit.SECONDS)
+ .atMost(timeoutSeconds, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ try {
+ validateResultSet(
+ query(receiverSession, tablet.getSchemas(), tablet.getTableName()),
+ expectedValues,
+ tablet.timestamps);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ });
+ tablet.reset();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ private void createDatabaseAndTable(
+ List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas,
+ boolean isLeft,
+ List<Tablet.ColumnCategory> categories,
+ BaseEnv env) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < measurementSchemas.size(); i++) {
+ final MeasurementSchema measurement =
+ isLeft ? measurementSchemas.get(i).getLeft() : measurementSchemas.get(i).getRight();
+ builder.append(
+ String.format(
+ "%s %s %s,",
+ measurement.getMeasurementName(), measurement.getType(), categories.get(i).name()));
+ }
+ builder.deleteCharAt(builder.length() - 1);
+ String tableCreation =
+ String.format("create table if not exists test (%s)", builder.toString());
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ null,
+ "table",
+ env,
+ Arrays.asList("create database if not exists test", "use test", tableCreation));
+ }
+
+ private void createDataPipe(boolean isTSFile) {
+ String sql =
+ String.format(
+ "create pipe test"
+ + " with source ('source'='iotdb-source','realtime.mode'='%s','realtime.enable'='%s','history.enable'='%s')"
+ + " with processor ('processor'='do-nothing-processor')"
+ + " with sink ('node-urls'='%s:%s','batch.enable'='false','sink.format'='%s')",
+ isTSFile ? "file" : "forced-log",
+ !isTSFile,
+ isTSFile,
+ receiverEnv.getIP(),
+ receiverEnv.getPort(),
+ isTSFile ? "tsfile" : "tablet");
+ TestUtils.tryExecuteNonQueriesWithRetry(senderEnv, Collections.singletonList(sql));
+ }
+
+ private void validateResultSet(
+ SessionDataSet dataSet, List<List<Object>> values, long[] timestamps)
+ throws IoTDBConnectionException, StatementExecutionException {
+ int index = 0;
+ while (dataSet.hasNext()) {
+ RowRecord record = dataSet.next();
+ List<Field> fields = record.getFields();
+
+ assertEquals(record.getTimestamp(), timestamps[index]);
+ List<Object> rowValues = values.get(index++);
+ for (int i = 0; i < fields.size(); i++) {
+ Field field = fields.get(i);
+ switch (field.getDataType()) {
+ case INT64:
+ case TIMESTAMP:
+ assertEquals(field.getLongV(), (long) rowValues.get(i));
+ break;
+ case DATE:
+ assertEquals(field.getDateV(), rowValues.get(i));
+ break;
+ case BLOB:
+ assertEquals(field.getBinaryV(), rowValues.get(i));
+ break;
+ case TEXT:
+ case STRING:
+ assertEquals(field.getStringValue(), rowValues.get(i));
+ break;
+ case INT32:
+ assertEquals(field.getIntV(), (int) rowValues.get(i));
+ break;
+ case DOUBLE:
+ assertEquals(0, Double.compare(field.getDoubleV(), (double) rowValues.get(i)));
+ break;
+ case FLOAT:
+ assertEquals(0, Float.compare(field.getFloatV(), (float) rowValues.get(i)));
+ break;
+ }
+ }
+ }
+ assertEquals(values.size(), index);
+ }
+
+ private void createTestDataForBoolean(Tablet tablet, int j) {
+ Random random = new Random();
+ for (int i = 0; i < generateDataSize; i++) {
+ if (random.nextBoolean()) {
+ tablet.addValue(i, j, random.nextBoolean());
+ }
+ }
+ }
+
+ private void createTestDataForInt32(Tablet tablet, int j) {
+ Random random = new Random();
+ for (int i = 0; i < generateDataSize; i++) {
+ if (random.nextBoolean()) {
+ tablet.addValue(i, j, random.nextInt());
+ }
+ }
+ }
+
+ private void createTestDataForInt64(Tablet tablet, int j) {
+ Random random = new Random();
+ for (int i = 0; i < generateDataSize; i++) {
+ if (random.nextBoolean()) {
+ tablet.addValue(i, j, random.nextLong());
+ }
+ }
+ }
+
+ private void createTestDataForFloat(Tablet tablet, int j) {
+ Random random = new Random();
+ for (int i = 0; i < generateDataSize; i++) {
+ if (random.nextBoolean()) {
+ tablet.addValue(i, j, random.nextFloat());
+ }
+ }
+ }
+
+ private void createTestDataForDouble(Tablet tablet, int j) {
+ Random random = new Random();
+ for (int i = 0; i < generateDataSize; i++) {
+ if (random.nextBoolean()) {
+ tablet.addValue(i, j, random.nextDouble());
+ }
+ }
+ }
+
+ private void createTestDataForTimestamp(Tablet tablet, int j) {
+ Random random = new Random();
+ long time = new Date().getTime();
+ for (int i = 0; i < generateDataSize; i++) {
+ if (random.nextBoolean()) {
+ tablet.addValue(i, j, time);
+ }
+ }
+ }
+
+ private long[] createTestDataForTimestamp() {
+ long time = new Date().getTime();
+ long[] data = new long[generateDataSize];
+ for (int i = 0; i < data.length; i++) {
+ data[i] = time++;
+ }
+ return data;
+ }
+
+ private void createTestDataForDate(Tablet tablet, int j) {
+ int year = 2025;
+ int month = 1;
+ int day = 1;
+ for (int i = 0; i < generateDataSize; i++) {
+ tablet.addValue(i, j, DateUtils.parseIntToLocalDate(year * 10000 + (month * 100) + day));
+ // update
+ day++;
+ if (day > 28) {
+ day = 1;
+ month++;
+ if (month > 12) {
+ month = 1;
+ year++;
+ }
+ }
+ }
+ }
+
+ private void createTestDataForString(Tablet tablet, int j) {
+ String[] stringData = {
+ "Hello",
+ "Hello World!",
+ "This is a test.",
+ "IoTDB Hello World!!!!",
+ "IoTDB is an excellent time series database!!!!!!!!!",
+ "12345678910!!!!!!!!",
+ "123456",
+ "1234567.123213",
+ "21232131.21",
+ "enable = true",
+ "true",
+ "false",
+ };
+ for (int i = 0; i < generateDataSize; i++) {
+ tablet.addValue(
+ i, j, stringData[(i % stringData.length)].getBytes(TSFileConfig.STRING_CHARSET));
+ }
+ }
+
+ private List<List<Object>> generateTabletResultSetForTable(
+ final Tablet tablet, List<Pair<MeasurementSchema, MeasurementSchema>> pairs) {
+ List<List<Object>> insertRecords = new ArrayList<>(tablet.getRowSize());
+ final List<IMeasurementSchema> schemas = tablet.getSchemas();
+ final Object[] values = tablet.values;
+ for (int i = 0; i < tablet.getRowSize(); i++) {
+ List<Object> insertRecord = new ArrayList<>();
+ for (int j = 0; j < schemas.size(); j++) {
+ TSDataType sourceType = pairs.get(j).left.getType();
+ TSDataType targetType = pairs.get(j).right.getType();
+ Object value = null;
+ switch (sourceType) {
+ case INT64:
+ case TIMESTAMP:
+ value = ValueConverter.convert(sourceType, targetType, ((long[]) values[j])[i]);
+ insertRecord.add(convert(value, targetType));
+ break;
+ case INT32:
+ value = ValueConverter.convert(sourceType, targetType, ((int[]) values[j])[i]);
+ insertRecord.add(convert(value, targetType));
+ break;
+ case DOUBLE:
+ value = ValueConverter.convert(sourceType, targetType, ((double[]) values[j])[i]);
+ insertRecord.add(convert(value, targetType));
+ break;
+ case FLOAT:
+ value = ValueConverter.convert(sourceType, targetType, ((float[]) values[j])[i]);
+ insertRecord.add(convert(value, targetType));
+ break;
+ case DATE:
+ value =
+ ValueConverter.convert(
+ sourceType,
+ targetType,
+ DateUtils.parseDateExpressionToInt(((LocalDate[]) values[j])[i]));
+ insertRecord.add(convert(value, targetType));
+ break;
+ case TEXT:
+ case STRING:
+ value = ValueConverter.convert(sourceType, targetType, ((Binary[]) values[j])[i]);
+ insertRecord.add(convert(value, targetType));
+ break;
+ case BLOB:
+ value = ValueConverter.convert(sourceType, targetType, ((Binary[]) values[j])[i]);
+ insertRecord.add(convert(value, targetType));
+ break;
+ case BOOLEAN:
+ value = ValueConverter.convert(sourceType, targetType, ((boolean[]) values[j])[i]);
+ insertRecord.add(convert(value, targetType));
+ break;
+ }
+ }
+ insertRecords.add(insertRecord);
+ }
+
+ return insertRecords;
+ }
+
+ private Object convert(Object value, TSDataType targetType) {
+ switch (targetType) {
+ case DATE:
+ return DateUtils.parseIntToLocalDate((Integer) value);
+ case TEXT:
+ case STRING:
+ return new String(((Binary) value).getValues(), TSFileConfig.STRING_CHARSET);
+ }
+ return value;
+ }
+
+ private Tablet generateTabletAndMeasurementSchema(
+ List<Pair<MeasurementSchema, MeasurementSchema>> pairs, String tableName) {
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ for (Pair<MeasurementSchema, MeasurementSchema> pair : pairs) {
+ schemaList.add(pair.left);
+ }
+
+ final List<Tablet.ColumnCategory> columnTypes = generateTabletColumnCategory(pairs.size());
+ Tablet tablet =
+ new Tablet(
+ tableName,
+ IMeasurementSchema.getMeasurementNameList(schemaList),
+ IMeasurementSchema.getDataTypeList(schemaList),
+ columnTypes,
+ generateDataSize);
+ tablet.initBitMaps();
+ tablet.timestamps = createTestDataForTimestamp();
+ for (int i = 0; i < pairs.size(); i++) {
+ MeasurementSchema schema = pairs.get(i).left;
+ switch (schema.getType()) {
+ case INT64:
+ createTestDataForInt64(tablet, i);
+ break;
+ case INT32:
+ createTestDataForInt32(tablet, i);
+ break;
+ case TIMESTAMP:
+ createTestDataForTimestamp(tablet, i);
+ break;
+ case DOUBLE:
+ createTestDataForDouble(tablet, i);
+ break;
+ case FLOAT:
+ createTestDataForFloat(tablet, i);
+ break;
+ case DATE:
+ createTestDataForDate(tablet, i);
+ break;
+ case STRING:
+ case BLOB:
+ case TEXT:
+ createTestDataForString(tablet, i);
+ break;
+ case BOOLEAN:
+ createTestDataForBoolean(tablet, i);
+ break;
+ }
+ }
+
+ return tablet;
+ }
+
+ private List<Tablet.ColumnCategory> generateTabletColumnCategory(int size) {
+ List<Tablet.ColumnCategory> columnTypes = new ArrayList<>(size);
+ columnTypes.add(Tablet.ColumnCategory.ID);
+ columnTypes.add(Tablet.ColumnCategory.ID);
+ columnTypes.add(Tablet.ColumnCategory.ID);
+ columnTypes.add(Tablet.ColumnCategory.ID);
+ for (int i = 0; i < size - 4; i++) {
+ columnTypes.add(Tablet.ColumnCategory.MEASUREMENT);
+ }
+ return columnTypes;
+ }
+
+ private List<Pair<MeasurementSchema, MeasurementSchema>> generateMeasurementSchemas() {
+ TSDataType[] dataTypes = {
+ TSDataType.STRING,
+ TSDataType.TEXT,
+ TSDataType.BLOB,
+ TSDataType.TIMESTAMP,
+ TSDataType.BOOLEAN,
+ TSDataType.DATE,
+ TSDataType.DOUBLE,
+ TSDataType.FLOAT,
+ TSDataType.INT32,
+ TSDataType.INT64
+ };
+ List<Pair<MeasurementSchema, MeasurementSchema>> pairs = new ArrayList<>();
+
+ for (int t = 0; t < 4; t++) {
+ pairs.add(
+ new Pair<>(
+ new MeasurementSchema("s" + t, TSDataType.STRING),
+ new MeasurementSchema("s" + t, TSDataType.STRING)));
+ }
+
+ for (TSDataType type : dataTypes) {
+ for (TSDataType dataType : dataTypes) {
+ String id = String.format("%s2%s", type.name(), dataType.name());
+ pairs.add(new Pair<>(new MeasurementSchema(id, type), new MeasurementSchema(id, dataType)));
+ }
+ }
+ return pairs;
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeTypeConversionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeTypeConversionIT.java
new file mode 100644
index 00000000000..0420c316a74
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeTypeConversionIT.java
@@ -0,0 +1,632 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.db.pipe.receiver.transform.converter.ValueConverter;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.RpcUtils;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BytesUtils;
+import org.apache.tsfile.utils.DateUtils;
+import org.apache.tsfile.utils.Pair;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.nio.charset.StandardCharsets;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeTypeConversionIT extends AbstractPipeTableModelTestIT {
+
+ private static final int generateDataSize = 100;
+
+ // Test for converting BOOLEAN to OtherType
+ @Test
+ public void testBooleanToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.TIMESTAMP);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.BOOLEAN, TSDataType.DATE);
+ }
+
+ // Test for converting INT32 to OtherType
+ @Test
+ public void testInt32ToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.TIMESTAMP);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.INT32, TSDataType.DATE);
+ }
+
+ // Test for converting INT64 to OtherType
+ @Test
+ public void testInt64ToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.TIMESTAMP);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.INT64, TSDataType.DATE);
+ }
+
+ // Test for converting FLOAT to OtherType
+ @Test
+ public void testFloatToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.TIMESTAMP);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.FLOAT, TSDataType.DATE);
+ }
+
+ // Test for converting DOUBLE to OtherType
+ @Test
+ public void testDoubleToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.TIMESTAMP);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.DOUBLE, TSDataType.DATE);
+ }
+
+ // Test for converting TEXT to OtherType
+ @Test
+ public void testTextToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.TIMESTAMP);
+ executeAndVerifyTypeConversion(TSDataType.TEXT, TSDataType.DATE);
+ }
+
+ // Test for converting TIMESTAMP to OtherType
+ @Test
+ public void testTimestampToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.TIMESTAMP, TSDataType.DATE);
+ }
+
+ // Test for converting DATE to OtherType
+ @Test
+ public void testDateToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.DATE, TSDataType.TIMESTAMP);
+ }
+
+ // Test for converting BLOB to OtherType
+ @Test
+ public void testBlobToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.STRING);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.TIMESTAMP);
+ executeAndVerifyTypeConversion(TSDataType.BLOB, TSDataType.DATE);
+ }
+
+ // Test for converting STRING to OtherType
+ @Test
+ public void testStringToOtherTypeConversion() {
+ createDataPipe();
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.TEXT);
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.BLOB);
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.BOOLEAN);
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.INT32);
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.INT64);
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.FLOAT);
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.DOUBLE);
+ executeAndVerifyTypeConversion(TSDataType.STRING, TSDataType.TIMESTAMP);
+ }
+
+ private void executeAndVerifyTypeConversion(TSDataType source, TSDataType target) {
+ List<Pair> pairs = prepareTypeConversionTest(source, target);
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ String.format("select time,status,s1 from %s2%s", source.name(), target.name()),
+ "time,status,s1,",
+ createExpectedResultSet(pairs, source, target),
+ "test");
+ }
+
+ private List<Pair> prepareTypeConversionTest(TSDataType sourceType, TSDataType targetType) {
+ String sourceTypeName = sourceType.name();
+ String targetTypeName = targetType.name();
+
+ createDataBaseAndTable(sourceTypeName, targetTypeName, sourceTypeName, senderEnv);
+ createDataBaseAndTable(sourceTypeName, targetTypeName, targetTypeName, receiverEnv);
+
+ List<Pair> pairs = createTestDataForType(sourceTypeName);
+
+ executeDataInsertions(pairs, sourceType, targetType);
+ return pairs;
+ }
+
+ private void createDataBaseAndTable(
+ String sourceTypeName, String targetTypeName, String dataType, BaseEnv env) {
+ String timeSeriesCreationQuery =
+ String.format(
+ "create table %s2%s (s1 string id,status %s measurement)",
+ sourceTypeName, targetTypeName, dataType);
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ null,
+ BaseEnv.TABLE_SQL_DIALECT,
+ env,
+ Arrays.asList("create database if not exists test", "use test", timeSeriesCreationQuery));
+ }
+
+ private void createDataPipe() {
+ String sql =
+ String.format(
+ "create pipe test"
+ + " with source ('source'='iotdb-source','realtime.mode'='forced-log','realtime.enable'='true','history.enable'='false')"
+ + " with processor ('processor'='do-nothing-processor')"
+ + " with sink ('node-urls'='%s:%s','batch.enable'='false','sink.format'='tablet')",
+ receiverEnv.getIP(), receiverEnv.getPort());
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ null, BaseEnv.TABLE_SQL_DIALECT, senderEnv, Collections.singletonList(sql));
+ }
+
+ private List<Pair> createTestDataForType(String sourceType) {
+ switch (sourceType) {
+ case "BOOLEAN":
+ return createTestDataForBoolean();
+ case "INT32":
+ return createTestDataForInt32();
+ case "INT64":
+ return createTestDataForInt64();
+ case "FLOAT":
+ return createTestDataForFloat();
+ case "DOUBLE":
+ return createTestDataForDouble();
+ case "TEXT":
+ return createTestDataForText();
+ case "TIMESTAMP":
+ return createTestDataForTimestamp();
+ case "DATE":
+ return createTestDataForDate();
+ case "BLOB":
+ return createTestDataForBlob();
+ case "STRING":
+ return createTestDataForString();
+ default:
+ throw new UnsupportedOperationException("Unsupported data type: " + sourceType);
+ }
+ }
+
+ private void executeDataInsertions(
+ List<Pair> testData, TSDataType sourceType, TSDataType targetType) {
+ switch (sourceType) {
+ case STRING:
+ case TEXT:
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ "test",
+ BaseEnv.TABLE_SQL_DIALECT,
+ senderEnv,
+ createInsertStatementsForString(testData, sourceType.name(), targetType.name()));
+ return;
+ case TIMESTAMP:
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ "test",
+ BaseEnv.TABLE_SQL_DIALECT,
+ senderEnv,
+ createInsertStatementsForTimestamp(testData, sourceType.name(), targetType.name()));
+ return;
+ case DATE:
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ "test",
+ BaseEnv.TABLE_SQL_DIALECT,
+ senderEnv,
+ createInsertStatementsForLocalDate(testData, sourceType.name(), targetType.name()));
+ return;
+ case BLOB:
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ "test",
+ BaseEnv.TABLE_SQL_DIALECT,
+ senderEnv,
+ createInsertStatementsForBlob(testData, sourceType.name(), targetType.name()));
+ return;
+ default:
+ TestUtils.tryExecuteNonQueriesWithRetry(
+ "test",
+ BaseEnv.TABLE_SQL_DIALECT,
+ senderEnv,
+ createInsertStatementsForNumeric(testData, sourceType.name(), targetType.name()));
+ }
+ }
+
+ private List<String> createInsertStatementsForString(
+ List<Pair> testData, String sourceType, String targetType) {
+ List<String> executes = new ArrayList<>();
+ for (Pair pair : testData) {
+ executes.add(
+ String.format(
+ "insert into %s2%s(time,status,s1) values (%s,'%s','t1')",
+ sourceType,
+ targetType,
+ pair.left,
+ new String(((Binary) (pair.right)).getValues(), StandardCharsets.UTF_8)));
+ }
+ executes.add("flush");
+ return executes;
+ }
+
+ private List<String> createInsertStatementsForNumeric(
+ List<Pair> testData, String sourceType, String targetType) {
+ List<String> executes = new ArrayList<>();
+ for (Pair pair : testData) {
+ executes.add(
+ String.format(
+ "insert into %s2%s (time,status,s1) values (%s,%s,'t1')",
+ sourceType, targetType, pair.left, pair.right));
+ }
+ executes.add("flush");
+ return executes;
+ }
+
+ private List<String> createInsertStatementsForTimestamp(
+ List<Pair> testData, String sourceType, String targetType) {
+ List<String> executes = new ArrayList<>();
+ for (Pair pair : testData) {
+ executes.add(
+ String.format(
+ "insert into %s2%s(time,status,s1) values (%s,%s,'t1')",
+ sourceType, targetType, pair.left, pair.right));
+ }
+ executes.add("flush");
+ return executes;
+ }
+
+ private List<String> createInsertStatementsForLocalDate(
+ List<Pair> testData, String sourceType, String targetType) {
+ List<String> executes = new ArrayList<>();
+ for (Pair pair : testData) {
+ executes.add(
+ String.format(
+ "insert into %s2%s(time,status,s1) values (%s,'%s','t1')",
+ sourceType, targetType, pair.left, DateUtils.formatDate((Integer) pair.right)));
+ }
+ executes.add("flush");
+ return executes;
+ }
+
+ private List<String> createInsertStatementsForBlob(
+ List<Pair> testData, String sourceType, String targetType) {
+ List<String> executes = new ArrayList<>();
+ for (Pair pair : testData) {
+ String value = BytesUtils.parseBlobByteArrayToString(((Binary) pair.right).getValues());
+ executes.add(
+ String.format(
+ "insert into %s2%s(time,status,s1) values (%s,X'%s','t1')",
+ sourceType, targetType, pair.left, value.substring(2)));
+ }
+ executes.add("flush");
+ return executes;
+ }
+
+ private Set<String> createExpectedResultSet(
+ List<Pair> pairs, TSDataType sourceType, TSDataType targetType) {
+ switch (targetType) {
+ case TIMESTAMP:
+ return generateTimestampResultSet(pairs, sourceType, targetType);
+ case DATE:
+ return generateLocalDateResultSet(pairs, sourceType, targetType);
+ case BLOB:
+ return generateBlobResultSet(pairs, sourceType, targetType);
+ case TEXT:
+ case STRING:
+ return generateStringResultSet(pairs, sourceType, targetType);
+ default:
+ HashSet<String> resultSet = new HashSet<>();
+ for (Pair pair : pairs) {
+ resultSet.add(
+ String.format(
+ "%s,%s,t1,",
+ toTimestamp((long) pair.left),
+ ValueConverter.convert(sourceType, targetType, pair.right)));
+ }
+ return resultSet;
+ }
+ }
+
+ private Set<String> generateTimestampResultSet(
+ List<Pair> pairs, TSDataType sourceType, TSDataType targetType) {
+ HashSet<String> resultSet = new HashSet<>();
+ for (Pair pair : pairs) {
+ resultSet.add(
+ String.format(
+ "%s,%s,t1,",
+ toTimestamp((long) pair.left),
+ toTimestamp((long) ValueConverter.convert(sourceType, targetType, pair.right))));
+ }
+ return resultSet;
+ }
+
+ private Set<String> generateLocalDateResultSet(
+ List<Pair> pairs, TSDataType sourceType, TSDataType targetType) {
+ HashSet<String> resultSet = new HashSet<>();
+ for (Pair pair : pairs) {
+ resultSet.add(
+ String.format(
+ "%s,%s,t1,",
+ toTimestamp((long) pair.left),
+ DateUtils.formatDate(
+ (Integer) ValueConverter.convert(sourceType, targetType, pair.right))));
+ }
+ return resultSet;
+ }
+
+ private Set<String> generateBlobResultSet(
+ List<Pair> pairs, TSDataType sourceType, TSDataType targetType) {
+ HashSet<String> resultSet = new HashSet<>();
+ for (Pair pair : pairs) {
+ resultSet.add(
+ String.format(
+ "%s,%s,t1,",
+ toTimestamp((long) pair.left),
+ BytesUtils.parseBlobByteArrayToString(
+ ((Binary) ValueConverter.convert(sourceType, targetType, pair.right))
+ .getValues())));
+ }
+ return resultSet;
+ }
+
+ private Set<String> generateStringResultSet(
+ List<Pair> pairs, TSDataType sourceType, TSDataType targetType) {
+ HashSet<String> resultSet = new HashSet<>();
+ for (Pair pair : pairs) {
+ resultSet.add(
+ String.format(
+ "%s,%s,t1,",
+ toTimestamp((long) pair.left),
+ new String(
+ ((Binary) ValueConverter.convert(sourceType, targetType, pair.right)).getValues(),
+ StandardCharsets.UTF_8)));
+ }
+ return resultSet;
+ }
+
+ private List<Pair> createTestDataForBoolean() {
+ List<Pair> pairs = new ArrayList<>();
+ Random random = new Random();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, random.nextBoolean()));
+ }
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForInt32() {
+ List<Pair> pairs = new ArrayList<>();
+ Random random = new Random();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, random.nextInt()));
+ }
+ pairs.add(new Pair<>(generateDataSize + 1L, -1));
+ pairs.add(new Pair<>(generateDataSize + 2L, -2));
+ pairs.add(new Pair<>(generateDataSize + 3L, -3));
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForInt64() {
+ List<Pair> pairs = new ArrayList<>();
+ Random random = new Random();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, random.nextLong()));
+ }
+ pairs.add(new Pair<>(generateDataSize + 1L, -1L));
+ pairs.add(new Pair<>(generateDataSize + 2L, -2L));
+ pairs.add(new Pair<>(generateDataSize + 3L, -3L));
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForFloat() {
+ List<Pair> pairs = new ArrayList<>();
+ Random random = new Random();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, random.nextFloat()));
+ }
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForDouble() {
+ List<Pair> pairs = new ArrayList<>();
+ Random random = new Random();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, random.nextDouble()));
+ }
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForText() {
+ List<Pair> pairs = new ArrayList<>();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, new Binary((String.valueOf(i)).getBytes(StandardCharsets.UTF_8))));
+ }
+ pairs.add(
+ new Pair(generateDataSize + 1L, new Binary("Hello".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 2L, new Binary("Hello World!".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 3L, new Binary("This is a test.".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 4L,
+ new Binary("IoTDB Hello World!!!!".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 5L,
+ new Binary(
+ "IoTDB is an excellent time series database!!!!!!!!!"
+ .getBytes(StandardCharsets.UTF_8))));
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForTimestamp() {
+ List<Pair> pairs = new ArrayList<>();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, new Date().getTime() + i));
+ }
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForDate() {
+ List<Pair> pairs = new ArrayList<>();
+ int year = 2023;
+ int month = 1;
+ int day = 1;
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, year * 10000 + (month * 100) + day));
+
+ // update
+ day++;
+ if (day > 28) {
+ day = 1;
+ month++;
+ if (month > 12) {
+ month = 1;
+ year++;
+ }
+ }
+ }
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForBlob() {
+ List<Pair> pairs = new ArrayList<>();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, new Binary((String.valueOf(i)).getBytes(StandardCharsets.UTF_8))));
+ }
+ pairs.add(
+ new Pair(generateDataSize + 1L, new Binary("Hello".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 2L, new Binary("Hello World!".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 3L, new Binary("This is a test.".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 4L,
+ new Binary("IoTDB Hello World!!!!".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 5L,
+ new Binary(
+ "IoTDB is an excellent time series database!!!!!!!!!"
+ .getBytes(StandardCharsets.UTF_8))));
+ return pairs;
+ }
+
+ private List<Pair> createTestDataForString() {
+ List<Pair> pairs = new ArrayList<>();
+ for (long i = 0; i < generateDataSize; i++) {
+ pairs.add(new Pair<>(i, new Binary((String.valueOf(i)).getBytes(StandardCharsets.UTF_8))));
+ }
+ pairs.add(
+ new Pair(generateDataSize + 1L, new Binary("Hello".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 2L, new Binary("Hello World!".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 3L, new Binary("This is a test.".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 4L,
+ new Binary("IoTDB Hello World!!!!".getBytes(StandardCharsets.UTF_8))));
+ pairs.add(
+ new Pair(
+ generateDataSize + 5L,
+ new Binary(
+ "IoTDB is an excellent time series database!!!!!!!!!"
+ .getBytes(StandardCharsets.UTF_8))));
+ return pairs;
+ }
+
+ private String toTimestamp(long timestamp) {
+ return RpcUtils.formatDatetime("default", "ms", timestamp, ZoneOffset.UTC);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/converter/ValueConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/converter/ValueConverter.java
index 09e2f265ec5..4947a948f37 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/converter/ValueConverter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/converter/ValueConverter.java
@@ -551,7 +551,7 @@ public class ValueConverter {
}
public static Binary convertTextToBlob(final Binary value) {
- return parseBlob(value.toString());
+ return value;
}
public static Binary convertTextToString(final Binary value) {
@@ -715,7 +715,7 @@ public class ValueConverter {
}
public static Binary convertStringToBlob(final Binary value) {
- return parseBlob(value.toString());
+ return value;
}
public static Binary convertStringToText(final Binary value) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java
index 390de44ab01..f249a773455 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.pipe.receiver.transform.converter.ValueConverter;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
+import org.apache.tsfile.annotations.TableModel;
import org.apache.tsfile.enums.TSDataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -116,4 +117,10 @@ public class PipeConvertedInsertRowStatement extends InsertRowStatement {
isNeedInferType = false;
}
+
+ @TableModel
+ @Override
+ public boolean isForceTypeConversion() {
+ return true;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java
index 0abe0f68df2..606309366cc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.receiver.transform.statement;
import org.apache.iotdb.db.pipe.receiver.transform.converter.ArrayConverter;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
+import org.apache.tsfile.annotations.TableModel;
import org.apache.tsfile.enums.TSDataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,4 +70,10 @@ public class PipeConvertedInsertTabletStatement extends InsertTabletStatement {
dataTypes[columnIndex] = dataType;
return true;
}
+
+ @TableModel
+ @Override
+ public boolean isForceTypeConversion() {
+ return true;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/WrappedInsertStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/WrappedInsertStatement.java
index 654c03a5220..60e3dddfa08 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/WrappedInsertStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/WrappedInsertStatement.java
@@ -187,7 +187,8 @@ public abstract class WrappedInsertStatement extends WrappedStatement
// the type is inferred and can be inconsistent with the existing one
innerTreeStatement.setDataType(InternalTypeManager.getTSDataType(real.getType()), i);
} else if (!InternalTypeManager.getTSDataType(real.getType())
- .isCompatible(InternalTypeManager.getTSDataType(incoming.getType()))) {
+ .isCompatible(InternalTypeManager.getTSDataType(incoming.getType()))
+ && !innerTreeStatement.isForceTypeConversion()) {
SemanticException semanticException =
new SemanticException(
String.format(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java
index 2064c2309a2..c9725f26f9e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertBaseStatement.java
@@ -585,7 +585,10 @@ public abstract class InsertBaseStatement extends Statement {
}
if (measurementSchemas != null) {
for (MeasurementSchema measurementSchema : measurementSchemas) {
- measurementSchema.setMeasurementName(measurementSchema.getMeasurementName().toLowerCase());
+ if (measurementSchema != null) {
+ measurementSchema.setMeasurementName(
+ measurementSchema.getMeasurementName().toLowerCase());
+ }
}
}
}
@@ -605,4 +608,10 @@ public abstract class InsertBaseStatement extends Statement {
public String getTableName() {
return devicePath.getFullPath();
}
+
+ // Only Pipe will set the return value to true
+ @TableModel
+ public boolean isForceTypeConversion() {
+ return false;
+ }
}