title: Flink : Rowtime attributes must not be in the input rows of a regular join
body: <p>Using flink SQL API, I want to join multiple tables together and do some computation over time window.
  I have 3 table coming from CSV files, and one coming from Kafka.
  In the Kafka table, I have a field <code>timestampMs</code>, that I want to use for my time window operations.</p>
  
  <p>For that I did the following code : </p>
  
  <pre><code>reamExecutionEnvironment env = ... ;
  StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  
  TableSource table1 = CsvTableSource.builder()
          .path("path/to/file1.csv")
          .ignoreFirstLine()
          .fieldDelimiter(",")
          .field("id1", Types.STRING)
          .field("someInfo1", Types.FLOAT)
          .build();
  
  TableSource table2 = CsvTableSource.builder()
          .path("path/to/file2.csv")
          .ignoreFirstLine()
          .fieldDelimiter(",")
          .field("id2", Types.STRING)
          .field("someInfo2", Types.STRING)
          .build();
  
  TableSource table3 = CsvTableSource.builder()
          .path("path/to/file3.csv")
          .ignoreFirstLine()
          .fieldDelimiter(",")
          .field("id2", Types.STRING)
          .field("id1", Types.STRING)
          .field("someInfo3", Types.FLOAT)
          .build();
  
  tableEnv.registerTableSource("Table1",table1);
  tableEnv.registerTableSource("Table2",table2);
  tableEnv.registerTableSource("Table3",table3);
  
  
  Schema schemaExt = new Schema().schema(SOME_SCHEMA);
  schemaExt = schemaExt.field("rowtime", Types.SQL_TIMESTAMP).rowtime(new Rowtime().timestampsFromField("timestampMs").watermarksPeriodicBounded(40000));
  
  tableEnv.connect(new Kafka()
          .version("universal")
          .topic(MY_TOPIC)
          .properties(MY_PROPERTIES)
          .sinkPartitionerRoundRobin()
  )
              .withFormat(...)
              .withSchema(schemaExt)
              .inAppendMode()
              .registerTableSource("KafkaInput");
  
  Table joined = tableEnv.sqlQuery("SELECT * FROM table1 " +
          "join table3 on table1.id2 = table3.id2 " +
          "join table2 on table3.id1 = table2.id1 " +
          "join KafkaInput on table3.id2 = KafkaInput.id2");
  
  tableEnv.registerTable("Joined", joined);
  
  int windowWidth = 5;
  int frequency = 2;
  Table processed = tableEnv.sqlQuery("SELECT id1 FROM Joined " +
          "GROUP BY id1, HOP(rowtime, INTERVAL '10' SECOND, INTERVAL '30' SECOND)");
  
  
  
  Sink s = createSink(this.esEndpoint, this.esPattern, this.schemaHandler.getSchemaStr());
  
  
  tableEnv.registerTableSink("MySink", ...);
  
  processed.insertInto("MySink");
  
  env.execute();
  </code></pre>
  
  <p>But when I run it, I have the following error : </p>
  
  <pre><code>Exception in thread "main" org.apache.flink.table.api.TableException: Cannot generate a valid execution plan for the given query: 
  Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.
  </code></pre>
  
  <p>But I don't understand the workaround tip part. How can I create a time attribute and do some windowed computation after joining my tables.</p>
  
  <p>--- EDIT ---</p>
  
  <p>In the above code, I replaced the following lines : </p>
  
  <pre><code>Table joined = tableEnv.sqlQuery("SELECT * FROM table1 " +
          "join table3 on table1.id2 = table3.id2 " +
          "join table2 on table3.id1 = table2.id1 " +
          "join KafkaInput on table3.id2 = KafkaInput.id2");
  
  tableEnv.registerTable("Joined", joined);
  </code></pre>
  
  <p>By : </p>
  
  <pre><code>Table staticJoined = tableEnv.sqlQuery("SELECT *, TIMESTAMP('1970-01-01 00:00:00') as rowtime FROM table1 " +
          "join table3 on table1.id2 = table3.id2 " +
          "join table2 on table3.id1 = table2.id1 ");
  
  TemporalTableFunction temporalFunction = staticJoined.createTemporalTableFunction( "rowtime" , "id2");
  tableEnv.registerFunction("CSVData", temporalFunction);
  
  tableEnv.registerTable("Joined",
      tableEnv.sqlQuery("SELECT * FROM KafkaInput, LATERAL TABLE (CSVData(KafkaInput.rowtime)) as Statics WHERE Statics.id2 = KafkaInput.id2")
  );
  </code></pre>
  
  <p>But I get an error with the TemporalTableFunction :</p>
  
  <pre><code>Exception in thread "main" java.lang.AssertionError: Cannot add expression of different type to set:
  set type is RecordType(BIGINT genTimestampMs, BIGINT timestampMs, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" streamId, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" sdkConfId, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" sdkId, FLOAT density, FLOAT count, FLOAT surface, TIMESTAMP(3) NOT NULL rowtime, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" streamId0, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" cameraName, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" streamId00, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" areaId, FLOAT coefficient, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" areaId0, FLOAT thresholdLow, FLOAT thresholdMedium, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" areaId1, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" name, TIMESTAMP(3) rowtime0, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" StationName) NOT NULL
  expression type is RecordType(BIGINT genTimestampMs, BIGINT timestampMs, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" streamId, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" sdkConfId, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" sdkId, FLOAT density, FLOAT count, FLOAT surface, TIMESTAMP(3) NOT NULL rowtime, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" streamId0, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" cameraName, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" streamId00, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" areaId, FLOAT coefficient, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" areaId0, FLOAT thresholdLow, FLOAT thresholdMedium, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" areaId1, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" name, TIMESTAMP(0) NOT NULL rowtime0, VARCHAR(65536) CHARACTER SET "UTF-16LE" COLLATE "ISO-8859-1$en_US$primary" StationName) NOT NULL
  set is rel#26:LogicalCorrelate.NONE(left=HepRelVertex#24,right=HepRelVertex#25,correlation=$cor0,joinType=inner,requiredColumns={8})
  expression is LogicalTemporalTableJoin#32
  </code></pre>
  
  <p>Where two fields do not match between the 'set type' and the 'expression type'.
  <code>TIMESTAMP(3) rowtime0</code> and <code>TIMESTAMP(0) NOT NULL rowtime0</code></p>
  
  <p>The problem is that I have no field named <code>rowtime0</code>. It look like it is an internal field. I don't really understand what's happening here</p>