This is an automated email from the ASF dual-hosted git repository.
CritasWang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
The following commit(s) were added to refs/heads/master by this push:
new 2bd53be Fix the Flink SQL lookup join for missing timestamps and null
values (#135)
2bd53be is described below
commit 2bd53be3ad8a09cc1ea5219ade48a4f478756eff
Author: ZIHAN DAI <[email protected]>
AuthorDate: Sat Oct 10 19:37:55 2026 +1100
Fix the Flink SQL lookup join for missing timestamps and null values (#135)
* Return null instead of failing when a looked-up IoTDB row has a null value
IoTDBLookupFunction.eval checked each declared column's type against the
per-row Field. When the series has no value at the looked-up timestamp,
IoTDB returns a Field with no data type, Utils.isTypeEqual looks up null in
its type map, and the lookup join fails with a NullPointerException. Rows
with a missing value are normal for non-aligned tree-model series.
Emit null for such a field before the type check, the same way
Utils.getValue already treats a field with no data type. Add unit tests for
the null and fully populated cases.
Signed-off-by: Zihan Dai <[email protected]>
* Emit no row when the looked-up timestamp has no data
When the lookup query returned no row, IoTDBLookupFunction.eval still
collected a row holding the looked-up timestamp and nulls. A lookup join
therefore treated every missing key as a match: an inner join returned
it with a Time_ that IoTDB does not have, and a left join could not tell
it apart from a real row.
Return without collecting anything, so the join itself decides: an inner
join drops the key and a left join pads it with nulls.
Signed-off-by: Zihan Dai <[email protected]>
---------
Signed-off-by: Zihan Dai <[email protected]>
---
connectors/flink-sql-iotdb-connector/pom.xml | 10 ++
.../flink/sql/function/IoTDBLookupFunction.java | 14 +--
.../sql/function/IoTDBLookupFunctionTest.java | 140 +++++++++++++++++++++
3 files changed, 157 insertions(+), 7 deletions(-)
diff --git a/connectors/flink-sql-iotdb-connector/pom.xml
b/connectors/flink-sql-iotdb-connector/pom.xml
index bfe60d0..9e8b80d 100644
--- a/connectors/flink-sql-iotdb-connector/pom.xml
+++ b/connectors/flink-sql-iotdb-connector/pom.xml
@@ -100,6 +100,16 @@
<artifactId>flink-table-common</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git
a/connectors/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java
b/connectors/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java
index 2053846..854d838 100644
---
a/connectors/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java
+++
b/connectors/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunction.java
@@ -111,13 +111,8 @@ public class IoTDBLookupFunction extends
TableFunction<RowData> {
columnNames.remove("Time");
RowRecord rowRecord = dataSet.next();
if (rowRecord == null) {
- ArrayList<Object> values = new ArrayList<>();
- values.add(timestamp);
- for (int i = 0; i < schema.size(); i++) {
- values.add(null);
- }
- GenericRowData rowData = GenericRowData.of(values.toArray());
- collect(rowData);
+ // No row at this timestamp: emit nothing, so that an inner join drops
the key and a left
+ // join pads it with nulls itself.
return;
}
List<Field> fields = rowRecord.getFields();
@@ -130,6 +125,11 @@ public class IoTDBLookupFunction extends
TableFunction<RowData> {
continue;
}
int index = columnNames.indexOf(field.f0);
+ // A series with no value at this timestamp comes back as a field with
no data type.
+ if (fields.get(index).getDataType() == null) {
+ values.add(null);
+ continue;
+ }
DataType flinkType = field.f1;
TSDataType iotdbType = fields.get(index).getDataType();
if (!Utils.isTypeEqual(iotdbType, flinkType)) {
diff --git
a/connectors/flink-sql-iotdb-connector/src/test/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunctionTest.java
b/connectors/flink-sql-iotdb-connector/src/test/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunctionTest.java
new file mode 100644
index 0000000..0b7d70e
--- /dev/null
+++
b/connectors/flink-sql-iotdb-connector/src/test/java/org/apache/iotdb/flink/sql/function/IoTDBLookupFunctionTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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.flink.sql.function;
+
+import org.apache.iotdb.flink.sql.common.Options;
+import org.apache.iotdb.flink.sql.wrapper.SchemaWrapper;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.session.Session;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.Field;
+import org.apache.tsfile.read.common.RowRecord;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class IoTDBLookupFunctionTest {
+
+ private IoTDBLookupFunction function;
+ private Session session;
+ private final List<RowData> collected = new ArrayList<>();
+
+ @Before
+ public void setUp() throws Exception {
+ Configuration options = new Configuration();
+ options.set(Options.SQL, "select ** from root.sg.d1");
+ TableSchema schema =
+ TableSchema.builder()
+ .field("Time_", DataTypes.BIGINT())
+ .field("root.sg.d1.s0", DataTypes.FLOAT())
+ .field("root.sg.d1.s1", DataTypes.FLOAT())
+ .build();
+ function = new IoTDBLookupFunction(options, new SchemaWrapper(schema));
+
+ session = mock(Session.class);
+ java.lang.reflect.Field sessionField =
IoTDBLookupFunction.class.getDeclaredField("session");
+ sessionField.setAccessible(true);
+ sessionField.set(function, session);
+
+ function.setCollector(
+ new Collector<RowData>() {
+ @Override
+ public void collect(RowData record) {
+ collected.add(record);
+ }
+
+ @Override
+ public void close() {}
+ });
+ }
+
+ private void serverReturns(long time, Field... fields) throws Exception {
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ when(dataSet.getColumnNames())
+ .thenReturn(new ArrayList<>(Arrays.asList("Time", "root.sg.d1.s0",
"root.sg.d1.s1")));
+ when(dataSet.next()).thenReturn(new RowRecord(time, new
ArrayList<>(Arrays.asList(fields))));
+ when(session.executeQueryStatement(anyString())).thenReturn(dataSet);
+ }
+
+ private static Field floatField(float value) {
+ Field field = new Field(TSDataType.FLOAT);
+ field.setFloatV(value);
+ return field;
+ }
+
+ @Test
+ public void nullFieldInMatchedRowIsReturnedAsNull() throws Exception {
+ // The series root.sg.d1.s1 has no value at time 3, so the server returns
a null field for it.
+ serverReturns(3L, floatField(23.25f), new Field(null));
+
+ function.eval(3L);
+
+ assertEquals(1, collected.size());
+ RowData row = collected.get(0);
+ assertEquals(3L, row.getLong(0));
+ assertEquals(23.25f, row.getFloat(1), 0f);
+ assertTrue(row.isNullAt(2));
+ }
+
+ @Test
+ public void missingTimestampReturnsNoRow() throws Exception {
+ // Nothing is stored at time 6, so the lookup must not emit a row;
otherwise an inner join
+ // would report a match for a key that does not exist.
+ SessionDataSet dataSet = mock(SessionDataSet.class);
+ when(dataSet.getColumnNames())
+ .thenReturn(new ArrayList<>(Arrays.asList("Time", "root.sg.d1.s0",
"root.sg.d1.s1")));
+ when(dataSet.next()).thenReturn(null);
+ when(session.executeQueryStatement(anyString())).thenReturn(dataSet);
+
+ function.eval(6L);
+
+ assertTrue(collected.isEmpty());
+ }
+
+ @Test
+ public void fullyPopulatedRowIsReturnedUnchanged() throws Exception {
+ serverReturns(4L, floatField(24.25f), floatField(34.75f));
+
+ function.eval(4L);
+
+ assertEquals(1, collected.size());
+ RowData row = collected.get(0);
+ assertEquals(4L, row.getLong(0));
+ assertEquals(24.25f, row.getFloat(1), 0f);
+ assertFalse(row.isNullAt(2));
+ assertEquals(34.75f, row.getFloat(2), 0f);
+ }
+}
