[GitHub] [flink] wuchong commented on a change in pull request #14687: [FLINK-17868][table-planner-blink] Fix proctime in DDL can not work in batch mode

[GitBox <gi...@apache.org>]

wuchong commented on a change in pull request #14687:
URL: https://github.com/apache/flink/pull/14687#discussion_r561821288



##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java
##########
@@ -20,18 +20,40 @@
 
 import org.apache.flink.core.memory.ManagedMemoryUseCase;
 import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.streaming.api.TimerService;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
 
-/** Table operator to invoke close always. */
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * Table operator to invoke close always. This is a base class for both batch and stream operators
+ * without key.
+ */
 public class TableStreamOperator<OUT> extends AbstractStreamOperator<OUT> {

Review comment:
       We can add an `abstract` modifier to the class. 

##########
File path: flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/TableScanITCase.scala
##########
@@ -54,14 +56,16 @@ class TableScanITCase extends BatchTestBase {
 
     val tableSource = new TestTableSourceWithTime(true, schema, returnType, data, null, "ptime")
     tEnv.asInstanceOf[TableEnvironmentInternal].registerTableSourceInternal(tableName, tableSource)
-
+    val expected = LocalDateTime
+      .now(ZoneId.of("UTC"))
+      .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm"))
     checkResult(
-      s"SELECT name FROM $tableName",
+      s"SELECT name, DATE_FORMAT(ptime, 'yyyy-MM-dd HH:mm') FROM $tableName",
       Seq(
-        row("Mary"),
-        row("Peter"),
-        row("Bob"),
-        row("Liz"))
+        row("Mary", expected),
+        row("Peter", expected),
+        row("Bob", expected),
+        row("Liz", expected))

Review comment:
       This is not a stable test. The results of proctime may span multiple minutes. 
   
   We can call `CHAR_LENGTH` on the DATE_FORMAT to have a deterministic result. 

##########
File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkBatchProgram.scala
##########
@@ -73,6 +74,8 @@ object FlinkBatchProgram {
           .build(), "convert table references after sub-queries removed")
         .build()
     )
+    // convert time indicators
+    chainedProgram.addLast(TIME_INDICATOR, new FlinkRelTimeIndicatorProgram)

Review comment:
       We put TIME_INDICATOR phase under decorrelate in streaming programs, I think we should keep the same order in batch program to put the phase after decorrelate. 




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org