From e742d0e591ba2802e462684514d25e020667676d Mon Sep 17 00:00:00 2001 From: Zihan Dai <1436286758@qq.com> Date: Tue, 20 Oct 2026 22:02:49 +1100 Subject: [PATCH] Fix timestamp filters and time zone handling in the table connector Timestamp literals were rendered as CAST( AS TIMESTAMP). IoTDB 2.0.11 fails a predicate on the time column written that way with error 301 (GenericLiteral cannot be cast to LongLiteral), so every scan with a pushed-down time filter and every lookup keyed on time failed. Render the plain number instead, which IoTDB accepts for the time column and for TIMESTAMP fields. The sink converted TimestampData with toTimestamp() and the source with fromTimestamp(), which both go through LocalDateTime in the JVM default time zone. On a non-UTC JVM the connector wrote times shifted by the zone offset and read them back shifted, while the filter and lookup literals use the epoch value, so they no longer matched the stored data. Convert through Instant on both sides. Signed-off-by: Zihan Dai <1436286758@qq.com> --- .../flink/sink/RowDataSinkDataConverter.java | 5 +- .../common/RowDataDeserializationSchema.java | 3 +- .../relational/flink/utils/IoTDBUtils.java | 4 +- .../sink/RowDataSinkDataConverterTest.java | 15 +++ .../RowDataDeserializationSchemaTest.java | 119 ++++++++++++++++++ .../source/lookup/IoTDBLookupReaderTest.java | 2 +- 6 files changed, 144 insertions(+), 4 deletions(-) create mode 100644 connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchemaTest.java diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java index 976afdd3..b5a887e9 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverter.java @@ -130,7 +130,10 @@ public LocalDate getDate(int columnIndex) { @Override public Timestamp getTimestamp(int columnIndex) { - return current.getTimestamp(columnIndex, timestampPrecisions[columnIndex]).toTimestamp(); + // toTimestamp() goes through LocalDateTime and the JVM default time zone; toInstant() keeps + // the epoch value, which is what the IoTDB time column stores. + return Timestamp.from( + current.getTimestamp(columnIndex, timestampPrecisions[columnIndex]).toInstant()); } } } diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchema.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchema.java index 68c7384e..5e5da5e1 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchema.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchema.java @@ -130,7 +130,8 @@ private static Object readField(IoTDBDataIterator iterator, int columnIndex, Dat return (int) iterator.getDate(columnIndex).toEpochDay(); case TIMESTAMP_WITHOUT_TIME_ZONE: case TIMESTAMP_WITH_LOCAL_TIME_ZONE: - return TimestampData.fromTimestamp(iterator.getTimestamp(columnIndex)); + // fromTimestamp() would shift the value by the JVM default time zone. + return TimestampData.fromInstant(iterator.getTimestamp(columnIndex).toInstant()); default: throw new IOException( "Unsupported Flink type at column " + columnIndex + ": " + dataType.getLogicalType()); 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 fbb5ba0f..8d4c7776 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 @@ -573,7 +573,9 @@ private static String renderTimestampLiteral( TimestampData timestamp, TimestampPrecision timestampPrecision) { long units = timestampPrecision.toUnits(timestamp.getMillisecond(), timestamp.getNanoOfMillisecond()); - return "CAST(" + units + " AS TIMESTAMP)"; + // A plain number, not CAST(... AS TIMESTAMP): IoTDB turns a time-column predicate into a time + // filter only from an integer literal, and fails the query with error 301 on the CAST form. + return Long.toString(units); } private static boolean isFinite(float value) { diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java index 7ac51b06..b82f7261 100644 --- a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/sink/RowDataSinkDataConverterTest.java @@ -31,6 +31,7 @@ import java.io.IOException; import java.time.LocalDate; import java.util.Arrays; +import java.util.TimeZone; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; @@ -77,6 +78,20 @@ public void testGetters() throws IOException { } } + @Test + public void testTimestampDoesNotDependOnDefaultTimeZone() throws IOException { + TimeZone defaultZone = TimeZone.getDefault(); + try { + TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai")); + try (SinkDataConverter.Iterator iterator = converter.getIterator(rowData())) { + assertTrue(iterator.next()); + assertEquals(1000L, iterator.getTimestamp(0).getTime()); + } + } finally { + TimeZone.setDefault(defaultZone); + } + } + @Test public void testNullField() throws IOException { GenericRowData row = rowData(); diff --git a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchemaTest.java b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchemaTest.java new file mode 100644 index 00000000..171ea8c8 --- /dev/null +++ b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/test/java/org/apache/iotdb/relational/flink/source/common/RowDataDeserializationSchemaTest.java @@ -0,0 +1,119 @@ +/* + * 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.common; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.data.RowData; +import org.junit.Test; + +import java.sql.Timestamp; +import java.time.LocalDate; +import java.util.Collections; +import java.util.List; +import java.util.TimeZone; + +import static org.junit.Assert.assertEquals; + +public class RowDataDeserializationSchemaTest { + + @Test + public void testTimestampDoesNotDependOnDefaultTimeZone() throws Exception { + RowDataDeserializationSchema schema = + new RowDataDeserializationSchema( + DataTypes.ROW(DataTypes.FIELD("time", DataTypes.TIMESTAMP(3)))); + TimeZone defaultZone = TimeZone.getDefault(); + try { + TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai")); + RowData row = schema.deserialize(new SingleTimestampIterator(new Timestamp(1000L))); + assertEquals(1000L, row.getTimestamp(0, 3).getMillisecond()); + } finally { + TimeZone.setDefault(defaultZone); + } + } + + /** One row with a single TIMESTAMP column named {@code time}. */ + private static class SingleTimestampIterator implements IoTDBDataIterator { + private final Timestamp timestamp; + + SingleTimestampIterator(Timestamp timestamp) { + this.timestamp = timestamp; + } + + @Override + public boolean next() { + return true; + } + + @Override + public List getColumnNames() { + return Collections.singletonList("time"); + } + + @Override + public boolean isNull(int columnIndex) { + return false; + } + + @Override + public Timestamp getTimestamp(int columnIndex) { + return timestamp; + } + + @Override + public boolean getBoolean(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public int getInt(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public long getLong(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public float getFloat(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public double getDouble(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public String getString(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] getBinary(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public LocalDate getDate(int columnIndex) { + throw new UnsupportedOperationException(); + } + } +} 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 index b7737911..07de0bf8 100644 --- 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 @@ -61,7 +61,7 @@ public void testStringAndTimestampKeys() { assertEquals( "SELECT \"time\", \"device_id\", \"temperature\" FROM \"sensor\" " - + "WHERE \"device_id\" = 'd1' AND \"time\" = CAST(1704112200000 AS TIMESTAMP)", + + "WHERE \"device_id\" = 'd1' AND \"time\" = 1704112200000", reader.buildLookupSql(keyRow)); }