[core] Introduce data structure for timestamp with local zone (#3857)

paimon-common/src/main/java/org/apache/paimon/data/LocalZoneTimestamp.java

@@ -0,0 +1,199 @@
+/*
+ * 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.paimon.data;
+
+import org.apache.paimon.annotation.Public;
+import org.apache.paimon.types.LocalZonedTimestampType;
+
+import java.io.Serializable;
+import java.time.Instant;
+
+import static org.apache.paimon.data.Timestamp.MICROS_PER_MILLIS;
+import static org.apache.paimon.data.Timestamp.NANOS_PER_MICROS;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * An internal data structure representing data of {@link LocalZonedTimestampType}.
+ *
+ * <p>This data structure is immutable and consists of a milliseconds and nanos-of-millisecond since
+ * {@code 1970-01-01 00:00:00}. It might be stored in a compact representation (as a long value) if
+ * values are small enough.
+ *
+ * @since 0.9.0
+ */
+@Public
+public final class LocalZoneTimestamp implements Comparable<LocalZoneTimestamp>, Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    // this field holds the integral second and the milli-of-second
+    private final long millisecond;
+
+    // this field holds the nano-of-millisecond
+    private final int nanoOfMillisecond;
+
+    private LocalZoneTimestamp(long millisecond, int nanoOfMillisecond) {
+        checkArgument(nanoOfMillisecond >= 0 && nanoOfMillisecond <= 999_999);
+        this.millisecond = millisecond;
+        this.nanoOfMillisecond = nanoOfMillisecond;
+    }
+
+    /** Returns the number of milliseconds since {@code 1970-01-01 00:00:00}. */
+    public long getMillisecond() {
+        return millisecond;
+    }
+
+    /**
+     * Returns the number of nanoseconds (the nanoseconds within the milliseconds).
+     *
+     * <p>The value range is from 0 to 999,999.
+     */
+    public int getNanoOfMillisecond() {
+        return nanoOfMillisecond;
+    }
+
+    /** Converts this {@link LocalZoneTimestamp} object to a {@link java.sql.Timestamp}. */
+    public java.sql.Timestamp toSQLTimestamp() {
+        return java.sql.Timestamp.from(toInstant());
+    }
+
+    public LocalZoneTimestamp toMillisTimestamp() {
+        return fromEpochMillis(millisecond);
+    }
+
+    /** Converts this {@link LocalZoneTimestamp} object to a {@link Instant}. */
+    public Instant toInstant() {
+        long epochSecond = millisecond / 1000;
+        int milliOfSecond = (int) (millisecond % 1000);
+        if (milliOfSecond < 0) {
+            --epochSecond;
+            milliOfSecond += 1000;
+        }
+        long nanoAdjustment = milliOfSecond * 1_000_000 + nanoOfMillisecond;
+        return Instant.ofEpochSecond(epochSecond, nanoAdjustment);
+    }
+
+    /** Converts this {@link LocalZoneTimestamp} object to micros. */
+    public long toMicros() {
+        long micros = Math.multiplyExact(millisecond, MICROS_PER_MILLIS);
+        return micros + nanoOfMillisecond / NANOS_PER_MICROS;
+    }
+
+    @Override
+    public int compareTo(LocalZoneTimestamp that) {
+        int cmp = Long.compare(this.millisecond, that.millisecond);
+        if (cmp == 0) {
+            cmp = this.nanoOfMillisecond - that.nanoOfMillisecond;
+        }
+        return cmp;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof LocalZoneTimestamp)) {
+            return false;
+        }
+        LocalZoneTimestamp that = (LocalZoneTimestamp) obj;
+        return this.millisecond == that.millisecond
+                && this.nanoOfMillisecond == that.nanoOfMillisecond;
+    }
+
+    @Override
+    public String toString() {
+        return toSQLTimestamp().toLocalDateTime().toString();
+    }
+
+    @Override
+    public int hashCode() {
+        int ret = (int) millisecond ^ (int) (millisecond >> 32);
+        return 31 * ret + nanoOfMillisecond;
+    }
+
+    // ------------------------------------------------------------------------------------------
+    // Constructor Utilities
+    // ------------------------------------------------------------------------------------------
+
+    /** Creates an instance of {@link LocalZoneTimestamp} for now. */
+    public static LocalZoneTimestamp now() {
+        return fromInstant(Instant.now());
+    }
+
+    /**
+     * Creates an instance of {@link LocalZoneTimestamp} from milliseconds.
+     *
+     * <p>The nanos-of-millisecond field will be set to zero.
+     *
+     * @param milliseconds the number of milliseconds since {@code 1970-01-01 00:00:00}; a negative
+     *     number is the number of milliseconds before {@code 1970-01-01 00:00:00}
+     */
+    public static LocalZoneTimestamp fromEpochMillis(long milliseconds) {
+        return new LocalZoneTimestamp(milliseconds, 0);
+    }
+
+    /**
+     * Creates an instance of {@link LocalZoneTimestamp} from milliseconds and a
+     * nanos-of-millisecond.
+     *
+     * @param milliseconds the number of milliseconds since {@code 1970-01-01 00:00:00}; a negative
+     *     number is the number of milliseconds before {@code 1970-01-01 00:00:00}
+     * @param nanosOfMillisecond the nanoseconds within the millisecond, from 0 to 999,999
+     */
+    public static LocalZoneTimestamp fromEpochMillis(long milliseconds, int nanosOfMillisecond) {
+        return new LocalZoneTimestamp(milliseconds, nanosOfMillisecond);
+    }
+
+    /**
+     * Creates an instance of {@link LocalZoneTimestamp} from an instance of {@link
+     * java.sql.Timestamp}.
+     *
+     * @param timestamp an instance of {@link java.sql.Timestamp}
+     */
+    public static LocalZoneTimestamp fromSQLTimestamp(java.sql.Timestamp timestamp) {
+        return fromInstant(timestamp.toInstant());
+    }
+
+    /**
+     * Creates an instance of {@link LocalZoneTimestamp} from an instance of {@link Instant}.
+     *
+     * @param instant an instance of {@link Instant}
+     */
+    public static LocalZoneTimestamp fromInstant(Instant instant) {
+        long epochSecond = instant.getEpochSecond();
+        int nanoSecond = instant.getNano();
+
+        long millisecond = epochSecond * 1_000 + nanoSecond / 1_000_000;
+        int nanoOfMillisecond = nanoSecond % 1_000_000;
+
+        return new LocalZoneTimestamp(millisecond, nanoOfMillisecond);
+    }
+
+    /** Creates an instance of {@link LocalZoneTimestamp} from micros. */
+    public static LocalZoneTimestamp fromMicros(long micros) {
+        long mills = Math.floorDiv(micros, MICROS_PER_MILLIS);
+        long nanos = (micros - mills * MICROS_PER_MILLIS) * NANOS_PER_MICROS;
+        return LocalZoneTimestamp.fromEpochMillis(mills, (int) nanos);
+    }
+
+    /**
+     * Returns whether the timestamp data is small enough to be stored in a long of milliseconds.
+     */
+    public static boolean isCompact(int precision) {
+        return precision <= 3;
+    }
+}

paimon-common/src/main/java/org/apache/paimon/data/Timestamp.java

@@ -30,13 +30,15 @@
 import java.time.LocalTime;
 
 /**
- * An internal data structure representing data of {@link TimestampType} and {@link
- * LocalZonedTimestampType}.
+ * An internal data structure representing data of {@link TimestampType}.
  *
  * <p>This data structure is immutable and consists of a milliseconds and nanos-of-millisecond since
  * {@code 1970-01-01 00:00:00}. It might be stored in a compact representation (as a long value) if
  * values are small enough.
  *
+ * <p>Legacy: This class represents {@link LocalZonedTimestampType} too, now it is recommended to
+ * use {@link LocalZoneTimestamp}.
+ *
  * @since 0.4.0
  */
 @Public
@@ -45,7 +47,7 @@ public final class Timestamp implements Comparable<Timestamp>, Serializable {
     private static final long serialVersionUID = 1L;
 
     // the number of milliseconds in a day
-    private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
+    public static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
 
     public static final long MICROS_PER_MILLIS = 1000L;
 
@@ -100,7 +102,12 @@ public LocalDateTime toLocalDateTime() {
         return LocalDateTime.of(localDate, localTime);
     }
 
-    /** Converts this {@link Timestamp} object to a {@link Instant}. */
+    /**
+     * Converts this {@link Timestamp} object to a {@link Instant}.
+     *
+     * @deprecated use {@link LocalZoneTimestamp}.
+     */
+    @Deprecated
     public Instant toInstant() {
         long epochSecond = millisecond / 1000;
         int milliOfSecond = (int) (millisecond % 1000);
@@ -208,7 +215,9 @@ public static Timestamp fromSQLTimestamp(java.sql.Timestamp timestamp) {
      * Creates an instance of {@link Timestamp} from an instance of {@link Instant}.
      *
      * @param instant an instance of {@link Instant}
+     * @deprecated use {@link LocalZoneTimestamp}.
      */
+    @Deprecated
     public static Timestamp fromInstant(Instant instant) {
         long epochSecond = instant.getEpochSecond();
         int nanoSecond = instant.getNano();

paimon-common/src/test/java/org/apache/paimon/data/LocalZoneTimestampTest.java

@@ -0,0 +1,121 @@
+/*
+ * 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.paimon.data;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.TimeZone;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link LocalZoneTimestamp}. */
+public class LocalZoneTimestampTest {
+
+    @Test
+    public void testNormal() {
+        // From long to TimestampData and vice versa
+        assertThat(LocalZoneTimestamp.fromEpochMillis(1123L).getMillisecond()).isEqualTo(1123L);
+        assertThat(LocalZoneTimestamp.fromEpochMillis(-1123L).getMillisecond()).isEqualTo(-1123L);
+
+        assertThat(LocalZoneTimestamp.fromEpochMillis(1123L, 45678).getMillisecond())
+                .isEqualTo(1123L);
+        assertThat(LocalZoneTimestamp.fromEpochMillis(1123L, 45678).getNanoOfMillisecond())
+                .isEqualTo(45678);
+
+        assertThat(LocalZoneTimestamp.fromEpochMillis(-1123L, 45678).getMillisecond())
+                .isEqualTo(-1123L);
+        assertThat(LocalZoneTimestamp.fromEpochMillis(-1123L, 45678).getNanoOfMillisecond())
+                .isEqualTo(45678);
+
+        // From TimestampData to TimestampData and vice versa
+        java.sql.Timestamp t19 = java.sql.Timestamp.valueOf("1969-01-02 00:00:00.123456789");
+        java.sql.Timestamp t16 = java.sql.Timestamp.valueOf("1969-01-02 00:00:00.123456");
+        java.sql.Timestamp t13 = java.sql.Timestamp.valueOf("1969-01-02 00:00:00.123");
+        java.sql.Timestamp t10 = java.sql.Timestamp.valueOf("1969-01-02 00:00:00");
+
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t19).toSQLTimestamp()).isEqualTo(t19);
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t16).toSQLTimestamp()).isEqualTo(t16);
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t13).toSQLTimestamp()).isEqualTo(t13);
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t10).toSQLTimestamp()).isEqualTo(t10);
+
+        java.sql.Timestamp t2 = java.sql.Timestamp.valueOf("1979-01-02 00:00:00.123456");
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t2).toSQLTimestamp()).isEqualTo(t2);
+
+        java.sql.Timestamp t3 = new java.sql.Timestamp(1572333940000L);
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t3).toSQLTimestamp()).isEqualTo(t3);
+
+        // From Instant to TimestampData and vice versa
+        Instant instant1 = Instant.ofEpochMilli(123L);
+        Instant instant2 = Instant.ofEpochSecond(0L, 123456789L);
+        Instant instant3 = Instant.ofEpochSecond(-2L, 123456789L);
+
+        assertThat(LocalZoneTimestamp.fromInstant(instant1).toInstant()).isEqualTo(instant1);
+        assertThat(LocalZoneTimestamp.fromInstant(instant2).toInstant()).isEqualTo(instant2);
+        assertThat(LocalZoneTimestamp.fromInstant(instant3).toInstant()).isEqualTo(instant3);
+    }
+
+    @Test
+    public void testDaylightSavingTime() {
+        TimeZone tz = TimeZone.getDefault();
+        TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"));
+
+        java.sql.Timestamp dstBegin2018 = java.sql.Timestamp.valueOf("2018-03-11 03:00:00");
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(dstBegin2018).toSQLTimestamp())
+                .isEqualTo(dstBegin2018);
+
+        java.sql.Timestamp dstBegin2019 = java.sql.Timestamp.valueOf("2019-03-10 02:00:00");
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(dstBegin2019).toSQLTimestamp())
+                .isEqualTo(dstBegin2019);
+
+        TimeZone.setDefault(tz);
+    }
+
+    @Test
+    public void testToString() {
+
+        java.sql.Timestamp t = java.sql.Timestamp.valueOf("1969-01-02 00:00:00.123456789");
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t).toString())
+                .isEqualTo("1969-01-02T00:00:00.123456789");
+
+        assertThat(LocalZoneTimestamp.fromEpochMillis(123L).toString())
+                .isEqualTo(
+                        Instant.ofEpochMilli(123)
+                                .atZone(ZoneId.systemDefault())
+                                .toLocalDateTime()
+                                .toString());
+
+        Instant instant = Instant.ofEpochSecond(0L, 123456789L);
+        assertThat(LocalZoneTimestamp.fromInstant(instant).toString())
+                .isEqualTo(instant.atZone(ZoneId.systemDefault()).toLocalDateTime().toString());
+    }
+
+    @Test
+    public void testToMicros() {
+        java.sql.Timestamp t = java.sql.Timestamp.valueOf("2005-01-02 00:00:00.123456789");
+        assertThat(LocalZoneTimestamp.fromSQLTimestamp(t).toString())
+                .isEqualTo("2005-01-02T00:00:00.123456789");
+        assertThat(
+                        LocalZoneTimestamp.fromMicros(
+                                        LocalZoneTimestamp.fromSQLTimestamp(t).toMicros())
+                                .toString())
+                .isEqualTo("2005-01-02T00:00:00.123456");
+    }
+}