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
commit 54ba6ad99a4814203a72ead1d0266562cd70ff2a Author: shuwenwei <[email protected]> AuthorDate: Fri Sep 18 18:09:42 2026 +0800 add ut --- .../source/pushdown/IoTDBExpressionVisitor.java | 60 +++- .../utils/IoTDBUtilsBuildSelectQueryTest.java | 52 ++++ .../flink-iotdb-table-connector-flink1/pom.xml | 7 +- .../relational/flink/sink/IoTDBSinkWriter.java | 7 +- .../table/IoTDBRelationalDynamicTableSource.java | 25 +- .../IoTDBRelationalSourcePushDownPlannerTest.java | 342 +++++++++++++++++++++ 6 files changed, 471 insertions(+), 22 deletions(-) diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBExpressionVisitor.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBExpressionVisitor.java index da9c874..9d3834d 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBExpressionVisitor.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/pushdown/IoTDBExpressionVisitor.java @@ -66,13 +66,11 @@ public class IoTDBExpressionVisitor implements ExpressionVisitor<String> { functionNames.put("log10", "log10"); functionNames.put("lower", "lower"); functionNames.put("lowercase", "lower"); - functionNames.put("ltrim", "ltrim"); functionNames.put("pi", "pi"); functionNames.put("radians", "radians"); functionNames.put("regexp", "regexp_like"); functionNames.put("replace", "replace"); functionNames.put("round", "round"); - functionNames.put("rtrim", "rtrim"); functionNames.put("sign", "sign"); functionNames.put("sin", "sin"); functionNames.put("sinh", "sinh"); @@ -84,6 +82,20 @@ public class IoTDBExpressionVisitor implements ExpressionVisitor<String> { functionNames.put("trim", "trim"); functionNames.put("upper", "upper"); functionNames.put("uppercase", "upper"); + functionNames.put("bitand", "bitwise_and"); + functionNames.put("bitor", "bitwise_or"); + functionNames.put("bitxor", "bitwise_xor"); + functionNames.put("bitnot", "bitwise_not"); + functionNames.put("bitshiftleft", "bitwise_left_shift"); + functionNames.put("bitshiftright", "bitwise_right_shift"); + functionNames.put("to_base64", "to_base64"); + functionNames.put("from_base64", "from_base64"); + functionNames.put("to_hex", "to_hex"); + functionNames.put("from_hex", "from_hex"); + functionNames.put("md5", "md5"); + functionNames.put("sha1", "sha1"); + functionNames.put("sha256", "sha256"); + functionNames.put("sha512", "sha512"); FLINK_TO_IOTDB_FUNCTION_NAMES = Collections.unmodifiableMap(functionNames); } @@ -184,18 +196,19 @@ public class IoTDBExpressionVisitor implements ExpressionVisitor<String> { return visitLocate(children); case "instr": return visitInstr(children); - case "current_database": - return visitCurrentTime("CURRENT_DATABASE", children); + case "ltrim": + return visitTrim("LEADING", children); + case "rtrim": + return visitTrim("TRAILING", children); case "current_date": - return visitCurrentTime("CURRENT_DATE", children); - case "current_time": - return visitCurrentTime("CURRENT_TIME", children); + return visitCurrentValue("CAST(now() AS DATE)", children); case "current_timestamp": - return visitCurrentTime("CURRENT_TIMESTAMP", children); - case "localtime": - return visitCurrentTime("LOCALTIME", children); case "localtimestamp": - return visitCurrentTime("LOCALTIMESTAMP", children); + return visitCurrentValue("now()", children); + case "current_database": + case "current_time": + case "localtime": + return null; default: return visitScalarFunction(expressionName, children); } @@ -377,8 +390,26 @@ public class IoTDBExpressionVisitor implements ExpressionVisitor<String> { return builder.append(')').toString(); } - private String visitCurrentTime(String keyword, List<ResolvedExpression> children) { - return children == null || children.isEmpty() ? keyword : null; + private String visitCurrentValue(String sql, List<ResolvedExpression> children) { + return children == null || children.isEmpty() ? sql : null; + } + + private String visitTrim(String specification, List<ResolvedExpression> children) { + if (children == null || (children.size() != 1 && children.size() != 2)) { + return null; + } + String value = buildIoTDBExpressionSQL(children.get(0)); + if (value == null) { + return null; + } + if (children.size() == 1) { + return "trim(" + specification + " FROM " + value + ")"; + } + String trimCharacter = buildIoTDBExpressionSQL(children.get(1)); + if (trimCharacter == null) { + return null; + } + return "trim(" + specification + " " + trimCharacter + " FROM " + value + ")"; } private String visitBinary(String operator, List<ResolvedExpression> children) { @@ -550,8 +581,7 @@ public class IoTDBExpressionVisitor implements ExpressionVisitor<String> { } String functionName = normalizeFunctionName(call.getFunctionName()); - String mappedFunctionName = FLINK_TO_IOTDB_FUNCTION_NAMES.get(functionName); - return mappedFunctionName == null ? functionName : mappedFunctionName; + return functionName; } private static String normalizeFunctionName(String functionName) { diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/utils/IoTDBUtilsBuildSelectQueryTest.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/utils/IoTDBUtilsBuildSelectQueryTest.java new file mode 100644 index 0000000..1c612bc --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/utils/IoTDBUtilsBuildSelectQueryTest.java @@ -0,0 +1,52 @@ +/* + * 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.utils; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.types.DataType; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; + +public class IoTDBUtilsBuildSelectQueryTest { + + private static final DataType ROW = + ResolvedSchema.physical( + new String[] {"a", "b"}, new DataType[] {DataTypes.INT(), DataTypes.STRING()}) + .toPhysicalRowDataType(); + + @Test + public void testSelectOnly() { + assertEquals( + "SELECT \"a\", \"b\" FROM \"sensor\"", + IoTDBUtils.buildSelectQuery("sensor", ROW, Collections.emptyList(), -1L)); + } + + @Test + public void testSelectWithFiltersAndLimit() { + assertEquals( + "SELECT \"a\", \"b\" FROM \"sensor\" WHERE (a = 1) AND (b = 'x') LIMIT 3", + IoTDBUtils.buildSelectQuery("sensor", ROW, Arrays.asList("(a = 1)", "(b = 'x')"), 3L)); + } +} diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/pom.xml b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/pom.xml index 2733253..ce7e73e 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/pom.xml +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/pom.xml @@ -66,9 +66,14 @@ </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-planner-loader</artifactId> + <artifactId>flink-table-planner_2.12</artifactId> <version>${flink.version}</version> <scope>test</scope> </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <scope>test</scope> + </dependency> </dependencies> </project> 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 cbb045c..c41d3dc 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 @@ -65,9 +65,7 @@ public class IoTDBSinkWriter<IN> implements SinkWriter<IN> { private int timeColumnIndex; public IoTDBSinkWriter( - IoTDBOptions options, - DataType physicalRowDataType, - SinkDataConverter<IN> converter) + IoTDBOptions options, DataType physicalRowDataType, SinkDataConverter<IN> converter) throws IOException { this.options = options; this.physicalRowDataType = physicalRowDataType; @@ -251,7 +249,8 @@ public class IoTDBSinkWriter<IN> implements SinkWriter<IN> { } private Object readValue( - SinkDataConverter.Iterator iterator, int columnIndex, TSDataType dataType) throws IOException { + SinkDataConverter.Iterator iterator, int columnIndex, TSDataType dataType) + throws IOException { if (iterator.isNull(columnIndex)) { return null; } 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 a3cd50e..6699168 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 @@ -23,10 +23,10 @@ import org.apache.iotdb.relational.flink.cfg.IoTDBOptions; import org.apache.iotdb.relational.flink.source.IoTDBSource; import org.apache.iotdb.relational.flink.source.deserializer.RowDataDeserializationSchema; import org.apache.iotdb.relational.flink.source.pushdown.IoTDBExpressionVisitor; +import org.apache.iotdb.relational.flink.utils.IoTDBUtils; import org.apache.flink.table.catalog.ResolvedSchema; import org.apache.flink.table.connector.ChangelogMode; -import org.apache.flink.table.connector.Projection; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.SourceProvider; @@ -87,7 +87,7 @@ public class IoTDBRelationalDynamicTableSource @Override public void applyProjection(int[][] projectedFields, DataType producedDataType) { - this.physicalRowDataType = Projection.of(projectedFields).project(physicalRowDataType); + this.physicalRowDataType = producedDataType; } @Override @@ -129,4 +129,25 @@ public class IoTDBRelationalDynamicTableSource public String asSummaryString() { return "IoTDB Relational Dynamic Table Source"; } + + /** + * Builds the IoTDB query that this source would execute. Exposed for tests so the pushed-down + * projection, filters and limit can be verified without executing any query. + */ + String buildQuery() { + return IoTDBUtils.buildSelectQuery( + options.getTable(), physicalRowDataType, resolvedFilterQueries, limit); + } + + List<String> getResolvedFilterQueries() { + return resolvedFilterQueries; + } + + long getLimit() { + return limit; + } + + DataType getPhysicalRowDataType() { + return physicalRowDataType; + } } 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 new file mode 100644 index 0000000..2e53e0f --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/test/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalSourcePushDownPlannerTest.java @@ -0,0 +1,342 @@ +/* + * 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.table; + +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.RelNode; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.internal.TableEnvironmentImpl; +import org.apache.flink.table.api.internal.TableImpl; +import org.apache.flink.table.planner.delegation.PlannerBase; +import org.apache.flink.table.planner.plan.schema.TableSourceTable; +import org.junit.Test; + +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * Planner-driven test for the source pushdown. The Flink SQL is optimized through the real planner + * (so the filter/projection/limit pushdown rules run and mutate the source), but the resulting + * IoTDB query is only rendered, never executed. + */ +public class IoTDBRelationalSourcePushDownPlannerTest { + + private static final String DDL = + "CREATE TABLE iotdb_t (\n" + + " `time` TIMESTAMP(3),\n" + + " `device_id` STRING,\n" + + " `temperature` DOUBLE,\n" + + " `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" + + ")"; + + @Test + public void testProjectionFilterAndLimitPushDown() { + // Filter on device_id and select columns that are not made constant by the filter. + IoTDBRelationalDynamicTableSource source = + optimize("SELECT `time`, temperature FROM iotdb_t WHERE device_id = 'd1' LIMIT 5"); + + assertEquals( + Collections.singletonList("(\"device_id\" = 'd1')"), source.getResolvedFilterQueries()); + assertEquals(5L, source.getLimit()); + assertEquals( + "SELECT \"time\", \"temperature\" FROM \"sensor\" " + + "WHERE (\"device_id\" = 'd1') LIMIT 5", + source.buildQuery()); + } + + @Test + public void testProjectionOnly() { + IoTDBRelationalDynamicTableSource source = optimize("SELECT device_id FROM iotdb_t"); + + assertEquals(Collections.emptyList(), source.getResolvedFilterQueries()); + assertEquals(-1L, source.getLimit()); + assertEquals("SELECT \"device_id\" FROM \"sensor\"", source.buildQuery()); + } + + @Test + public void testIsNullPushDown() { + // Under `device_id IS NULL`, Flink constant-folds device_id, so select another column. + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE device_id IS NULL"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" WHERE (\"device_id\" IS NULL)", + source.buildQuery()); + } + + @Test + public void testInPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE device_id IN ('d1', 'd2')"); + + // Flink normalizes IN into a chain of OR comparisons before pushdown. + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" " + + "WHERE ((\"device_id\" = 'd1') OR (\"device_id\" = 'd2'))", + source.buildQuery()); + } + + @Test + public void testLikePushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE device_id LIKE 'd%'"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (\"device_id\" LIKE 'd%')", + source.buildQuery()); + } + + @Test + public void testArithmeticFilterPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE temperature + humidity > 30.0E0"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" " + + "WHERE ((\"temperature\" + \"humidity\") > 30.0)", + source.buildQuery()); + } + + @Test + public void testMultiplicationFilterPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE temperature * 2.0E0 > 40.0E0"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE ((\"temperature\" * 2.0) > 40.0)", + source.buildQuery()); + } + + @Test + public void testOrPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE device_id = 'd1' OR device_id = 'd2'"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" " + + "WHERE ((\"device_id\" = 'd1') OR (\"device_id\" = 'd2'))", + source.buildQuery()); + } + + @Test + public void testIsNotNullPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE temperature IS NOT NULL"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (\"temperature\" IS NOT NULL)", + source.buildQuery()); + } + + @Test + public void testFunctionPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE LOWER(device_id) = 'd1'"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" WHERE (lower(\"device_id\") = 'd1')", + source.buildQuery()); + } + + @Test + public void testNotEqualsPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE temperature <> 0.0E0"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (\"temperature\" <> 0.0)", + source.buildQuery()); + } + + @Test + public void testLessOrEqualPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE temperature <= 30.0E0"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (\"temperature\" <= 30.0)", + source.buildQuery()); + } + + @Test + public void testGreaterOrEqualPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE temperature >= 30.0E0"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (\"temperature\" >= 30.0)", + source.buildQuery()); + } + + @Test + public void testCastPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE CAST(temperature AS INT) > 0"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (CAST(\"temperature\" AS INT32) > 0)", + source.buildQuery()); + } + + @Test + public void testCoalescePushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE COALESCE(temperature, 0.0E0) > 10.0E0"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (coalesce(\"temperature\", 0.0) > 10.0)", + source.buildQuery()); + } + + @Test + public void testUpperFunctionPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE UPPER(device_id) = 'D1'"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" WHERE (upper(\"device_id\") = 'D1')", + source.buildQuery()); + } + + @Test + public void testPartialPushDown() { + // device_id = 'd1' is mappable, RAND() is not; select a non-constrained column so it is not + // constant-folded. + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE device_id = 'd1' AND RAND() > 0.5"); + + assertEquals( + Collections.singletonList("(\"device_id\" = 'd1')"), source.getResolvedFilterQueries()); + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" WHERE (\"device_id\" = 'd1')", + source.buildQuery()); + } + + @Test + public void testLtrimPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE LTRIM(device_id) = 'd1'"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" " + + "WHERE (trim(LEADING FROM \"device_id\") = 'd1')", + source.buildQuery()); + } + + @Test + public void testRtrimPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE RTRIM(device_id) = 'd1'"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" " + + "WHERE (trim(TRAILING FROM \"device_id\") = 'd1')", + source.buildQuery()); + } + + @Test + public void testMd5PushDown() { + // Flink wraps the hash result in a CAST to STRING before the comparison. + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE MD5(device_id) = 'x'"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" " + + "WHERE (CAST(md5(\"device_id\") AS STRING) = 'x')", + source.buildQuery()); + } + + @Test + public void testSha256PushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT temperature FROM iotdb_t WHERE SHA256(device_id) = 'x'"); + + assertEquals( + "SELECT \"temperature\" FROM \"sensor\" " + + "WHERE (CAST(sha256(\"device_id\") AS STRING) = 'x')", + source.buildQuery()); + } + + @Test + public void testNowPushDown() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE `time` < NOW()"); + + assertEquals( + "SELECT \"device_id\" FROM \"sensor\" WHERE (\"time\" < now())", + source.buildQuery()); + } + + @Test + public void testUnsupportedPredicateIsNotPushed() { + IoTDBRelationalDynamicTableSource source = + optimize("SELECT device_id FROM iotdb_t WHERE device_id = uuid()"); + + assertEquals(Collections.emptyList(), source.getResolvedFilterQueries()); + assertEquals("SELECT \"device_id\" FROM \"sensor\"", source.buildQuery()); + } + + private static IoTDBRelationalDynamicTableSource optimize(String query) { + TableEnvironmentImpl tableEnvironment = + (TableEnvironmentImpl) TableEnvironment.create(EnvironmentSettings.inBatchMode()); + tableEnvironment.executeSql(DDL); + Table table = tableEnvironment.sqlQuery(query); + + PlannerBase planner = (PlannerBase) tableEnvironment.getPlanner(); + RelNode logical = + planner.createRelBuilder().queryOperation(((TableImpl) table).getQueryOperation()).build(); + RelNode optimized = planner.optimize(logical); + + IoTDBRelationalDynamicTableSource source = findSource(optimized); + assertNotNull(source); + return source; + } + + private static IoTDBRelationalDynamicTableSource findSource(RelNode rel) { + RelOptTable table = rel.getTable(); + if (table != null) { + TableSourceTable sourceTable = table.unwrap(TableSourceTable.class); + if (sourceTable != null + && sourceTable.tableSource() instanceof IoTDBRelationalDynamicTableSource) { + return (IoTDBRelationalDynamicTableSource) sourceTable.tableSource(); + } + } + for (RelNode input : rel.getInputs()) { + IoTDBRelationalDynamicTableSource source = findSource(input); + if (source != null) { + return source; + } + } + return null; + } +}
