[FLINK-21013] Blink planner does not ingest timestamp into StreamRecord

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.12.0
• Fix Version/s: 1.11.4, 1.12.2, 1.13.0
• Component/s: Table SQL / Planner, Table SQL / Runtime
• Labels: pull-request-available
• Assignee: Leonard Xu
• Reporter: Timo Walther
• Created: 18/Jan/21 16:23
• Updated: 28/May/21 09:02
• Resolved: 09/Feb/21 10:38

Description

Currently, the rowtime attribute is not put into the StreamRecord when leaving the Table API to DataStream API. The legacy planner supports this, but the timestamp is null when using the Blink planner.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
            EnvironmentSettings settings =
                    EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
            StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        DataStream<Order> orderA =
                env.fromCollection(
                        Arrays.asList(
                                new Order(1L, "beer", 3),
                                new Order(1L, "diaper", 4),
                                new Order(3L, "rubber", 2)));

        DataStream<Order> orderB =
                orderA.assignTimestampsAndWatermarks(
                        new AssignerWithPunctuatedWatermarks<Order>() {
                            @Nullable
                            @Override
                            public Watermark checkAndGetNextWatermark(
                                    Order lastElement, long extractedTimestamp) {
                                return new Watermark(extractedTimestamp);
                            }

                            @Override
                            public long extractTimestamp(Order element, long recordTimestamp) {
                                return element.user;
                            }
                        });

        Table tableA = tEnv.fromDataStream(orderB, $("user").rowtime(), $("product"), $("amount"));

        // union the two tables
        Table result = tEnv.sqlQuery("SELECT * FROM " + tableA);

        tEnv.toAppendStream(result, Row.class)
                .process(
                        new ProcessFunction<Row, Row>() {
                            @Override
                            public void processElement(Row value, Context ctx, Collector<Row> out)
                                    throws Exception {
                                System.out.println("TIMESTAMP" + ctx.timestamp());
                            }
                        })
                .print();

        env.execute();

Issue Links

• links to: GitHub Pull Request #14785
• links to: GitHub Pull Request #14787