[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Timestamp conversion problem in Flink Table/SQL

Hi community,

Recently I have encountered a problem with time conversion in Flink Table/SQL . When the processed field contains a timestamp type, the code of the flink table codegen first converts the timestamp type to a long type, and then converts the long type to a timestamp type on output. 
In the code generated by codegen, 
 “public static long toLong (Timestamp v)” and 
“public static java.sql.Timestamp internalToTimestamp (long v)” 
 are used in the conversion.
The internal implementation of these two methods will add or subtract the time zone offset.
In some cases, the two methods do not appear in pairs which causes the conversion time to be incorrect, resulting in watermark timestamp metrics on the web ui is equal to the correct value plus time zone offset, and the output of the process time field is equal to the correct value minus the time zone offset.

Why the time conversion method in calcite (  add or subtract time zones?Why flink Table/SQL uses these time conversion methods instead of using timestamp.getTime() .

/** Converts the Java type used for UDF parameters of SQL TIMESTAMP type
 * ({@link java.sql.Timestamp}) to internal representation (long).
 * <p>Converse of {@link #internalToTimestamp(long)}. */
public static long toLong(Timestamp v) {
  return toLong(v, LOCAL_TZ);

// mainly intended for java.sql.Timestamp but works for other dates also
public static long toLong(java.util.Date v, TimeZone timeZone) {
  final long time = v.getTime();
  return time + timeZone.getOffset(time);

/** Converts the internal representation of a SQL TIMESTAMP (long) to the Java
 * type used for UDF parameters ({@link java.sql.Timestamp}). */
public static java.sql.Timestamp internalToTimestamp(long v) {
  return new java.sql.Timestamp(v - LOCAL_TZ.getOffset(v));

Related issue: