Fix timestamp filters and time zone handling in the Flink table connector

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

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

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) {

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

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<String> 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();
+    }
+  }
+}

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