Fix insert to Clickhouse TimestampWithTimeZone

The time zone is not stored in the rows of the table,
but is stored in the column metadata.

From ClickHouse documentation
https://clickhouse.com/docs/en/sql-reference/data-types/datetime

The point in time is saved as a Unix timestamp, regardless of the time
zone or daylight saving time. The time zone affects how the values of
the DateTime type values are displayed in text format and how the values
specified as strings are parsed (‘2020-01-01 05:00:01’).

Timezone agnostic Unix timestamp is stored in tables, and the timezone
is used to transform it to text format or back during data import/export
or to make calendar calculations on the values (example: toDate, toHour
functions etc.). The time zone is not stored in the rows of the table
(or in resultset), but is stored in the column metadata.

plugin/trino-clickhouse/src/main/java/io/trino/plugin/clickhouse/ClickHouseClient.java

@@ -74,7 +74,6 @@
 import io.trino.spi.type.Decimals;
 import io.trino.spi.type.Int128;
 import io.trino.spi.type.StandardTypes;
-import io.trino.spi.type.TimeZoneKey;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.TypeManager;
 import io.trino.spi.type.TypeSignature;
@@ -104,6 +103,7 @@
 import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.TimeZone;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
@@ -167,7 +167,6 @@
 import static io.trino.spi.type.BooleanType.BOOLEAN;
 import static io.trino.spi.type.DateTimeEncoding.packDateTimeWithZone;
 import static io.trino.spi.type.DateTimeEncoding.unpackMillisUtc;
-import static io.trino.spi.type.DateTimeEncoding.unpackZoneKey;
 import static io.trino.spi.type.DateType.DATE;
 import static io.trino.spi.type.DecimalType.createDecimalType;
 import static io.trino.spi.type.DoubleType.DOUBLE;
@@ -733,7 +732,7 @@ public Optional<ColumnMapping> toColumnMapping(ConnectorSession session, Connect
                     return Optional.of(ColumnMapping.longMapping(
                             TIMESTAMP_TZ_SECONDS,
                             shortTimestampWithTimeZoneReadFunction(),
-                            shortTimestampWithTimeZoneWriteFunction()));
+                            shortTimestampWithTimeZoneWriteFunction(column.getTimeZone())));
                 }
         }
 
@@ -915,12 +914,12 @@ private static LongReadFunction shortTimestampWithTimeZoneReadFunction()
         };
     }
 
-    private static LongWriteFunction shortTimestampWithTimeZoneWriteFunction()
+    private static LongWriteFunction shortTimestampWithTimeZoneWriteFunction(TimeZone columnTimeZone)
     {
         return (statement, index, value) -> {
             long millisUtc = unpackMillisUtc(value);
-            TimeZoneKey timeZoneKey = unpackZoneKey(value);
-            statement.setObject(index, Instant.ofEpochMilli(millisUtc).atZone(timeZoneKey.getZoneId()));
+            // Clickhouse JDBC driver inserts datetime as string value as yyyy-MM-dd HH:mm:ss and zone from the Column metadata would be used.
+            statement.setObject(index, Instant.ofEpochMilli(millisUtc).atZone(columnTimeZone.toZoneId()));
         };
     }
 

plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/BaseClickHouseTypeMapping.java

@@ -1003,6 +1003,12 @@ public void testClickHouseDateTimeWithTimeZone()
             Session session = Session.builder(getSession())
                     .setTimeZoneKey(TimeZoneKey.getTimeZoneKey(sessionZone.getId()))
                     .build();
+            SqlDataTypeTest.create()
+                    .addRoundTrip("DateTime('Asia/Kathmandu')", "timestamp '2024-01-01 12:34:56'", TIMESTAMP_TZ_SECONDS, "TIMESTAMP '2024-01-01 05:19:56 +05:45'")
+                    .addRoundTrip("DateTime('Asia/Kathmandu')", "timestamp '2024-01-01 12:34:56 Asia/Kathmandu'", TIMESTAMP_TZ_SECONDS, "TIMESTAMP '2024-01-01 12:34:56 +05:45'")
+                    .addRoundTrip("DateTime('Asia/Kathmandu')", "timestamp '2024-01-01 12:34:56 +00:00'", TIMESTAMP_TZ_SECONDS, "TIMESTAMP '2024-01-01 18:19:56 +05:45'")
+                    .addRoundTrip("DateTime('Asia/Kathmandu')", "timestamp '2024-01-01 12:34:56 -01:00'", TIMESTAMP_TZ_SECONDS, "TIMESTAMP '2024-01-01 19:19:56 +05:45'")
+                    .execute(getQueryRunner(), session, clickhouseCreateAndTrinoInsert("tpch.test_timestamp_with_time_zone"));
 
             dateTimeWithTimeZoneTest(clickhouseDateTimeInputTypeFactory("datetime"))
                     .execute(getQueryRunner(), session, clickhouseCreateAndInsert("tpch.datetime_tz"));