[FLINK-36862][table] Implement additional TO_TIMESTAMP_LTZ() functions #25763

yiyutian1 · 2024-12-09T07:49:44Z

What is the purpose of the change

The Flink TO_TIMESTAMP_LTZ is lacking some behaviors, compared with other vendors. eg. https://docs.snowflake.com/en/sql-reference/functions/to_timestamp

This PR adds the following additional functions for TO_TIMESTAMP_LTZ.

TO_TIMESTAMP_LTZ(string1) -- Parse timestamp with default format 
TO_TIMESTAMP_LTZ(string1, string2) -- Parse timestamp with custom format 
TO_TIMESTAMP_LTZ(string1, string2, string3) -- Parse timestamp with format and timezone 
TO_TIMESTAMP_LTZ(numeric) -- For epoch milliseconds

Brief change log

Add the additional overloading functions for TO_TIMESTAMP_LTZ.
Add java test cases.
Make sure the current scala tests pass.

Verifying this change

This change is already covered by existing tests, TemporalTypesTest.scala.
Added additional tests in TimeFunctionsITCase.java.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: ( no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (documented in sql_functions.yml)

flinkbot · 2024-12-09T07:52:04Z

CI report:

5a6f93c Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

...able-planner/src/test/java/org/apache/flink/table/planner/functions/TimeFunctionsITCase.java

...-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

...le-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala

docs/data/sql_functions.yml

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

snuyanzin · 2024-12-12T11:06:51Z

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

+        return DateTimeUtils.toTimestampData(epoch, precision);
+    }
+
+    public TimestampData eval(Double epoch, Integer precision) {


This I didn't get, why do we have anything with floating point here?
Can you please clarify
first what does fractional epoch actually means? Does any vendor support it?
Second floating points have issues with exact data

Hi Sergey, they are for existing tests like:

flink/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala

Line 1159 in 2d9fb79

toTimestampLtz(100.01.cast(DataTypes.FLOAT()), 0),

snuyanzin · 2024-12-12T11:08:08Z

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

+    public TimestampData eval(Float value, Integer precision) {
+        if (value == null || precision == null) {
+            return null;
+        }
+        return DateTimeUtils.toTimestampData(value.longValue(), precision);
+    }
+
+    public TimestampData eval(Byte value, Integer precision) {
+        if (value == null || precision == null) {
+            return null;
+        }
+        return DateTimeUtils.toTimestampData(value.longValue(), precision);
+    }
+
+    public TimestampData eval(DecimalData epoch, Integer precision) {
+        if (epoch == null || precision == null) {
+            return null;
+        }
+
+        return DateTimeUtils.toTimestampData(epoch, precision);
+    }


We need only numeric (exact numeric)
or did I miss anything?

same as above

.../main/java/org/apache/flink/table/types/inference/strategies/ToTimestampLtzTypeStrategy.java

snuyanzin · 2024-12-12T11:24:59Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java

+    public static TimestampData toTimestampData(long epoch) {
+        return toTimestampData(epoch, 3);
+    }
+
+    public static TimestampData toTimestampData(double epoch) {
+        return toTimestampData(epoch, 3);
+    }
+
+    public static TimestampData toTimestampData(DecimalData epoch) {
+        return toTimestampData(epoch, 3);
+    }
+


from any user outside of TO_TIMESTAMP_LTZ 3 here is a magic number.
I would suggest to have such calls only in your toTimestampLtzFunction class and don't put them here

addressed here: 5335325#r1886093369

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

snuyanzin · 2024-12-12T11:42:05Z

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

+    public TimestampData eval(Long epoch) {
+        if (epoch == null) {
+            return null;
+        }
+
+        return eval(epoch, 3);
+    }
+
+    public TimestampData eval(Float epoch) {
+        if (epoch == null) {
+            return null;
+        }
+
+        return eval(epoch, 3);
+    }


Can not we replace a number of these duplicating methods with something like

public TimestampData eval(Number epoch) { if (epoch == null) { return null; } return eval(epoch, 3); }

?

Address here: 5335325#r1886090826

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

yiyutian1 · 2024-12-13T06:18:48Z

@flinkbot run azure

davidradl · 2024-12-13T15:06:11Z

Reviewed by Chi on 12/12/24 Group to test and/or review outside of the meeting (Nic Townsend from IBM has some insight he will add on this)

snuyanzin · 2024-12-13T16:28:21Z

...le-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala

@@ -1158,7 +1157,7 @@ class TemporalTypesTest extends ExpressionTestBase {
    testAllApis(
      toTimestampLtz(100.01.cast(DataTypes.FLOAT()), 0),
      "TO_TIMESTAMP_LTZ(CAST(100.01 AS FLOAT), 0)",
-      "1970-01-01 08:01:40.010")
+      "1970-01-01 08:01:40.000")


why is it like that?
why are we missing millis now?

Or are we casting to timestamp with precision zero?
If so, do we have a similar test with cast to timestamp with precision 3?

Hi @snuyanzin, the 010 is the original test case. I reverted it back to 010 and the test passes now.
I added a precision with 3 in the new Java test.

yiyutian1 · 2024-12-16T02:55:57Z

...-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java

@@ -2332,7 +2332,8 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)

    public static final BuiltInFunctionDefinition TO_TIMESTAMP_LTZ =
            BuiltInFunctionDefinition.newBuilder()
-                    .name("TO_TIMESTAMP_LTZ")
+                    .name("toTimestampLtz")
+                    .sqlName("TO_TIMESTAMP_LTZ")


Reverted the change of function name. Added necessary logic register the function.

yiyutian1 · 2024-12-16T02:58:09Z

...able-planner/src/test/java/org/apache/flink/table/planner/functions/TimeFunctionsITCase.java

-                                TIMESTAMP_LTZ(0).nullable())
+                                null,
+                                TIMESTAMP_LTZ(3).nullable())
+                        .testResult(


To address this comment, added a FLOAT case with precision 3.

yiyutian1 · 2024-12-16T02:58:46Z

...le-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala


    // FLOAT -> TIMESTAMP_LTZ
    testAllApis(
      toTimestampLtz(100.01.cast(DataTypes.FLOAT()), 0),
      "TO_TIMESTAMP_LTZ(CAST(100.01 AS FLOAT), 0)",
-      "1970-01-01 08:01:40")
+      "1970-01-01 08:01:40.010")


Revert to the original test case.

yiyutian1 · 2024-12-16T02:59:31Z

...le-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala

-      "Cannot apply 'TO_TIMESTAMP_LTZ' to arguments of type" +
-        " 'TO_TIMESTAMP_LTZ(<CHAR(16)>, <INTEGER>)'. Supported form(s):" +
-        " 'TO_TIMESTAMP_LTZ(<NUMERIC>, <INTEGER>)'",
+      "SQL validation failed. From line 1, column 8 to line 1, column 46: Cannot apply " +


Need to change these existing scala tests, because I changed the function signature. The change reflects the expected function behaviors.

yiyutian1 · 2024-12-16T03:06:49Z

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java

@@ -92,74 +121,72 @@ public TimestampData eval(Integer epoch) {
            return null;
        }

-        return eval(epoch, 3);


Replying to this comment:
This is not possible, because we handle exact numeric and fractional data differently. Therefore, we can’t use the umbrella Number for them.
I keep them as is to be consistent with existing logic.
@snuyanzin , if you have a better way, please shed some lights here.

Can not we replace a number of these duplicating methods with something like

public TimestampData eval(Number epoch) { if (epoch == null) { return null; } return eval(epoch, 3); }

yiyutian1 · 2024-12-16T03:09:44Z

flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/DateTimeUtils.java

@@ -385,6 +388,18 @@ public static TimestampData toTimestampData(int v, int precision) {
        }
    }



To address this comment, I replaced 3 with a constant.
After giving it some thought, I believe it's better to keep these functions in DateTimeUtils.java rather than moving them to toTimestampLtzFunction, since toTimestampLtzFunction already depends on DateTimeUtils.java as a library. What are your thoughts on this? @snuyanzin

from any user outside of TO_TIMESTAMP_LTZ 3 here is a magic number.
I would suggest to have such calls only in your toTimestampLtzFunction class and don't put them here

yiyutian1 · 2024-12-16T03:17:40Z

Reviewed by Chi on 12/12/24 Group to test and/or review outside of the meeting (Nic Townsend from IBM has some insight he will add on this)

Hi @davidradl, I appreciate your involvement with this PR. As a new member to this community, this means a lot to me!
However, this function is currently a blocker for a user at my company, Confluent.

I've also just submitted another commit to address the previous comments, added tests, and made other updates, which may have rendered some of your current feedback outdated.

Given the time constraints, can I proceed with the Flink committer's approval, and address your comments in a separate PR?

flinkbot added the component=<none> label Dec 9, 2024

davidradl reviewed Dec 9, 2024

View reviewed changes

...able-planner/src/test/java/org/apache/flink/table/planner/functions/TimeFunctionsITCase.java Outdated Show resolved Hide resolved

davidradl reviewed Dec 9, 2024

View reviewed changes

...-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java Outdated Show resolved Hide resolved

davidradl reviewed Dec 9, 2024

View reviewed changes

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java Outdated Show resolved Hide resolved

davidradl reviewed Dec 9, 2024

View reviewed changes

...me/src/main/java/org/apache/flink/table/runtime/functions/scalar/ToTimestampLtzFunction.java Outdated Show resolved Hide resolved

yiyutian1 force-pushed the flink36862 branch 2 times, most recently from bd3ff92 to 77a36af Compare December 11, 2024 05:50

yiyutian1 changed the title ~~[FLINK-36862][table] Implement additional TO_TIMESTAMP_LTZ() functions~~ [WIP][FLINK-36862][table] Implement additional TO_TIMESTAMP_LTZ() functions Dec 11, 2024

yiyutian1 force-pushed the flink36862 branch 3 times, most recently from 1936203 to 0bf526d Compare December 12, 2024 09:45

[FLINK-36862][table] Implement additional TO_TIMESTAMP_LTZ() functions

750f965

yiyutian1 force-pushed the flink36862 branch from 0bf526d to 750f965 Compare December 12, 2024 09:56

yiyutian1 marked this pull request as ready for review December 12, 2024 10:05

yiyutian1 changed the title ~~[WIP][FLINK-36862][table] Implement additional TO_TIMESTAMP_LTZ() functions~~ [FLINK-36862][table] Implement additional TO_TIMESTAMP_LTZ() functions Dec 12, 2024

yiyutian1 commented Dec 12, 2024

View reviewed changes

...le-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala Outdated Show resolved Hide resolved