Kudu: add kudu.timezone for TIMESTAMP conversion

docs/src/main/sphinx/connector/kudu.md

@@ -56,6 +56,9 @@ kudu.client.master-addresses=localhost
 
 ## Assign Kudu splits to replica host if worker and kudu share the same cluster
 #kudu.allow-local-scheduling = false
+
+## Optional timezone used for TIMESTAMP values (default: UTC)
+#kudu.timezone = UTC
 ```
 
 ## Kerberos support

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduClientConfig.java

@@ -44,6 +44,7 @@ public class KuduClientConfig
     private String schemaEmulationPrefix = "presto::";
     private Duration dynamicFilteringWaitTimeout = new Duration(0, MINUTES);
     private boolean allowLocalScheduling;
+    private String timeZone = "UTC";
 
     @NotNull
     @Size(min = 1)
@@ -138,6 +139,19 @@ public KuduClientConfig setDynamicFilteringWaitTimeout(Duration dynamicFiltering
         return this;
     }
 
+    @Config("kudu.timezone")
+    public KuduClientConfig setTimeZone(String timeZone)
+    {
+        this.timeZone = timeZone;
+        TimestampHelper.setTimeZoneOffset(TimestampHelper.getZoneOffset(timeZone));
+        return this;
+    }
+
+    public String getTimeZone()
+    {
+        return timeZone;
+    }
+
     public boolean isAllowLocalScheduling()
     {
         return allowLocalScheduling;

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduPageSink.java

@@ -157,7 +157,7 @@ else if (DATE.equals(type)) {
             row.addDate(destChannel, epochDaysToSqlDate(INTEGER.getInt(block, position)));
         }
         else if (TIMESTAMP_MILLIS.equals(type)) {
-            row.addLong(destChannel, truncateEpochMicrosToMillis(TIMESTAMP_MILLIS.getLong(block, position)));
+            row.addLong(destChannel, truncateEpochMicrosToMillis(TIMESTAMP_MILLIS.getLong(block, position)) - (TimestampHelper.getTimeZoneOffset() * 1_000));
         }
         else if (REAL.equals(type)) {
             row.addFloat(destChannel, REAL.getFloat(block, position));

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/TimestampHelper.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed 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 io.trino.plugin.kudu;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+
+public final class TimestampHelper
+{
+    private TimestampHelper() {}
+
+    private static long timeZoneOffsetMillis;
+
+    public static long getTimeZoneOffset()
+    {
+        return timeZoneOffsetMillis;
+    }
+
+    public static void setTimeZoneOffset(long timeZoneOffsetMillis)
+    {
+        TimestampHelper.timeZoneOffsetMillis = timeZoneOffsetMillis;
+    }
+
+    public static long getZoneOffset(String zone)
+    {
+        ZoneId zoneId = ZoneId.of(zone);
+        ZoneOffset offset = zoneId.getRules().getOffset(Instant.now());
+        return offset.getTotalSeconds() * 1000L;
+    }
+}

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/TypeHelper.java

@@ -173,7 +173,7 @@ public static Object getJavaValue(Type type, Object nativeValue)
         }
         if (type.equals(TIMESTAMP_MILLIS)) {
             // Kudu's native format is in microseconds
-            return nativeValue;
+            return ((Long) nativeValue) - (TimestampHelper.getTimeZoneOffset() * 1_000);
         }
         throw new IllegalStateException("Back conversion not implemented for " + type);
     }
@@ -215,7 +215,7 @@ public static long getLong(Type type, RowResult row, int field)
             return row.getInt(field);
         }
         if (type.equals(TIMESTAMP_MILLIS)) {
-            return truncateEpochMicrosToMillis(row.getLong(field));
+            return truncateEpochMicrosToMillis(row.getLong(field)) + (TimestampHelper.getTimeZoneOffset() * 1_000);
         }
         throw new IllegalStateException("getLong not implemented for " + type);
     }