The query contains more than one rowtime attribute column error

[mbroecheler]

Queries with multiple timestamp columns in temporal joins may cause a runtime error in flink. During SQRL analysis, we derive the correct timestamp column so we need to cast all other columns to TIMESTAMP_LTZ. The timestamp may not be fully known during script analysis, so we should determine the cast after we do the graphql query generation. Current the graphql query generation is a separate process so it makes it difficult to track that column though the process, but this could be unified when we implement table functions.

Stacktrace

deploy-flink-jobmanager-1     | org.apache.flink.table.api.TableException: The query contains more than one rowtime attribute column [first_order, time$0] for writing into table 'default_catalog.default_database.order_stats$26'.
deploy-flink-jobmanager-1     | Please select the column that should be used as the event-time timestamp for the table sink by casting all other columns to regular TIMESTAMP or TIMESTAMP_LTZ.
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.java:159)
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:158)
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.planner.delegation.StreamPlanner.$anonfun$translateToPlan$1(StreamPlanner.scala:85)
deploy-flink-jobmanager-1     | 	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
deploy-flink-jobmanager-1     | 	at scala.collection.Iterator.foreach(Iterator.scala:937)
deploy-flink-jobmanager-1     | 	at scala.collection.Iterator.foreach$(Iterator.scala:937)
deploy-flink-jobmanager-1     | 	at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
deploy-flink-jobmanager-1     | 	at scala.collection.IterableLike.foreach(IterableLike.scala:70)
deploy-flink-jobmanager-1     | 	at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
deploy-flink-jobmanager-1     | 	at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
deploy-flink-jobmanager-1     | 	at scala.collection.TraversableLike.map(TraversableLike.scala:233)
deploy-flink-jobmanager-1     | 	at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
deploy-flink-jobmanager-1     | 	at scala.collection.AbstractTraversable.map(Traversable.scala:104)
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:84)
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:197)
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1733)
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:825)
deploy-flink-jobmanager-1     | 	at org.apache.flink.table.api.internal.StatementSetImpl.execute(StatementSetImpl.java:108)
deploy-flink-jobmanager-1     | 	at com.datasqrl.FlinkMain.run(FlinkMain.java:44)
deploy-flink-jobmanager-1     | 	at com.datasqrl.FlinkMain.main(FlinkMain.java:26)
deploy-flink-jobmanager-1     | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
deploy-flink-jobmanager-1     | 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
deploy-flink-jobmanager-1     | 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
deploy-flink-jobmanager-1     | 	at java.base/java.lang.reflect.Method.invoke(Unknown Source)
deploy-flink-jobmanager-1     | 	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
deploy-flink-jobmanager-1     | 	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
deploy-flink-jobmanager-1     | 	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98)
deploy-flink-jobmanager-1     | 	at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84)
deploy-flink-jobmanager-1     | 	at org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:70)
deploy-flink-jobmanager-1     | 	at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$0(JarRunHandler.java:107)
deploy-flink-jobmanager-1     | 	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
deploy-flink-jobmanager-1     | 	at java.base/java.lang.Thread.run(Unknown Source)