This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch flink-iotdb-table-connector
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
The following commit(s) were added to refs/heads/flink-iotdb-table-connector by
this push:
new b9c61c1 refactor: unify option keys under iotdb. prefix
b9c61c1 is described below
commit b9c61c1ae4d5586b084f0335c1c46d99ec2ccc43
Author: shuwenwei <[email protected]>
AuthorDate: Wed Sep 23 15:54:36 2026 +0800
refactor: unify option keys under iotdb. prefix
Consolidate literal/aggregate pushdown helpers into IoTDBUtils and update
option keys in code, tests and messages.
---
.../iotdb/relational/flink/cfg/IoTDBOptions.java | 40 +--
.../source/cdc/client/IoTDBSubscriptionClient.java | 2 +-
.../flink/source/lookup/IoTDBLookupReader.java | 3 +-
.../source/lookup/IoTDBRuntimeLiteralUtils.java | 108 -------
.../scan/pushdown/IoTDBAggregatePushDownUtils.java | 192 ------------
.../scan/pushdown/IoTDBExpressionVisitor.java | 2 +-
.../source/scan/pushdown/IoTDBLiteralUtils.java | 115 -------
.../iotdb/relational/flink/utils/IoTDBUtils.java | 330 ++++++++++++++++++++-
.../relational/flink/catalog/IoTDBCatalog.java | 6 +-
.../relational/flink/sink/IoTDBSinkWriter.java | 4 +-
.../table/IoTDBRelationalDynamicTableSource.java | 11 +-
.../flink/IoTDBRelationalLocalQueryManual.java | 16 +-
.../IoTDBRelationalSourcePushDownPlannerTest.java | 14 +-
13 files changed, 371 insertions(+), 472 deletions(-)
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
index 7ab12c6..1463679 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/cfg/IoTDBOptions.java
@@ -39,58 +39,60 @@ public class IoTDBOptions implements Serializable {
public static final String IDENTIFIER = "iotdb-relational";
public static final ConfigOption<String> NODE_URLS =
- ConfigOptions.key("nodeUrls").stringType().noDefaultValue();
+ ConfigOptions.key("iotdb.node-urls").stringType().noDefaultValue();
public static final ConfigOption<String> USER =
- ConfigOptions.key("user").stringType().defaultValue("root");
+ ConfigOptions.key("iotdb.user").stringType().defaultValue("root");
public static final ConfigOption<String> PASSWORD =
- ConfigOptions.key("password").stringType().defaultValue("root");
+ ConfigOptions.key("iotdb.password").stringType().defaultValue("root");
public static final ConfigOption<String> DATABASE =
- ConfigOptions.key("database").stringType().noDefaultValue();
+ ConfigOptions.key("iotdb.database").stringType().noDefaultValue();
public static final ConfigOption<String> TABLE =
- ConfigOptions.key("table").stringType().noDefaultValue();
+ ConfigOptions.key("iotdb.table").stringType().noDefaultValue();
public static final ConfigOption<String> DEFAULT_DATABASE =
-
ConfigOptions.key("default-database").stringType().defaultValue("public");
+
ConfigOptions.key("iotdb.default-database").stringType().defaultValue("public");
public static final ConfigOption<String> TIME_COLUMN =
- ConfigOptions.key("time-column").stringType().noDefaultValue();
+ ConfigOptions.key("iotdb.time-column").stringType().noDefaultValue();
public static final ConfigOption<String> TAG_COLUMNS =
- ConfigOptions.key("tag-columns").stringType().defaultValue("");
+ ConfigOptions.key("iotdb.tag-columns").stringType().defaultValue("");
public static final ConfigOption<String> ATTRIBUTE_COLUMNS =
- ConfigOptions.key("attribute-columns").stringType().defaultValue("");
+
ConfigOptions.key("iotdb.attribute-columns").stringType().defaultValue("");
public static final ConfigOption<Boolean> LOOKUP_ASYNC =
- ConfigOptions.key("lookup.async").booleanType().defaultValue(false);
+
ConfigOptions.key("iotdb.lookup.async").booleanType().defaultValue(false);
public static final ConfigOption<Integer> LOOKUP_THREAD_SIZE =
- ConfigOptions.key("lookup.thread-size").intType().defaultValue(5);
+ ConfigOptions.key("iotdb.lookup.thread-size").intType().defaultValue(5);
public static final ConfigOption<ScanMode> SCAN_MODE =
-
ConfigOptions.key("scan.mode").enumType(ScanMode.class).defaultValue(ScanMode.SNAPSHOT);
+ ConfigOptions.key("iotdb.scan.mode")
+ .enumType(ScanMode.class)
+ .defaultValue(ScanMode.SNAPSHOT);
public static final ConfigOption<String> CDC_TOPIC =
- ConfigOptions.key("cdc.topic").stringType().noDefaultValue();
+ ConfigOptions.key("iotdb.cdc.topic").stringType().noDefaultValue();
public static final ConfigOption<String> CDC_CONSUMER_GROUP =
- ConfigOptions.key("cdc.consumer-group").stringType().noDefaultValue();
+
ConfigOptions.key("iotdb.cdc.consumer-group").stringType().noDefaultValue();
public static final ConfigOption<String> CDC_MODE =
- ConfigOptions.key("cdc.mode").stringType().defaultValue("live");
+ ConfigOptions.key("iotdb.cdc.mode").stringType().defaultValue("live");
public static final ConfigOption<String> CDC_START_TIME =
- ConfigOptions.key("cdc.start-time").stringType().noDefaultValue();
+ ConfigOptions.key("iotdb.cdc.start-time").stringType().noDefaultValue();
public static final ConfigOption<Long> CDC_POLL_TIMEOUT_MS =
- ConfigOptions.key("cdc.poll-timeout-ms").longType().defaultValue(1000L);
+
ConfigOptions.key("iotdb.cdc.poll-timeout-ms").longType().defaultValue(1000L);
public static final ConfigOption<Boolean> CDC_AUTO_COMMIT =
- ConfigOptions.key("cdc.auto-commit").booleanType().defaultValue(true);
+
ConfigOptions.key("iotdb.cdc.auto-commit").booleanType().defaultValue(true);
private static final String CDC_TOPIC_PREFIX = "flink_iotdb_table_";
@@ -183,7 +185,7 @@ public class IoTDBOptions implements Serializable {
}
/**
- * @return the configured IoTDB time-column name, or {@code null} when
unspecified.
+ * @return the configured IoTDB {@code iotdb.time-column} name, or {@code
null} when unspecified.
*/
public String getTimeColumn() {
return timeColumn;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/client/IoTDBSubscriptionClient.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/client/IoTDBSubscriptionClient.java
index 8df5242..e76b8ed 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/client/IoTDBSubscriptionClient.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/client/IoTDBSubscriptionClient.java
@@ -38,7 +38,7 @@ public final class IoTDBSubscriptionClient {
public static void createTopicIfNotExists(IoTDBOptions options) throws
Exception {
List<String> nodeUrls = options.getNodeUrls();
if (nodeUrls == null || nodeUrls.isEmpty()) {
- throw new IllegalArgumentException("IoTDB nodeUrls must not be empty for
CDC.");
+ throw new IllegalArgumentException("IoTDB 'iotdb.node-urls' must not be
empty for CDC.");
}
String[] hostPort = splitNodeUrl(nodeUrls.get(0));
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
index eb973c4..c92e6c6 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
@@ -117,8 +117,7 @@ public class IoTDBLookupReader implements AutoCloseable {
List<String> predicates = new ArrayList<>(keyIndices.length);
for (int position = 0; position < keyIndices.length; position++) {
int fieldIndex = keyIndices[position];
- String literal =
- IoTDBRuntimeLiteralUtils.render(keyRow, position,
fieldTypes.get(fieldIndex));
+ String literal = IoTDBUtils.renderLiteral(keyRow, position,
fieldTypes.get(fieldIndex));
if (literal == null) {
return null;
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBRuntimeLiteralUtils.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBRuntimeLiteralUtils.java
deleted file mode 100644
index 9993c91..0000000
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBRuntimeLiteralUtils.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * 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.relational.flink.source.lookup;
-
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.logical.LogicalTypeRoot;
-import org.apache.flink.table.types.logical.TimestampType;
-
-import java.time.LocalDate;
-
-/**
- * Renders a runtime {@link RowData} field as an IoTDB SQL literal, used to
build lookup queries.
- *
- * <p>Returns {@code null} when the value is {@code null} or cannot be
represented in IoTDB SQL.
- */
-public final class IoTDBRuntimeLiteralUtils {
-
- private IoTDBRuntimeLiteralUtils() {}
-
- public static String render(RowData row, int position, DataType dataType) {
- if (row == null || row.isNullAt(position) || dataType == null) {
- return null;
- }
-
- LogicalTypeRoot typeRoot = dataType.getLogicalType().getTypeRoot();
- switch (typeRoot) {
- case BOOLEAN:
- return Boolean.toString(row.getBoolean(position));
- case TINYINT:
- return Byte.toString(row.getByte(position));
- case SMALLINT:
- return Short.toString(row.getShort(position));
- case INTEGER:
- return Integer.toString(row.getInt(position));
- case BIGINT:
- return Long.toString(row.getLong(position));
- case FLOAT:
- float floatValue = row.getFloat(position);
- return isFinite(floatValue) ? Float.toString(floatValue) : null;
- case DOUBLE:
- double doubleValue = row.getDouble(position);
- return isFinite(doubleValue) ? Double.toString(doubleValue) : null;
- case CHAR:
- case VARCHAR:
- return quoteString(row.getString(position).toString());
- case BINARY:
- case VARBINARY:
- byte[] bytes = row.getBinary(position);
- return bytes == null ? null : formatBinary(bytes);
- case DATE:
- return "CAST('" + LocalDate.ofEpochDay(row.getInt(position)) + "' AS
DATE)";
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- return "CAST('"
- + getTimestamp(row, position, dataType).toLocalDateTime()
- + "' AS TIMESTAMP)";
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return "CAST('" + getTimestamp(row, position, dataType).toInstant() +
"' AS TIMESTAMP)";
- default:
- return null;
- }
- }
-
- private static TimestampData getTimestamp(RowData row, int position,
DataType dataType) {
- int precision = ((TimestampType) dataType.getLogicalType()).getPrecision();
- return row.getTimestamp(position, precision);
- }
-
- private static boolean isFinite(float value) {
- return !Float.isNaN(value) && !Float.isInfinite(value);
- }
-
- private static boolean isFinite(double value) {
- return !Double.isNaN(value) && !Double.isInfinite(value);
- }
-
- private static String quoteString(String value) {
- return "'" + value.replace("'", "''") + "'";
- }
-
- private static String formatBinary(byte[] value) {
- StringBuilder builder = new StringBuilder(value.length * 2 + 3);
- builder.append("X'");
- for (byte b : value) {
- builder.append(String.format("%02X", b));
- }
- builder.append("'");
- return builder.toString();
- }
-}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBAggregatePushDownUtils.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBAggregatePushDownUtils.java
deleted file mode 100644
index 3e74d95..0000000
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBAggregatePushDownUtils.java
+++ /dev/null
@@ -1,192 +0,0 @@
-/*
- * 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.relational.flink.source.scan.pushdown;
-
-import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
-
-import org.apache.flink.table.expressions.AggregateExpression;
-import org.apache.flink.table.expressions.FieldReferenceExpression;
-import org.apache.flink.table.types.DataType;
-import org.apache.tsfile.enums.TSDataType;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Translates Flink aggregate pushdown information into a serializable {@link
AggregateSpec}.
- *
- * <p>The pushdown is all-or-nothing: any unsupported aggregate, argument or
grouping makes this
- * return {@code null} so the whole aggregation stays in Flink.
- *
- * <p>Each aggregate is classified and translated in a single pass from its
function class name;
- * grouping columns and aggregate arguments are rendered through {@link
IoTDBExpressionVisitor}.
- */
-public final class IoTDBAggregatePushDownUtils {
-
- private IoTDBAggregatePushDownUtils() {}
-
- public static AggregateSpec translate(
- List<int[]> groupingSets,
- List<AggregateExpression> aggregateExpressions,
- DataType sourceRowDataType,
- DataType producedDataType) {
- if (groupingSets == null
- || groupingSets.size() != 1
- || aggregateExpressions == null
- || aggregateExpressions.isEmpty()) {
- return null;
- }
-
- final List<String> sourceFieldNames;
- final List<DataType> producedFieldTypes;
- try {
- sourceFieldNames = DataType.getFieldNames(sourceRowDataType);
- producedFieldTypes = DataType.getFieldDataTypes(producedDataType);
- } catch (RuntimeException e) {
- return null;
- }
-
- int[] grouping = groupingSets.get(0);
- if (grouping == null) {
- return null;
- }
-
- List<String> groupByExpressions = new ArrayList<>(grouping.length);
- List<String> selectExpressions = new ArrayList<>(grouping.length +
aggregateExpressions.size());
- for (int index : grouping) {
- if (index < 0 || index >= sourceFieldNames.size()) {
- return null;
- }
- String column = IoTDBUtils.quoteIdentifier(sourceFieldNames.get(index));
- groupByExpressions.add(column);
- selectExpressions.add(column);
- }
-
- IoTDBExpressionVisitor visitor = new IoTDBExpressionVisitor();
- for (int i = 0; i < aggregateExpressions.size(); i++) {
- int producedIndex = grouping.length + i;
- if (producedIndex >= producedFieldTypes.size()) {
- return null;
- }
- String sql =
- translateAggregate(
- aggregateExpressions.get(i),
producedFieldTypes.get(producedIndex), visitor);
- if (sql == null) {
- return null;
- }
- selectExpressions.add(sql);
- }
-
- return new AggregateSpec(selectExpressions, groupByExpressions);
- }
-
- private static String translateAggregate(
- AggregateExpression aggregate, DataType producedType,
IoTDBExpressionVisitor visitor) {
- if (!isSupportedAggregate(aggregate)) {
- return null;
- }
-
- Class<?> functionClass = aggregate.getFunctionDefinition().getClass();
- String simpleName = functionClass == null ? null :
functionClass.getSimpleName();
- TSDataType outType = toTsDataType(producedType);
- if (simpleName == null || outType == null) {
- return null;
- }
-
- List<FieldReferenceExpression> args = aggregate.getArgs();
- int argCount = args == null ? 0 : args.size();
-
- String expression;
- if (simpleName.endsWith("Count1AggFunction")) {
- if (argCount != 0 || !isCountType(outType)) {
- return null;
- }
- expression = "COUNT(*)";
- } else if (simpleName.endsWith("CountAggFunction")) {
- if (argCount != 1 || !isCountType(outType)) {
- return null;
- }
- String argSql = args.get(0).accept(visitor);
- if (argSql == null || toTsDataType(args.get(0).getOutputDataType()) ==
null) {
- return null;
- }
- expression = "COUNT(" + argSql + ")";
- } else if (simpleName.endsWith("Sum0AggFunction") ||
simpleName.endsWith("SumAggFunction")) {
- if (argCount != 1 || !isNumeric(outType)) {
- return null;
- }
- String argSql = args.get(0).accept(visitor);
- TSDataType argType = argSql == null ? null :
toTsDataType(args.get(0).getOutputDataType());
- if (argType == null || !isNumeric(argType)) {
- return null;
- }
- expression = "SUM(" + argSql + ")";
- } else if (simpleName.endsWith("MaxAggFunction") ||
simpleName.endsWith("MinAggFunction")) {
- if (argCount != 1) {
- return null;
- }
- String argSql = args.get(0).accept(visitor);
- if (argSql == null || toTsDataType(args.get(0).getOutputDataType()) ==
null) {
- return null;
- }
- String functionName = simpleName.endsWith("MaxAggFunction") ? "MAX" :
"MIN";
- expression = functionName + "(" + argSql + ")";
- } else {
- return null;
- }
-
- return "CAST(" + expression + " AS " + outType.name() + ")";
- }
-
- private static boolean isSupportedAggregate(AggregateExpression aggregate) {
- return aggregate != null
- && !aggregate.isDistinct()
- && !aggregate.isApproximate()
- && !aggregate.isIgnoreNulls()
- && !aggregate.getFilterExpression().isPresent();
- }
-
- private static boolean isNumeric(TSDataType dataType) {
- switch (dataType) {
- case INT32:
- case INT64:
- case FLOAT:
- case DOUBLE:
- return true;
- default:
- return false;
- }
- }
-
- private static boolean isCountType(TSDataType dataType) {
- return dataType == TSDataType.INT32 || dataType == TSDataType.INT64;
- }
-
- private static TSDataType toTsDataType(DataType dataType) {
- if (dataType == null) {
- return null;
- }
- try {
- return IoTDBUtils.toIoTDBDataType(dataType);
- } catch (RuntimeException e) {
- return null;
- }
- }
-}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBExpressionVisitor.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBExpressionVisitor.java
index e40e3cf..4884e39 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBExpressionVisitor.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBExpressionVisitor.java
@@ -114,7 +114,7 @@ public class IoTDBExpressionVisitor implements
ExpressionVisitor<String> {
@Override
public String visit(ValueLiteralExpression valueLiteral) {
- return IoTDBLiteralUtils.render(valueLiteral);
+ return IoTDBUtils.renderLiteral(valueLiteral);
}
@Override
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBLiteralUtils.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBLiteralUtils.java
deleted file mode 100644
index d8e7968..0000000
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/scan/pushdown/IoTDBLiteralUtils.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * 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.relational.flink.source.scan.pushdown;
-
-import org.apache.flink.table.expressions.ValueLiteralExpression;
-import org.apache.flink.table.types.logical.LogicalTypeRoot;
-
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.util.Optional;
-
-/** Renders Flink literal expressions as IoTDB SQL literals. */
-public final class IoTDBLiteralUtils {
-
- private IoTDBLiteralUtils() {}
-
- /**
- * Renders a literal for pushdown. Returns {@code null} if the literal type
or value cannot be
- * represented in IoTDB SQL.
- */
- public static String render(ValueLiteralExpression literal) {
- if (literal == null || literal.isNull()) {
- return null;
- }
-
- try {
- LogicalTypeRoot typeRoot =
literal.getOutputDataType().getLogicalType().getTypeRoot();
- switch (typeRoot) {
- case BOOLEAN:
- return
literal.getValueAs(Boolean.class).map(String::valueOf).orElse(null);
- case TINYINT:
- return value(literal.getValueAs(Byte.class));
- case SMALLINT:
- return value(literal.getValueAs(Short.class));
- case INTEGER:
- return value(literal.getValueAs(Integer.class));
- case BIGINT:
- return value(literal.getValueAs(Long.class));
- case FLOAT:
- return literal
- .getValueAs(Float.class)
- .filter(value -> !value.isNaN() && !value.isInfinite())
- .map(String::valueOf)
- .orElse(null);
- case DOUBLE:
- return literal
- .getValueAs(Double.class)
- .filter(value -> !value.isNaN() && !value.isInfinite())
- .map(String::valueOf)
- .orElse(null);
- case CHAR:
- case VARCHAR:
- return
literal.getValueAs(String.class).map(IoTDBLiteralUtils::quoteString).orElse(null);
- case BINARY:
- case VARBINARY:
- return
literal.getValueAs(byte[].class).map(IoTDBLiteralUtils::formatBinary).orElse(null);
- case DATE:
- return literal
- .getValueAs(LocalDate.class)
- .map(value -> "CAST('" + value + "' AS DATE)")
- .orElse(null);
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- return literal
- .getValueAs(LocalDateTime.class)
- .map(value -> "CAST('" + value + "' AS TIMESTAMP)")
- .orElse(null);
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return literal
- .getValueAs(Instant.class)
- .map(value -> "CAST('" + value + "' AS TIMESTAMP)")
- .orElse(null);
- default:
- return null;
- }
- } catch (RuntimeException e) {
- return null;
- }
- }
-
- private static String value(Optional<?> value) {
- return value.map(Object::toString).orElse(null);
- }
-
- private static String quoteString(String value) {
- return "'" + value.replace("'", "''") + "'";
- }
-
- private static String formatBinary(byte[] value) {
- StringBuilder builder = new StringBuilder(value.length * 2 + 3);
- builder.append("X'");
- for (byte b : value) {
- builder.append(String.format("%02X", b));
- }
- builder.append("'");
- return builder.toString();
- }
-}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
index f3c9237..183b1ee 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBUtils.java
@@ -19,12 +19,25 @@
package org.apache.iotdb.relational.flink.utils;
+import org.apache.iotdb.relational.flink.source.scan.pushdown.AggregateSpec;
+import
org.apache.iotdb.relational.flink.source.scan.pushdown.IoTDBExpressionVisitor;
+
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.TimestampType;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -32,13 +45,16 @@ import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
/**
* Shared utilities for the IoTDB relational (table model) Flink connector.
*
- * <p>This class centralizes identifier quoting, Flink/IoTDB data type
conversion and the column
- * name/category resolution used by both the Flink catalog (DDL) and the sink
serializer.
+ * <p>This class centralizes identifier quoting, Flink/IoTDB data type
conversion, the column
+ * name/category resolution used by both the Flink catalog (DDL) and the sink
serializer, the
+ * rendering of Flink planner/runtime values as IoTDB SQL literals, and the
aggregate pushdown
+ * translation.
*/
public final class IoTDBUtils {
@@ -140,8 +156,8 @@ public final class IoTDBUtils {
}
/**
- * Validates the {@code time-column}, {@code tag-columns} and {@code
attribute-columns} options
- * against an existing table schema.
+ * Validates the {@code iotdb.time-column}, {@code iotdb.tag-columns} and
{@code
+ * iotdb.attribute-columns} options against an existing table schema.
*
* @param timeColumn normalized TIME column name
* @param tagColumns normalized TAG column names
@@ -154,7 +170,8 @@ public final class IoTDBUtils {
Collection<String> attributeColumns,
Map<String, TSDataType> dataTypesByColumn) {
if (timeColumn == null || timeColumn.isEmpty()) {
- throw new CatalogException("Table option 'time-column' must specify the
IoTDB TIME column.");
+ throw new CatalogException(
+ "Table option 'iotdb.time-column' must specify the IoTDB TIME
column.");
}
if (tagColumns.contains(timeColumn) ||
attributeColumns.contains(timeColumn)) {
throw new CatalogException("The TIME column cannot also be a TAG or
ATTRIBUTE column.");
@@ -165,12 +182,12 @@ public final class IoTDBUtils {
throw new CatalogException(
"TAG and ATTRIBUTE columns must not overlap: " +
overlappingColumns.iterator().next());
}
- validateColumnExists(timeColumn, "time-column", dataTypesByColumn);
+ validateColumnExists(timeColumn, "iotdb.time-column", dataTypesByColumn);
for (String columnName : tagColumns) {
- validateColumnExists(columnName, "tag-columns", dataTypesByColumn);
+ validateColumnExists(columnName, "iotdb.tag-columns", dataTypesByColumn);
}
for (String columnName : attributeColumns) {
- validateColumnExists(columnName, "attribute-columns", dataTypesByColumn);
+ validateColumnExists(columnName, "iotdb.attribute-columns",
dataTypesByColumn);
}
if (dataTypesByColumn.get(timeColumn) != TSDataType.TIMESTAMP) {
throw new CatalogException("The IoTDB TIME column must use the TIMESTAMP
data type.");
@@ -278,6 +295,303 @@ public final class IoTDBUtils {
return sql.toString();
}
+ /**
+ * Renders a Flink planner literal as an IoTDB SQL literal. Returns {@code
null} if the literal
+ * type or value cannot be represented in IoTDB SQL.
+ */
+ public static String renderLiteral(ValueLiteralExpression literal) {
+ if (literal == null || literal.isNull()) {
+ return null;
+ }
+
+ try {
+ LogicalTypeRoot typeRoot =
literal.getOutputDataType().getLogicalType().getTypeRoot();
+ switch (typeRoot) {
+ case BOOLEAN:
+ return
literal.getValueAs(Boolean.class).map(String::valueOf).orElse(null);
+ case TINYINT:
+ return value(literal.getValueAs(Byte.class));
+ case SMALLINT:
+ return value(literal.getValueAs(Short.class));
+ case INTEGER:
+ return value(literal.getValueAs(Integer.class));
+ case BIGINT:
+ return value(literal.getValueAs(Long.class));
+ case FLOAT:
+ return literal
+ .getValueAs(Float.class)
+ .filter(value -> !value.isNaN() && !value.isInfinite())
+ .map(String::valueOf)
+ .orElse(null);
+ case DOUBLE:
+ return literal
+ .getValueAs(Double.class)
+ .filter(value -> !value.isNaN() && !value.isInfinite())
+ .map(String::valueOf)
+ .orElse(null);
+ case CHAR:
+ case VARCHAR:
+ return
literal.getValueAs(String.class).map(IoTDBUtils::quoteString).orElse(null);
+ case BINARY:
+ case VARBINARY:
+ return
literal.getValueAs(byte[].class).map(IoTDBUtils::formatBinary).orElse(null);
+ case DATE:
+ return literal
+ .getValueAs(LocalDate.class)
+ .map(value -> "CAST('" + value + "' AS DATE)")
+ .orElse(null);
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return literal
+ .getValueAs(LocalDateTime.class)
+ .map(value -> "CAST('" + value + "' AS TIMESTAMP)")
+ .orElse(null);
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return literal
+ .getValueAs(Instant.class)
+ .map(value -> "CAST('" + value + "' AS TIMESTAMP)")
+ .orElse(null);
+ default:
+ return null;
+ }
+ } catch (RuntimeException e) {
+ return null;
+ }
+ }
+
+ /**
+ * Renders a runtime {@link RowData} field as an IoTDB SQL literal. Returns
{@code null} when the
+ * value is {@code null} or cannot be represented in IoTDB SQL.
+ */
+ public static String renderLiteral(RowData row, int position, DataType
dataType) {
+ if (row == null || row.isNullAt(position) || dataType == null) {
+ return null;
+ }
+
+ LogicalTypeRoot typeRoot = dataType.getLogicalType().getTypeRoot();
+ switch (typeRoot) {
+ case BOOLEAN:
+ return Boolean.toString(row.getBoolean(position));
+ case TINYINT:
+ return Byte.toString(row.getByte(position));
+ case SMALLINT:
+ return Short.toString(row.getShort(position));
+ case INTEGER:
+ return Integer.toString(row.getInt(position));
+ case BIGINT:
+ return Long.toString(row.getLong(position));
+ case FLOAT:
+ float floatValue = row.getFloat(position);
+ return isFinite(floatValue) ? Float.toString(floatValue) : null;
+ case DOUBLE:
+ double doubleValue = row.getDouble(position);
+ return isFinite(doubleValue) ? Double.toString(doubleValue) : null;
+ case CHAR:
+ case VARCHAR:
+ return quoteString(row.getString(position).toString());
+ case BINARY:
+ case VARBINARY:
+ byte[] bytes = row.getBinary(position);
+ return bytes == null ? null : formatBinary(bytes);
+ case DATE:
+ return "CAST('" + LocalDate.ofEpochDay(row.getInt(position)) + "' AS
DATE)";
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return "CAST('"
+ + getTimestamp(row, position, dataType).toLocalDateTime()
+ + "' AS TIMESTAMP)";
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return "CAST('" + getTimestamp(row, position, dataType).toInstant() +
"' AS TIMESTAMP)";
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Translates Flink aggregate pushdown information into a serializable
{@link AggregateSpec}.
+ *
+ * <p>The pushdown is all-or-nothing: any unsupported aggregate, argument or
grouping makes this
+ * return {@code null} so the whole aggregation stays in Flink. Each
aggregate is classified and
+ * translated in a single pass from its function class name; grouping
columns and aggregate
+ * arguments are rendered through {@link IoTDBExpressionVisitor}.
+ */
+ public static AggregateSpec translateAggregate(
+ List<int[]> groupingSets,
+ List<AggregateExpression> aggregateExpressions,
+ DataType sourceRowDataType,
+ DataType producedDataType) {
+ if (groupingSets == null
+ || groupingSets.size() != 1
+ || aggregateExpressions == null
+ || aggregateExpressions.isEmpty()) {
+ return null;
+ }
+
+ final List<String> sourceFieldNames;
+ final List<DataType> producedFieldTypes;
+ try {
+ sourceFieldNames = DataType.getFieldNames(sourceRowDataType);
+ producedFieldTypes = DataType.getFieldDataTypes(producedDataType);
+ } catch (RuntimeException e) {
+ return null;
+ }
+
+ int[] grouping = groupingSets.get(0);
+ if (grouping == null) {
+ return null;
+ }
+
+ List<String> groupByExpressions = new ArrayList<>(grouping.length);
+ List<String> selectExpressions = new ArrayList<>(grouping.length +
aggregateExpressions.size());
+ for (int index : grouping) {
+ if (index < 0 || index >= sourceFieldNames.size()) {
+ return null;
+ }
+ String column = quoteIdentifier(sourceFieldNames.get(index));
+ groupByExpressions.add(column);
+ selectExpressions.add(column);
+ }
+
+ IoTDBExpressionVisitor visitor = new IoTDBExpressionVisitor();
+ for (int i = 0; i < aggregateExpressions.size(); i++) {
+ int producedIndex = grouping.length + i;
+ if (producedIndex >= producedFieldTypes.size()) {
+ return null;
+ }
+ String sql =
+ translateAggregateFunction(
+ aggregateExpressions.get(i),
producedFieldTypes.get(producedIndex), visitor);
+ if (sql == null) {
+ return null;
+ }
+ selectExpressions.add(sql);
+ }
+
+ return new AggregateSpec(selectExpressions, groupByExpressions);
+ }
+
+ private static String translateAggregateFunction(
+ AggregateExpression aggregate, DataType producedType,
IoTDBExpressionVisitor visitor) {
+ if (!isSupportedAggregate(aggregate)) {
+ return null;
+ }
+
+ Class<?> functionClass = aggregate.getFunctionDefinition().getClass();
+ String simpleName = functionClass == null ? null :
functionClass.getSimpleName();
+ TSDataType outType = toTsDataType(producedType);
+ if (simpleName == null || outType == null) {
+ return null;
+ }
+
+ List<FieldReferenceExpression> args = aggregate.getArgs();
+ int argCount = args == null ? 0 : args.size();
+
+ String expression;
+ if (simpleName.endsWith("Count1AggFunction")) {
+ if (argCount != 0 || !isCountType(outType)) {
+ return null;
+ }
+ expression = "COUNT(*)";
+ } else if (simpleName.endsWith("CountAggFunction")) {
+ if (argCount != 1 || !isCountType(outType)) {
+ return null;
+ }
+ String argSql = args.get(0).accept(visitor);
+ if (argSql == null || toTsDataType(args.get(0).getOutputDataType()) ==
null) {
+ return null;
+ }
+ expression = "COUNT(" + argSql + ")";
+ } else if (simpleName.endsWith("Sum0AggFunction") ||
simpleName.endsWith("SumAggFunction")) {
+ if (argCount != 1 || !isNumeric(outType)) {
+ return null;
+ }
+ String argSql = args.get(0).accept(visitor);
+ TSDataType argType = argSql == null ? null :
toTsDataType(args.get(0).getOutputDataType());
+ if (argType == null || !isNumeric(argType)) {
+ return null;
+ }
+ expression = "SUM(" + argSql + ")";
+ } else if (simpleName.endsWith("MaxAggFunction") ||
simpleName.endsWith("MinAggFunction")) {
+ if (argCount != 1) {
+ return null;
+ }
+ String argSql = args.get(0).accept(visitor);
+ if (argSql == null || toTsDataType(args.get(0).getOutputDataType()) ==
null) {
+ return null;
+ }
+ String functionName = simpleName.endsWith("MaxAggFunction") ? "MAX" :
"MIN";
+ expression = functionName + "(" + argSql + ")";
+ } else {
+ return null;
+ }
+
+ return "CAST(" + expression + " AS " + outType.name() + ")";
+ }
+
+ private static boolean isSupportedAggregate(AggregateExpression aggregate) {
+ return aggregate != null
+ && !aggregate.isDistinct()
+ && !aggregate.isApproximate()
+ && !aggregate.isIgnoreNulls()
+ && !aggregate.getFilterExpression().isPresent();
+ }
+
+ private static boolean isNumeric(TSDataType dataType) {
+ switch (dataType) {
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ private static boolean isCountType(TSDataType dataType) {
+ return dataType == TSDataType.INT32 || dataType == TSDataType.INT64;
+ }
+
+ private static TSDataType toTsDataType(DataType dataType) {
+ if (dataType == null) {
+ return null;
+ }
+ try {
+ return toIoTDBDataType(dataType);
+ } catch (RuntimeException e) {
+ return null;
+ }
+ }
+
+ private static TimestampData getTimestamp(RowData row, int position,
DataType dataType) {
+ int precision = ((TimestampType) dataType.getLogicalType()).getPrecision();
+ return row.getTimestamp(position, precision);
+ }
+
+ private static boolean isFinite(float value) {
+ return !Float.isNaN(value) && !Float.isInfinite(value);
+ }
+
+ private static boolean isFinite(double value) {
+ return !Double.isNaN(value) && !Double.isInfinite(value);
+ }
+
+ private static String value(Optional<?> value) {
+ return value.map(Object::toString).orElse(null);
+ }
+
+ private static String quoteString(String value) {
+ return "'" + value.replace("'", "''") + "'";
+ }
+
+ private static String formatBinary(byte[] value) {
+ StringBuilder builder = new StringBuilder(value.length * 2 + 3);
+ builder.append("X'");
+ for (byte b : value) {
+ builder.append(String.format("%02X", b));
+ }
+ builder.append("'");
+ return builder.toString();
+ }
+
private static void validateColumnExists(
String columnName, String optionName, Map<String, TSDataType>
dataTypesByColumn) {
if (!dataTypesByColumn.containsKey(columnName)) {
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
index 3e3d6c0..25f4ebd 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/catalog/IoTDBCatalog.java
@@ -342,10 +342,10 @@ public class IoTDBCatalog extends AbstractCatalog {
Map<String, String> tableOptions = table.getOptions();
String timeColumn = getRequiredTimeColumn(tableOptions);
Set<String> tagColumns =
- parseColumnNames(tableOptions.get(IoTDBOptions.TAG_COLUMNS.key()),
"tag-columns");
+ parseColumnNames(tableOptions.get(IoTDBOptions.TAG_COLUMNS.key()),
"iotdb.tag-columns");
Set<String> attributeColumns =
parseColumnNames(
- tableOptions.get(IoTDBOptions.ATTRIBUTE_COLUMNS.key()),
"attribute-columns");
+ tableOptions.get(IoTDBOptions.ATTRIBUTE_COLUMNS.key()),
"iotdb.attribute-columns");
List<String> columnNames = new ArrayList<>();
List<TSDataType> dataTypes = new ArrayList<>();
@@ -383,7 +383,7 @@ public class IoTDBCatalog extends AbstractCatalog {
String timeColumn = tableOptions.get(IoTDBOptions.TIME_COLUMN.key());
if (timeColumn == null || timeColumn.trim().isEmpty()) {
throw new CatalogException(
- "Table option 'time-column' must specify the IoTDB TIME column for
CREATE TABLE.");
+ "Table option 'iotdb.time-column' must specify the IoTDB TIME column
for CREATE TABLE.");
}
return IoTDBUtils.normalizeColumnName(timeColumn);
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
index c41d3dc..0422e34 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/sink/IoTDBSinkWriter.java
@@ -159,13 +159,13 @@ public class IoTDBSinkWriter<IN> implements
SinkWriter<IN> {
private void initializeTablet() throws IOException {
String tableName = options.getTable();
if (tableName == null || tableName.trim().isEmpty()) {
- throw new IOException("Table option 'table' must specify the IoTDB table
name.");
+ throw new IOException("Table option 'iotdb.table' must specify the IoTDB
table name.");
}
String rawTimeColumn = options.getTimeColumn();
if (rawTimeColumn == null || rawTimeColumn.trim().isEmpty()) {
throw new IOException(
- "Table option 'time-column' is required to write into an IoTDB table
model table.");
+ "Table option 'iotdb.time-column' is required to write into an IoTDB
table model table.");
}
try {
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
index 64bbd76..966651f 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
@@ -26,7 +26,6 @@ import
org.apache.iotdb.relational.flink.source.lookup.IoTDBAsyncLookupFunction;
import org.apache.iotdb.relational.flink.source.lookup.IoTDBLookupFunction;
import org.apache.iotdb.relational.flink.source.scan.IoTDBSource;
import org.apache.iotdb.relational.flink.source.scan.pushdown.AggregateSpec;
-import
org.apache.iotdb.relational.flink.source.scan.pushdown.IoTDBAggregatePushDownUtils;
import
org.apache.iotdb.relational.flink.source.scan.pushdown.IoTDBExpressionVisitor;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
@@ -63,10 +62,10 @@ import java.util.List;
* <p>It supports three read paths:
*
* <ul>
- * <li>bounded scan reads ({@code scan.mode=snapshot}, the default) with
filter, projection, limit
- * and aggregate pushdown;
- * <li>lookup reads, either synchronous or asynchronous ({@code
lookup.async});
- * <li>unbounded CDC reads backed by the IoTDB subscription API ({@code
scan.mode=cdc}).
+ * <li>bounded scan reads ({@code iotdb.scan.mode=snapshot}, the default)
with filter, projection,
+ * limit and aggregate pushdown;
+ * <li>lookup reads, either synchronous or asynchronous ({@code
iotdb.lookup.async});
+ * <li>unbounded CDC reads backed by the IoTDB subscription API ({@code
iotdb.scan.mode=cdc}).
* </ul>
*
* <p>All rows are emitted as inserts, so the changelog mode is insert-only.
@@ -196,7 +195,7 @@ public class IoTDBRelationalDynamicTableSource
// Grouping and argument indices refer to the scan's current row type,
which is the row type
// after any projection that has already been pushed into this source.
AggregateSpec spec =
- IoTDBAggregatePushDownUtils.translate(
+ IoTDBUtils.translateAggregate(
groupingSets, aggregateExpressions, physicalRowDataType,
producedDataType);
if (spec == null) {
return false;
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java
index 68c6712..6bff06a 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/IoTDBRelationalLocalQueryManual.java
@@ -140,25 +140,25 @@ public class IoTDBRelationalLocalQueryManual {
.append(" `temperature` DOUBLE\n")
.append(") WITH (\n")
.append(" 'connector' = 'iotdb-relational',\n")
- .append(" 'nodeUrls' = '")
+ .append(" 'iotdb.node-urls' = '")
.append(nodeUrls)
.append("',\n")
- .append(" 'user' = '")
+ .append(" 'iotdb.user' = '")
.append(user)
.append("',\n")
- .append(" 'password' = '")
+ .append(" 'iotdb.password' = '")
.append(password)
.append("',\n")
- .append(" 'database' = '")
+ .append(" 'iotdb.database' = '")
.append(database)
.append("',\n")
- .append(" 'table' = '")
+ .append(" 'iotdb.table' = '")
.append(table)
.append("',\n")
- .append(" 'time-column' = 'time',\n")
- .append(" 'tag-columns' = 'device_id'");
+ .append(" 'iotdb.time-column' = 'time',\n")
+ .append(" 'iotdb.tag-columns' = 'device_id'");
if (async != null) {
- ddl.append(",\n 'lookup.async' = '").append(async).append("'");
+ ddl.append(",\n 'iotdb.lookup.async' = '").append(async).append("'");
}
ddl.append("\n)");
return ddl.toString();
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalSourcePushDownPlannerTest.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalSourcePushDownPlannerTest.java
index 497a85f..559efea 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalSourcePushDownPlannerTest.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalSourcePushDownPlannerTest.java
@@ -50,13 +50,13 @@ public class IoTDBRelationalSourcePushDownPlannerTest {
+ " `humidity` DOUBLE\n"
+ ") WITH (\n"
+ " 'connector' = 'iotdb-relational',\n"
- + " 'nodeUrls' = '127.0.0.1:6667',\n"
- + " 'user' = 'root',\n"
- + " 'password' = 'root',\n"
- + " 'database' = 'test',\n"
- + " 'table' = 'sensor',\n"
- + " 'time-column' = 'time',\n"
- + " 'tag-columns' = 'device_id'\n"
+ + " 'iotdb.node-urls' = '127.0.0.1:6667',\n"
+ + " 'iotdb.user' = 'root',\n"
+ + " 'iotdb.password' = 'root',\n"
+ + " 'iotdb.database' = 'test',\n"
+ + " 'iotdb.table' = 'sensor',\n"
+ + " 'iotdb.time-column' = 'time',\n"
+ + " 'iotdb.tag-columns' = 'device_id'\n"
+ ")";
@Test
