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 7b194bc feat: support lookup
7b194bc is described below
commit 7b194bcecf4e435211be0291cc91680062296aca
Author: shuwenwei <[email protected]>
AuthorDate: Mon Sep 21 17:41:16 2026 +0800
feat: support lookup
---
.../iotdb/relational/flink/cfg/IoTDBOptions.java | 36 ++++++
.../source/lookup/IoTDBAsyncLookupFunction.java | 106 ++++++++++++++++
.../flink/source/lookup/IoTDBLookupFunction.java | 70 +++++++++++
.../flink/source/lookup/IoTDBLookupReader.java | 138 +++++++++++++++++++++
.../source/lookup/IoTDBRuntimeLiteralUtils.java | 108 ++++++++++++++++
.../lookup/IoTDBAsyncLookupFunctionTest.java | 54 ++++++++
.../flink/source/lookup/IoTDBLookupReaderTest.java | 120 ++++++++++++++++++
.../table/IoTDBRelationalDynamicTableFactory.java | 14 +--
.../table/IoTDBRelationalDynamicTableSource.java | 38 +++++-
.../flink/IoTDBRelationalLocalQueryManual.java | 107 ++++++++++++----
10 files changed, 760 insertions(+), 31 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 fcd21e5..0bc6cf9 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
@@ -65,6 +65,12 @@ public class IoTDBOptions implements Serializable {
public static final ConfigOption<String> ATTRIBUTE_COLUMNS =
ConfigOptions.key("attribute-columns").stringType().defaultValue("");
+ public static final ConfigOption<Boolean> LOOKUP_ASYNC =
+ ConfigOptions.key("lookup.async").booleanType().defaultValue(false);
+
+ public static final ConfigOption<Integer> LOOKUP_THREAD_SIZE =
+ ConfigOptions.key("lookup.thread-size").intType().defaultValue(5);
+
private final List<String> nodeUrls;
private final String username;
private final String password;
@@ -74,6 +80,8 @@ public class IoTDBOptions implements Serializable {
private final String timeColumn;
private final List<String> tagColumns;
private final List<String> attributeColumns;
+ private final boolean lookupAsync;
+ private final int lookupThreadSize;
private IoTDBOptions(Builder builder) {
this.nodeUrls = builder.nodeUrls;
@@ -85,6 +93,8 @@ public class IoTDBOptions implements Serializable {
this.timeColumn = builder.timeColumn;
this.tagColumns = builder.tagColumns;
this.attributeColumns = builder.attributeColumns;
+ this.lookupAsync = builder.lookupAsync;
+ this.lookupThreadSize = builder.lookupThreadSize;
}
/**
@@ -150,6 +160,20 @@ public class IoTDBOptions implements Serializable {
return attributeColumns;
}
+ /**
+ * @return whether an asynchronous lookup function is used for lookup joins.
+ */
+ public boolean isLookupAsync() {
+ return lookupAsync;
+ }
+
+ /**
+ * @return the number of concurrent lookup query threads, also the lookup
session pool size.
+ */
+ public int getLookupThreadSize() {
+ return lookupThreadSize;
+ }
+
/**
* @return a new builder
*/
@@ -169,6 +193,8 @@ public class IoTDBOptions implements Serializable {
private String timeColumn;
private List<String> tagColumns = Collections.emptyList();
private List<String> attributeColumns = Collections.emptyList();
+ private boolean lookupAsync = false;
+ private int lookupThreadSize = 5;
public Builder withNodeUrls(List<String> nodeUrls) {
this.nodeUrls = nodeUrls;
@@ -215,6 +241,16 @@ public class IoTDBOptions implements Serializable {
return this;
}
+ public Builder withLookupAsync(boolean lookupAsync) {
+ this.lookupAsync = lookupAsync;
+ return this;
+ }
+
+ public Builder withLookupThreadSize(int lookupThreadSize) {
+ this.lookupThreadSize = lookupThreadSize;
+ return this;
+ }
+
/**
* @return the built options
*/
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBAsyncLookupFunction.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBAsyncLookupFunction.java
new file mode 100644
index 0000000..5b7e223
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBAsyncLookupFunction.java
@@ -0,0 +1,106 @@
+/*
+ * 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.iotdb.relational.flink.cfg.IoTDBOptions;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.functions.AsyncLookupFunction;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.types.DataType;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Asynchronous lookup function that queries an IoTDB table-model table by the
join keys.
+ *
+ * <p>{@code ITableSession} only offers blocking queries, so the blocking
{@link IoTDBLookupReader}
+ * is executed on a bounded thread pool and results are delivered through
{@link CompletableFuture}.
+ */
+public class IoTDBAsyncLookupFunction extends AsyncLookupFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IoTDBOptions options;
+ private final DataType rowDataType;
+ private final int[] keyIndices;
+ private final int threadSize;
+
+ private transient ExecutorService executor;
+ private transient IoTDBLookupReader reader;
+
+ public IoTDBAsyncLookupFunction(
+ IoTDBOptions options, DataType rowDataType, int[] keyIndices, int
threadSize) {
+ this.options = options;
+ this.rowDataType = rowDataType;
+ this.keyIndices = keyIndices.clone();
+ this.threadSize = threadSize;
+ }
+
+ @Override
+ public void open(FunctionContext context) {
+ reader = new IoTDBLookupReader(options, rowDataType, keyIndices);
+ reader.open();
+ executor = Executors.newFixedThreadPool(Math.max(1, threadSize),
createThreadFactory());
+ }
+
+ @Override
+ public CompletableFuture<Collection<RowData>> asyncLookup(RowData keyRow) {
+ if (keyRow == null) {
+ return CompletableFuture.completedFuture(Collections.emptyList());
+ }
+ return CompletableFuture.<Collection<RowData>>supplyAsync(
+ () -> {
+ try {
+ return reader.get(keyRow);
+ } catch (IOException e) {
+ throw new CompletionException(e);
+ }
+ },
+ executor);
+ }
+
+ @Override
+ public void close() {
+ if (reader != null) {
+ reader.close();
+ }
+ if (executor != null) {
+ executor.shutdown();
+ }
+ }
+
+ private static ThreadFactory createThreadFactory() {
+ AtomicInteger counter = new AtomicInteger();
+ return runnable -> {
+ Thread thread = new Thread(runnable, "iotdb-lookup-" +
counter.incrementAndGet());
+ thread.setDaemon(true);
+ return thread;
+ };
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupFunction.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupFunction.java
new file mode 100644
index 0000000..da07a69
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupFunction.java
@@ -0,0 +1,70 @@
+/*
+ * 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.iotdb.relational.flink.cfg.IoTDBOptions;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.LookupFunction;
+import org.apache.flink.table.types.DataType;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+
+/** Synchronous lookup function that queries an IoTDB table-model table by the
join keys. */
+public class IoTDBLookupFunction extends LookupFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IoTDBOptions options;
+ private final DataType rowDataType;
+ private final int[] keyIndices;
+
+ private transient IoTDBLookupReader reader;
+
+ public IoTDBLookupFunction(IoTDBOptions options, DataType rowDataType, int[]
keyIndices) {
+ this.options = options;
+ this.rowDataType = rowDataType;
+ this.keyIndices = keyIndices.clone();
+ }
+
+ @Override
+ public void open(FunctionContext context) {
+ reader = new IoTDBLookupReader(options, rowDataType, keyIndices);
+ reader.open();
+ }
+
+ @Override
+ public Collection<RowData> lookup(RowData keyRow) throws IOException {
+ if (keyRow == null) {
+ return Collections.emptyList();
+ }
+ return reader.get(keyRow);
+ }
+
+ @Override
+ public void close() {
+ if (reader != null) {
+ reader.close();
+ }
+ }
+}
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
new file mode 100644
index 0000000..cd4e6ad
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReader.java
@@ -0,0 +1,138 @@
+/*
+ * 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.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.pool.ITableSessionPool;
+import org.apache.iotdb.relational.flink.cfg.IoTDBOptions;
+import
org.apache.iotdb.relational.flink.source.deserializer.RowDataDeserializationSchema;
+import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
+import org.apache.iotdb.session.pool.TableSessionPoolBuilder;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Blocking lookup reader for an IoTDB table-model table.
+ *
+ * <p>It builds a point query from the lookup keys and converts the result
rows back to {@link
+ * RowData}. The instance is not thread-safe until {@link #open()} has been
called; afterwards the
+ * underlying session pool is safe to use from multiple threads.
+ */
+public class IoTDBLookupReader implements AutoCloseable {
+
+ private final IoTDBOptions options;
+ private final DataType rowDataType;
+ private final int[] keyIndices;
+ private final List<String> fieldNames;
+ private final List<DataType> fieldTypes;
+ private final RowDataDeserializationSchema deserializer;
+
+ private volatile ITableSessionPool sessionPool;
+
+ public IoTDBLookupReader(IoTDBOptions options, DataType rowDataType, int[]
keyIndices) {
+ this.options = options;
+ this.rowDataType = rowDataType;
+ this.keyIndices = keyIndices.clone();
+ this.fieldNames = DataType.getFieldNames(rowDataType);
+ this.fieldTypes = DataType.getFieldDataTypes(rowDataType);
+ this.deserializer = new RowDataDeserializationSchema(rowDataType);
+ }
+
+ /** Opens the IoTDB session pool. The database is taken from the connector
options. */
+ public synchronized void open() {
+ if (sessionPool != null) {
+ return;
+ }
+ TableSessionPoolBuilder builder =
+ new TableSessionPoolBuilder()
+ .nodeUrls(options.getNodeUrls())
+ .user(options.getUsername())
+ .password(options.getPassword());
+ if (options.getDatabase() != null) {
+ builder.database(options.getDatabase());
+ }
+ sessionPool = builder.build();
+ }
+
+ /**
+ * Runs the lookup query for the given key row and returns the matching rows.
+ *
+ * @param keyRow row whose fields are the lookup keys, in {@link
#keyIndices} order
+ * @return matching rows, or an empty list when a key is {@code null}
+ */
+ public List<RowData> get(RowData keyRow) throws IOException {
+ String sql = buildLookupSql(keyRow);
+ if (sql == null) {
+ return Collections.emptyList();
+ }
+
+ ITableSessionPool pool = sessionPool;
+ if (pool == null) {
+ throw new IOException("IoTDB lookup reader is not open.");
+ }
+
+ try (ITableSession session = pool.getSession();
+ SessionDataSet dataSet = session.executeQueryStatement(sql)) {
+ List<RowData> rows = new ArrayList<>();
+ SessionDataSet.DataIterator iterator = dataSet.iterator();
+ while (iterator.next()) {
+ rows.add(deserializer.deserialize(iterator));
+ }
+ return rows;
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException("Failed to execute IoTDB lookup query: " + sql, e);
+ }
+ }
+
+ /** Builds the lookup SQL, or {@code null} if a key value cannot be
represented. */
+ String buildLookupSql(RowData keyRow) {
+ 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));
+ if (literal == null) {
+ return null;
+ }
+ predicates.add(IoTDBUtils.quoteIdentifier(fieldNames.get(fieldIndex)) +
" = " + literal);
+ }
+ return IoTDBUtils.buildSelectQuery(options.getTable(), rowDataType,
predicates, -1);
+ }
+
+ @Override
+ public synchronized void close() {
+ if (sessionPool != null) {
+ try {
+ sessionPool.close();
+ } finally {
+ sessionPool = 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
new file mode 100644
index 0000000..9993c91
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBRuntimeLiteralUtils.java
@@ -0,0 +1,108 @@
+/*
+ * 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/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBAsyncLookupFunctionTest.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBAsyncLookupFunctionTest.java
new file mode 100644
index 0000000..2adc756
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBAsyncLookupFunctionTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.iotdb.relational.flink.cfg.IoTDBOptions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.junit.Assert.assertTrue;
+
+/** Unit tests for {@link IoTDBAsyncLookupFunction}. */
+public class IoTDBAsyncLookupFunctionTest {
+
+ private static final IoTDBOptions OPTIONS =
+ IoTDBOptions.builder()
+ .withNodeUrls(Collections.singletonList("127.0.0.1:6667"))
+ .withDatabase("test")
+ .withTable("sensor")
+ .build();
+
+ @Test
+ public void testNullKeyReturnsEmptyResult() throws Exception {
+ DataType rowType = DataTypes.ROW(DataTypes.FIELD("device_id",
DataTypes.STRING()));
+ IoTDBAsyncLookupFunction function =
+ new IoTDBAsyncLookupFunction(OPTIONS, rowType, new int[] {0}, 2);
+
+ Collection<RowData> result = function.asyncLookup(null).get();
+
+ assertTrue(result.isEmpty());
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReaderTest.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReaderTest.java
new file mode 100644
index 0000000..f2b6134
--- /dev/null
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/lookup/IoTDBLookupReaderTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.iotdb.relational.flink.cfg.IoTDBOptions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
+import org.junit.Test;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+/** Unit tests for the lookup SQL rendering of {@link IoTDBLookupReader}. */
+public class IoTDBLookupReaderTest {
+
+ private static final IoTDBOptions OPTIONS =
+ IoTDBOptions.builder()
+ .withNodeUrls(Collections.singletonList("127.0.0.1:6667"))
+ .withDatabase("test")
+ .withTable("sensor")
+ .build();
+
+ @Test
+ public void testStringAndTimestampKeys() {
+ DataType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD("time", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD("device_id", DataTypes.STRING()),
+ DataTypes.FIELD("temperature", DataTypes.DOUBLE()));
+ IoTDBLookupReader reader = new IoTDBLookupReader(OPTIONS, rowType, new
int[] {1, 0});
+
+ GenericRowData keyRow =
+ GenericRowData.of(
+ StringData.fromString("d1"),
+ TimestampData.fromLocalDateTime(LocalDateTime.of(2024, 1, 1, 12,
30)));
+
+ assertEquals(
+ "SELECT \"time\", \"device_id\", \"temperature\" FROM \"sensor\" "
+ + "WHERE \"device_id\" = 'd1' AND \"time\" =
CAST('2024-01-01T12:30' AS TIMESTAMP)",
+ reader.buildLookupSql(keyRow));
+ }
+
+ @Test
+ public void testStringKeyIsEscaped() {
+ DataType rowType = DataTypes.ROW(DataTypes.FIELD("device_id",
DataTypes.STRING()));
+ IoTDBLookupReader reader = new IoTDBLookupReader(OPTIONS, rowType, new
int[] {0});
+
+ GenericRowData keyRow =
GenericRowData.of(StringData.fromString("O'Brien"));
+
+ assertEquals(
+ "SELECT \"device_id\" FROM \"sensor\" WHERE \"device_id\" =
'O''Brien'",
+ reader.buildLookupSql(keyRow));
+ }
+
+ @Test
+ public void testDateKey() {
+ DataType rowType = DataTypes.ROW(DataTypes.FIELD("day", DataTypes.DATE()));
+ IoTDBLookupReader reader = new IoTDBLookupReader(OPTIONS, rowType, new
int[] {0});
+ int epochDay = 19000;
+
+ assertEquals(
+ "SELECT \"day\" FROM \"sensor\" "
+ + "WHERE \"day\" = CAST('"
+ + LocalDate.ofEpochDay(epochDay)
+ + "' AS DATE)",
+ reader.buildLookupSql(GenericRowData.of(epochDay)));
+ }
+
+ @Test
+ public void testCompositeNumericKey() {
+ DataType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD("code", DataTypes.INT()), DataTypes.FIELD("value",
DataTypes.BIGINT()));
+ IoTDBLookupReader reader = new IoTDBLookupReader(OPTIONS, rowType, new
int[] {0, 1});
+
+ assertEquals(
+ "SELECT \"code\", \"value\" FROM \"sensor\" WHERE \"code\" = 1 AND
\"value\" = 2",
+ reader.buildLookupSql(GenericRowData.of(1, 2L)));
+ }
+
+ @Test
+ public void testNullKeyProducesNoQuery() {
+ DataType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD("time", DataTypes.TIMESTAMP(3)),
+ DataTypes.FIELD("device_id", DataTypes.STRING()));
+ IoTDBLookupReader reader = new IoTDBLookupReader(OPTIONS, rowType, new
int[] {1, 0});
+
+ GenericRowData keyRow =
+ GenericRowData.of(
+ null, TimestampData.fromLocalDateTime(LocalDateTime.of(2024, 1, 1,
12, 30)));
+
+ assertNull(reader.buildLookupSql(keyRow));
+ }
+}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
index 5c45b73..e755082 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableFactory.java
@@ -69,10 +69,7 @@ public class IoTDBRelationalDynamicTableFactory
@Override
public Set<ConfigOption<?>> requiredOptions() {
return new HashSet<>(
- Arrays.asList(
- IoTDBOptions.NODE_URLS,
- IoTDBOptions.DATABASE,
- IoTDBOptions.TABLE));
+ Arrays.asList(IoTDBOptions.NODE_URLS, IoTDBOptions.DATABASE,
IoTDBOptions.TABLE));
}
@Override
@@ -83,7 +80,9 @@ public class IoTDBRelationalDynamicTableFactory
IoTDBOptions.PASSWORD,
IoTDBOptions.TIME_COLUMN,
IoTDBOptions.TAG_COLUMNS,
- IoTDBOptions.ATTRIBUTE_COLUMNS));
+ IoTDBOptions.ATTRIBUTE_COLUMNS,
+ IoTDBOptions.LOOKUP_ASYNC,
+ IoTDBOptions.LOOKUP_THREAD_SIZE));
}
private static IoTDBOptions toOptions(ReadableConfig config) {
@@ -95,8 +94,9 @@ public class IoTDBRelationalDynamicTableFactory
.withTable(config.get(IoTDBOptions.TABLE))
.withTimeColumn(config.get(IoTDBOptions.TIME_COLUMN))
.withTagColumns(parseColumnNames(config.get(IoTDBOptions.TAG_COLUMNS)))
- .withAttributeColumns(
- parseColumnNames(config.get(IoTDBOptions.ATTRIBUTE_COLUMNS)))
+
.withAttributeColumns(parseColumnNames(config.get(IoTDBOptions.ATTRIBUTE_COLUMNS)))
+ .withLookupAsync(config.get(IoTDBOptions.LOOKUP_ASYNC))
+ .withLookupThreadSize(config.get(IoTDBOptions.LOOKUP_THREAD_SIZE))
.build();
}
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 f9d43c9..7adcb7e 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
@@ -22,20 +22,26 @@ package org.apache.iotdb.relational.flink.table;
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.lookup.IoTDBAsyncLookupFunction;
+import org.apache.iotdb.relational.flink.source.lookup.IoTDBLookupFunction;
import org.apache.iotdb.relational.flink.source.pushdown.AggregateSpec;
import
org.apache.iotdb.relational.flink.source.pushdown.IoTDBAggregatePushDownUtils;
import
org.apache.iotdb.relational.flink.source.pushdown.IoTDBExpressionVisitor;
import org.apache.iotdb.relational.flink.utils.IoTDBUtils;
+import org.apache.flink.table.api.TableException;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceProvider;
import
org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown;
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.connector.source.lookup.AsyncLookupFunctionProvider;
+import org.apache.flink.table.connector.source.lookup.LookupFunctionProvider;
import org.apache.flink.table.expressions.AggregateExpression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.types.DataType;
@@ -47,11 +53,12 @@ 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 is supported
for top-level fields.
- * Lookup reads are intentionally disabled until their runtime behavior is
implemented.
+ * <p>Scan reads are fully implemented. Lookup reads are declared through
{@link LookupTableSource}
+ * but their runtime behavior is still a stub.
*/
public class IoTDBRelationalDynamicTableSource
implements ScanTableSource,
+ LookupTableSource,
SupportsFilterPushDown,
SupportsLimitPushDown,
SupportsProjectionPushDown,
@@ -87,6 +94,33 @@ public class IoTDBRelationalDynamicTableSource
aggregateSpec));
}
+ @Override
+ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext
lookupContext) {
+ // Lookup key indices refer to the scan's current row type, i.e. after any
projection that has
+ // already been pushed into this source.
+ DataType lookupRowDataType = physicalRowDataType;
+ List<String> fieldNames = DataType.getFieldNames(lookupRowDataType);
+ int[][] keys = lookupContext.getKeys();
+ int[] keyIndices = new int[keys.length];
+ for (int i = 0; i < keys.length; i++) {
+ if (keys[i] == null || keys[i].length != 1) {
+ throw new TableException("IoTDB lookup supports only top-level lookup
keys.");
+ }
+ int keyIndex = keys[i][0];
+ if (keyIndex < 0 || keyIndex >= fieldNames.size()) {
+ throw new TableException("Invalid IoTDB lookup key index: " +
keyIndex);
+ }
+ keyIndices[i] = keyIndex;
+ }
+ if (options.isLookupAsync()) {
+ return AsyncLookupFunctionProvider.of(
+ new IoTDBAsyncLookupFunction(
+ options, lookupRowDataType, keyIndices,
options.getLookupThreadSize()));
+ }
+ return LookupFunctionProvider.of(
+ new IoTDBLookupFunction(options, lookupRowDataType, keyIndices));
+ }
+
@Override
public boolean supportsNestedProjection() {
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 02b4b5e..68c6712 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
@@ -19,9 +19,11 @@
package org.apache.iotdb.relational.flink;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
@@ -30,7 +32,7 @@ import org.apache.flink.util.CloseableIterator;
* committed as a production test.
*
* <p>The flow is write first and then query: rows are inserted through the
connector and read back
- * with a SELECT.
+ * with a SELECT, followed by a lookup join (both sync and async) executed as
Flink SQL.
*
* <p>Run with system properties such as:
*
@@ -59,24 +61,7 @@ public class IoTDBRelationalLocalQueryManual {
tableEnvironment.executeSql("DROP TABLE IF EXISTS iotdb_table");
// Replace these columns with the actual columns and categories in the
local IoTDB table.
- String ddl =
- String.format(
- "CREATE TABLE iotdb_table (\n"
- + " `time` TIMESTAMP(3),\n"
- + " `device_id` STRING,\n"
- + " `temperature` DOUBLE\n"
- + ") WITH (\n"
- + " 'connector' = 'iotdb-relational',\n"
- + " 'nodeUrls' = '%s',\n"
- + " 'user' = '%s',\n"
- + " 'password' = '%s',\n"
- + " 'database' = '%s',\n"
- + " 'table' = '%s',\n"
- + " 'time-column' = 'time',\n"
- + " 'tag-columns' = 'device_id'\n"
- + ")",
- nodeUrls, user, password, database, table);
-
+ String ddl = buildDdl("iotdb_table", nodeUrls, user, password, database,
table, null);
System.out.println("DDL:\n" + ddl);
tableEnvironment.executeSql(ddl);
@@ -96,9 +81,87 @@ public class IoTDBRelationalLocalQueryManual {
run(
tableEnvironment,
"SELECT device_id, temperature FROM iotdb_table WHERE temperature > 0
LIMIT 1");
- run(
- tableEnvironment,
- "SELECT temperature FROM iotdb_table WHERE temperature + 1 > 21 LIMIT
5");
+ run(tableEnvironment, "SELECT temperature FROM iotdb_table WHERE
temperature + 1 > 21 LIMIT 5");
+
+ // 3) Lookup: run a temporal join through SQL for both sync and async
lookup.
+ verifyLookupJoin(nodeUrls, user, password, database, table, false);
+ verifyLookupJoin(nodeUrls, user, password, database, table, true);
+ }
+
+ private static void verifyLookupJoin(
+ String nodeUrls, String user, String password, String database, String
table, boolean async)
+ throws Exception {
+ StreamExecutionEnvironment environment =
StreamExecutionEnvironment.getExecutionEnvironment();
+ environment.setParallelism(1);
+ StreamTableEnvironment tableEnvironment =
StreamTableEnvironment.create(environment);
+
+ String flinkTable = "iotdb_lookup_" + (async ? "async" : "sync");
+ tableEnvironment.executeSql("DROP TABLE IF EXISTS " + flinkTable);
+ tableEnvironment.executeSql(
+ buildDdl(flinkTable, nodeUrls, user, password, database, table,
async));
+
+ tableEnvironment.executeSql(
+ "CREATE TEMPORARY VIEW probe AS "
+ + "SELECT device_id, PROCTIME() AS proc_time FROM "
+ + flinkTable);
+
+ System.out.println("\n=== LOOKUP " + (async ? "ASYNC" : "SYNC") + " ===");
+ String sql =
+ "SELECT p.device_id, d.temperature FROM probe AS p "
+ + "JOIN "
+ + flinkTable
+ + " FOR SYSTEM_TIME AS OF p.proc_time AS d "
+ + "ON p.device_id = d.device_id";
+ System.out.println(sql);
+
+ TableResult result = tableEnvironment.executeSql(sql);
+ try (CloseableIterator<Row> iterator = result.collect()) {
+ while (iterator.hasNext()) {
+ System.out.println(iterator.next());
+ }
+ }
+ }
+
+ private static String buildDdl(
+ String flinkTable,
+ String nodeUrls,
+ String user,
+ String password,
+ String database,
+ String table,
+ Boolean async) {
+ StringBuilder ddl =
+ new StringBuilder()
+ .append("CREATE TABLE ")
+ .append(flinkTable)
+ .append(" (\n")
+ .append(" `time` TIMESTAMP(3),\n")
+ .append(" `device_id` STRING,\n")
+ .append(" `temperature` DOUBLE\n")
+ .append(") WITH (\n")
+ .append(" 'connector' = 'iotdb-relational',\n")
+ .append(" 'nodeUrls' = '")
+ .append(nodeUrls)
+ .append("',\n")
+ .append(" 'user' = '")
+ .append(user)
+ .append("',\n")
+ .append(" 'password' = '")
+ .append(password)
+ .append("',\n")
+ .append(" 'database' = '")
+ .append(database)
+ .append("',\n")
+ .append(" 'table' = '")
+ .append(table)
+ .append("',\n")
+ .append(" 'time-column' = 'time',\n")
+ .append(" 'tag-columns' = 'device_id'");
+ if (async != null) {
+ ddl.append(",\n 'lookup.async' = '").append(async).append("'");
+ }
+ ddl.append("\n)");
+ return ddl.toString();
}
private static void execute(TableEnvironment tableEnvironment, String sql)
throws Exception {
