package com.mypackage.streaming;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import static org.apache.flink.table.api.Expressions.$;

public class HelloFlinkWindowJoin {
    public static void main(String[] args) throws Exception {
        System.out.println("Start Flink Application");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Configuration configuration = tableEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "UTC");

        DataStream<Row> dataStream = env.fromElements(
            Row.of("U1234", "ORDER_ID1", 1, 1676900949000L),
            Row.of("U1234", "ORDER_ID2", 1, 1676910949000L),
            Row.of("U1234", "ORDER_ID3", 2, 1676920949000L),
            Row.of("U1234", "ORDER_ID4", 2, 1676930949000L),
            Row.of("U1234", "ORDER_ID5", 3, 1676940949000L),
            Row.of("U1234", "ORDER_ID6", 4, 1676950949000L),
            Row.of("U1234", "ORDER_ID7", 5, 1676960949000L)

        ).returns(
            Types.ROW(
                Types.STRING,
                Types.STRING,
                <http://Types.INT|Types.INT>,
                Types.LONG));
        dataStream = dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps() );

        Table table = tableEnv.fromDataStream(dataStream,$("user_id"), $("order_id"), $("amount"), $("event_time").rowtime());
        table.printSchema();
        table.execute().print();

        tableEnv.createTemporaryView("foodorder", table);
        String sql = "WITH food AS (" +
            " SELECT user_id, window_start, window_end" +
            " FROM TABLE(TUMBLE(TABLE foodorder, DESCRIPTOR(event_time), INTERVAL '1' MINUTES))" +
            " GROUP BY user_id, window_start, window_end)\n"+
            "select food.window_start\n" +
            "     ,food.window_end as id\n" +
            "     ,food.user_id as id\n" +
            "     ,DATE_FORMAT(food.window_end + INTERVAL '7' HOUR, 'yyyyMMdd') AS dt\n" +
            "     ,DATE_FORMAT(food.window_end + INTERVAL '7' HOUR, 'HH') AS `hour`\n" +
            "from food\n" +
            "left join food as a on food.user_id = a.user_id\n" +
            "and food.window_start = a.window_start\n" +
            "and food.window_end = a.window_end";
        String sinkSQL = "CREATE TABLE sink (\n" +
            "    window_start TIMESTAMP\n" +
            "    ,window_end TIMESTAMP\n" +
            "    ,user_id STRING\n" +
            "    ,dt STRING\n" +
            "    ,`hour` STRING\n" +
            ") WITH ( \n" +
            "    'connector' = 'print'\n" +
            ");";
        System.out.println(sinkSQL);
        tableEnv.executeSql(sinkSQL);
        table = tableEnv.sqlQuery(sql);
        StatementSet statementSet = tableEnv.createStatementSet();
        statementSet.add(table.insertInto("sink"));
        statementSet.add(table.insertInto("sink"));
        statementSet.printExplain();
        statementSet.execute();
    }
}