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 dd114b60bf637bb654920f2d9642167e16d70f36 Author: shuwenwei <[email protected]> AuthorDate: Fri Sep 18 08:39:07 2026 +0800 source --- .../relational/flink/utils/IoTDBSQLBuilder.java | 68 ++++++++++++++++++++++ .../iotdb/relational/flink/source/IoTDBSource.java | 16 ++++- .../flink/source/IoTDBSourceEnumerator.java | 39 ++++--------- .../table/IoTDBRelationalDynamicTableSource.java | 25 ++++++-- 4 files changed, 113 insertions(+), 35 deletions(-) diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBSQLBuilder.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBSQLBuilder.java new file mode 100644 index 0000000..12ce03c --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/utils/IoTDBSQLBuilder.java @@ -0,0 +1,68 @@ +/* + * 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.types.DataType; + +import java.util.List; + +/** Builds reusable IoTDB SQL statements for the relational table connector. */ +public final class IoTDBSQLBuilder { + + private IoTDBSQLBuilder() {} + + /** + * Builds a bounded table-model scan query. + * + * @param table IoTDB table name + * @param rowDataType projected Flink row type whose field names define the SELECT list + * @param filterQueries already-rendered IoTDB predicate fragments + * @param limit maximum number of rows, or a negative value for no limit + * @return IoTDB SELECT SQL + */ + public static String buildSelectQuery( + String table, DataType rowDataType, List<String> filterQueries, long limit) { + List<String> fieldNames = DataType.getFieldNames(rowDataType); + if (fieldNames.isEmpty()) { + throw new IllegalArgumentException("IoTDB source requires at least one selected column."); + } + + StringBuilder columns = new StringBuilder(); + for (String fieldName : fieldNames) { + if (columns.length() > 0) { + columns.append(", "); + } + columns.append(IoTDBIdentifierUtils.quoteIdentifier(fieldName)); + } + + StringBuilder sql = + new StringBuilder("SELECT ") + .append(columns) + .append(" FROM ") + .append(IoTDBIdentifierUtils.quoteIdentifier(table)); + if (filterQueries != null && !filterQueries.isEmpty()) { + sql.append(" WHERE ").append(String.join(" AND ", filterQueries)); + } + if (limit >= 0) { + sql.append(" LIMIT ").append(limit); + } + return sql.toString(); + } +} diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSource.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSource.java index 0c7e10e..ecaeb0a 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSource.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSource.java @@ -53,16 +53,27 @@ public class IoTDBSource<OUT> implements Source<OUT, IoTDBSourceSplit, IoTDBSour private final DataType rowDataType; private final IoTDBDeserializationSchema<OUT> deserializer; private final List<String> filterQueries; + private final long limit; public IoTDBSource( IoTDBRelationalOptions options, DataType rowDataType, IoTDBDeserializationSchema<OUT> deserializer, List<String> filterQueries) { + this(options, rowDataType, deserializer, filterQueries, -1L); + } + + public IoTDBSource( + IoTDBRelationalOptions options, + DataType rowDataType, + IoTDBDeserializationSchema<OUT> deserializer, + List<String> filterQueries, + long limit) { this.options = options; this.rowDataType = rowDataType; this.deserializer = deserializer; this.filterQueries = filterQueries == null ? new ArrayList<>() : new ArrayList<>(filterQueries); + this.limit = limit; } @Override @@ -78,13 +89,14 @@ public class IoTDBSource<OUT> implements Source<OUT, IoTDBSourceSplit, IoTDBSour @Override public SplitEnumerator<IoTDBSourceSplit, IoTDBSourceEnumeratorState> createEnumerator( SplitEnumeratorContext<IoTDBSourceSplit> enumContext) { - return new IoTDBSourceEnumerator(enumContext, options, rowDataType, filterQueries); + return new IoTDBSourceEnumerator(enumContext, options, rowDataType, filterQueries, limit); } @Override public SplitEnumerator<IoTDBSourceSplit, IoTDBSourceEnumeratorState> restoreEnumerator( SplitEnumeratorContext<IoTDBSourceSplit> enumContext, IoTDBSourceEnumeratorState checkpoint) { - return new IoTDBSourceEnumerator(enumContext, options, rowDataType, filterQueries, checkpoint); + return new IoTDBSourceEnumerator( + enumContext, options, rowDataType, filterQueries, limit, checkpoint); } @Override diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceEnumerator.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceEnumerator.java index 00cb723..ef0dd5d 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceEnumerator.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/IoTDBSourceEnumerator.java @@ -22,7 +22,7 @@ package org.apache.iotdb.relational.flink.source; import org.apache.iotdb.relational.flink.cfg.IoTDBRelationalOptions; import org.apache.iotdb.relational.flink.source.enumerator.IoTDBSourceEnumeratorState; import org.apache.iotdb.relational.flink.source.split.IoTDBSourceSplit; -import org.apache.iotdb.relational.flink.utils.IoTDBIdentifierUtils; +import org.apache.iotdb.relational.flink.utils.IoTDBSQLBuilder; import org.apache.flink.api.connector.source.SplitEnumerator; import org.apache.flink.api.connector.source.SplitEnumeratorContext; @@ -47,6 +47,7 @@ public class IoTDBSourceEnumerator private final IoTDBRelationalOptions options; private final DataType rowDataType; private final List<String> filterQueries; + private final long limit; private final Deque<IoTDBSourceSplit> pendingSplits = new ArrayDeque<>(); private final Deque<Integer> readersAwaitingSplit = new ArrayDeque<>(); private final Set<Integer> assignedReaders = new HashSet<>(); @@ -58,8 +59,9 @@ public class IoTDBSourceEnumerator SplitEnumeratorContext<IoTDBSourceSplit> context, IoTDBRelationalOptions options, DataType rowDataType, - List<String> filterQueries) { - this(context, options, rowDataType, filterQueries, null); + List<String> filterQueries, + long limit) { + this(context, options, rowDataType, filterQueries, limit, null); } public IoTDBSourceEnumerator( @@ -67,11 +69,13 @@ public class IoTDBSourceEnumerator IoTDBRelationalOptions options, DataType rowDataType, List<String> filterQueries, + long limit, @Nullable IoTDBSourceEnumeratorState checkpoint) { this.context = context; this.options = options; this.rowDataType = rowDataType; this.filterQueries = filterQueries == null ? new ArrayList<>() : new ArrayList<>(filterQueries); + this.limit = limit; if (checkpoint != null) { pendingSplits.addAll(checkpoint.getRemainingSplits()); allSplitsCreated = true; @@ -140,31 +144,10 @@ public class IoTDBSourceEnumerator } private IoTDBSourceSplit createSingleSplit() { - List<String> fieldNames = DataType.getFieldNames(rowDataType); - StringBuilder columns = new StringBuilder(); - for (String fieldName : fieldNames) { - if (columns.length() > 0) { - columns.append(", "); - } - columns.append(quoteIdentifier(fieldName)); - } - if (columns.length() == 0) { - throw new IllegalArgumentException("IoTDB source requires at least one selected column."); - } - String splitId = UUID.randomUUID().toString(); - StringBuilder sql = - new StringBuilder("SELECT ") - .append(columns) - .append(" FROM ") - .append(quoteIdentifier(options.getTable())); - if (!filterQueries.isEmpty()) { - sql.append(" WHERE ").append(String.join(" AND ", filterQueries)); - } - return new IoTDBSourceSplit(splitId, options.getDatabase(), options.getTable(), sql.toString()); - } - - private static String quoteIdentifier(String identifier) { - return IoTDBIdentifierUtils.quoteIdentifier(identifier); + String sql = + IoTDBSQLBuilder.buildSelectQuery( + options.getTable(), rowDataType, filterQueries, limit); + return new IoTDBSourceSplit(splitId, options.getDatabase(), options.getTable(), sql); } } 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 a81291d..87d359d 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,11 +26,13 @@ import org.apache.iotdb.relational.flink.source.pushdown.IoTDBExpressionVisitor; 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; import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown; +import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown; import org.apache.flink.table.expressions.ResolvedExpression; import org.apache.flink.table.types.DataType; @@ -41,11 +43,14 @@ import java.util.List; /** * Dynamic table source of the IoTDB relational (table model) Flink connector. * - * <p>Only scan reads are exposed for now. Projection pushdown and lookup reads are intentionally - * disabled until their runtime behavior is implemented. + * <p>Only scan reads are exposed for now. Projection pushdown is supported for top-level fields. + * Lookup reads are intentionally disabled until their runtime behavior is implemented. */ public class IoTDBRelationalDynamicTableSource - implements ScanTableSource, SupportsFilterPushDown, SupportsLimitPushDown { + implements ScanTableSource, + SupportsFilterPushDown, + SupportsLimitPushDown, + SupportsProjectionPushDown { private final IoTDBRelationalOptions options; private final ResolvedSchema schema; @@ -71,7 +76,18 @@ public class IoTDBRelationalDynamicTableSource options, physicalRowDataType, new RowDataDeserializationSchema(physicalRowDataType), - resolvedFilterQueries)); + resolvedFilterQueries, + limit)); + } + + @Override + public boolean supportsNestedProjection() { + return false; + } + + @Override + public void applyProjection(int[][] projectedFields, DataType producedDataType) { + this.physicalRowDataType = Projection.of(projectedFields).project(physicalRowDataType); } @Override @@ -97,7 +113,6 @@ public class IoTDBRelationalDynamicTableSource @Override public void applyLimit(long limit) { - // TODO: push this limit into the single-split SQL when the optimization is enabled. this.limit = limit; }
