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 2023/09/15 07:48:57 UTC
[iotdb] branch rel/1.2 updated: [To rel/1.2] Fix some minor bugs of flink-sql-iotdb-connector (#11143) (#11144)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new b1b4aa8b3ff [To rel/1.2] Fix some minor bugs of flink-sql-iotdb-connector (#11143) (#11144)
b1b4aa8b3ff is described below
commit b1b4aa8b3ff3852ef8c4c3ba5d8969e48b73af6a
Author: Xuan Ronaldo <xu...@qq.com>
AuthorDate: Fri Sep 15 15:48:50 2023 +0800
[To rel/1.2] Fix some minor bugs of flink-sql-iotdb-connector (#11143) (#11144)
---
.../org/apache/iotdb/flink/sql/common/Utils.java | 11 ++++++++++-
.../flink/sql/factory/IoTDBDynamicTableFactory.java | 20 ++++++++++++++++++++
.../flink/sql/function/IoTDBBoundedScanFunction.java | 17 ++++++++++++++++-
.../flink/sql/function/IoTDBLookupFunction.java | 18 ++++++++++++++++--
4 files changed, 62 insertions(+), 4 deletions(-)
diff --git a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/common/Utils.java b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/common/Utils.java
index e54f677938f..031dbd1ed99 100644
--- a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/common/Utils.java
+++ b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/common/Utils.java
@@ -32,6 +32,8 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.types.DataType;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.lang.reflect.Array;
import java.net.Socket;
@@ -83,6 +85,9 @@ public class Utils {
}
public static Object getValue(Field value, DataType dataType) {
+ if (value.getDataType() == null) {
+ return null;
+ }
if (dataType.equals(DataTypes.INT())) {
return value.getIntV();
} else if (dataType.equals(DataTypes.BIGINT())) {
@@ -132,6 +137,10 @@ public class Utils {
values.add(rowRecord.getTimestamp());
List<Field> fields = rowRecord.getFields();
for (Tuple2<String, DataType> field : tableSchema) {
+ if (!columnNames.contains(field.f0)) {
+ values.add(null);
+ continue;
+ }
values.add(getValue(fields.get(columnNames.indexOf(field.f0) - 1), field.f1));
}
return GenericRowData.of(values.toArray());
@@ -159,7 +168,7 @@ public class Utils {
}
}
- public static boolean isTypeEqual(TSDataType iotdbType, DataType flinkType) {
+ public static boolean isTypeEqual(@Nullable TSDataType iotdbType, DataType flinkType) {
return typeMap.get(iotdbType).equals(flinkType);
}
}
diff --git a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/factory/IoTDBDynamicTableFactory.java b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/factory/IoTDBDynamicTableFactory.java
index 46fa223dda1..5ad27778f18 100644
--- a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/factory/IoTDBDynamicTableFactory.java
+++ b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/factory/IoTDBDynamicTableFactory.java
@@ -189,6 +189,26 @@ public class IoTDBDynamicTableFactory
throw new IllegalOptionException(
"The option `sql` is required when option `mode` equals `BOUNDED`");
}
+ String sqlLower = options.get(Options.SQL).toLowerCase();
+ if (!sqlLower.contains("select")
+ && (sqlLower.contains("count(")
+ || sqlLower.contains("sum(")
+ || sqlLower.contains("avg(")
+ || sqlLower.contains("extreme(")
+ || sqlLower.contains("max_value(")
+ || sqlLower.contains("min_value(")
+ || sqlLower.contains("first_value(")
+ || sqlLower.contains("last_value(")
+ || sqlLower.contains("max_time(")
+ || sqlLower.contains("min_time(")
+ || sqlLower.contains("group")
+ || sqlLower.contains("where")
+ || sqlLower.contains("create")
+ || sqlLower.contains("count")
+ || sqlLower.contains("delete")
+ || sqlLower.contains("show"))) {
+ throw new IllegalOptionException("The option `sql` only supports basic query statements.");
+ }
}
}
diff --git a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBBoundedScanFunction.java b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBBoundedScanFunction.java
index f31578df150..d8a49e7e776 100644
--- a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBBoundedScanFunction.java
+++ b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBBoundedScanFunction.java
@@ -20,11 +20,13 @@ package org.apache.iotdb.flink.sql.function;
import org.apache.iotdb.flink.sql.common.Options;
import org.apache.iotdb.flink.sql.common.Utils;
+import org.apache.iotdb.flink.sql.exception.IllegalSchemaException;
import org.apache.iotdb.flink.sql.wrapper.SchemaWrapper;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.flink.api.common.io.DefaultInputSplitAssigner;
@@ -110,8 +112,21 @@ public class IoTDBBoundedScanFunction extends RichInputFormat<RowData, InputSpli
}
try {
dataSet = session.executeQueryStatement(sql);
-
columnNames = dataSet.getColumnNames();
+
+ for (Tuple2<String, DataType> field : tableSchema) {
+ if (!columnNames.contains(field.f0)) {
+ continue;
+ }
+ int index = columnNames.indexOf(field.f0);
+ TSDataType iotdbType = TSDataType.valueOf(dataSet.getColumnTypes().get(index));
+ DataType flinkType = field.f1;
+ if (!Utils.isTypeEqual(iotdbType, flinkType)) {
+ throw new IllegalSchemaException(
+ String.format(
+ "The data type of column `%s` is different in IoTDB and Flink", field.f0));
+ }
+ }
} catch (StatementExecutionException | IoTDBConnectionException e) {
throw new RuntimeException(e);
}
diff --git a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java
index f9018dfaf83..0685e770a23 100644
--- a/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java
+++ b/iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java
@@ -20,11 +20,13 @@ package org.apache.iotdb.flink.sql.function;
import org.apache.iotdb.flink.sql.common.Options;
import org.apache.iotdb.flink.sql.common.Utils;
+import org.apache.iotdb.flink.sql.exception.IllegalSchemaException;
import org.apache.iotdb.flink.sql.wrapper.SchemaWrapper;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
@@ -122,8 +124,20 @@ public class IoTDBLookupFunction extends TableFunction<RowData> {
ArrayList<Object> values = new ArrayList<>();
values.add(timestamp);
- for (Tuple2<String, DataType> filed : schema) {
- values.add(Utils.getValue(fields.get(columnNames.indexOf(filed.f0)), filed.f1));
+ for (Tuple2<String, DataType> field : schema) {
+ if (!columnNames.contains(field.f0)) {
+ values.add(null);
+ continue;
+ }
+ int index = columnNames.indexOf(field.f0);
+ DataType flinkType = field.f1;
+ TSDataType iotdbType = fields.get(index).getDataType();
+ if (!Utils.isTypeEqual(iotdbType, flinkType)) {
+ throw new IllegalSchemaException(
+ String.format(
+ "The data type of column `%s` is different in IoTDB and Flink", field.f0));
+ }
+ values.add(Utils.getValue(fields.get(index), field.f1));
}
GenericRowData rowData = GenericRowData.of(values.toArray());