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)); }