import java.sql.Timestamp;
import java.util.Arrays;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
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.$;

class SimpleTimeIntervalJoin
{
    public static void main(String[] args) throws Exception
    {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//--------------------------------------------------------------------------------------

        // 构造订单数据
        DataStream<Tuple3<String,String,Timestamp>> ordersData = env.fromCollection(Arrays.asList(
                new Tuple3<>("001", "iphone",   new Timestamp(1545800002000L)),//2018-12-26 12:53:22
                new Tuple3<>("002", "mac",      new Timestamp(1545800003000L)),//2018-12-26 12:53:23
                new Tuple3<>("003", "book",     new Timestamp(1545800004000L)),//2018-12-26 12:53:24
                new Tuple3<>("004", "cup",      new Timestamp(1545800018000L)) //2018-12-26 12:53:38
        ));


        // 构造付款表
        DataStream<Tuple3<String,String,Timestamp>>paymentData= env.fromCollection(Arrays.asList(
            new Tuple3<>("001", "alipay",   new Timestamp(1545803501000L)),//2018-12-26 13:51:41
            new Tuple3<>("002", "card",     new Timestamp(1545803602000L)),//2018-12-26 13:53:22
            new Tuple3<>("003", "card",     new Timestamp(1545803610000L)),//2018-12-26 13:53:30
            new Tuple3<>("004", "alipay",   new Timestamp(1545803611000L)) //2018-12-26 13:53:31
         ));

//--------------------------------------------------------------------------------------


        Table orders=tEnv.fromDataStream(ordersData.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String,String,Timestamp>>(Time.milliseconds(100)) {
            @Override
            public long extractTimestamp(Tuple3<String,String,Timestamp> element)
            {

                return element.f2.getTime();
            }
        }),$("orderId"),$("productName"),$("orderTime").rowtime());




        Table ratesHistory=tEnv.fromDataStream(paymentData.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String,String,Timestamp>>(Time.milliseconds(100)) {
            @Override
            public long extractTimestamp(Tuple3<String,String,Timestamp> element)
            {

                return element.f2.getTime();
            }
        }),$("orderId"),$("payType"),$("payTime").rowtime());

//--------------------------------------------------------------------------------------
        tEnv.registerTable("Orders", orders);
        tEnv.registerTable("Payment", ratesHistory);

//--------------------------------------------------------------------------------------
        String sqlQuery="select o.orderId," +
                        "o.productName," +
                        "p.payType," +
                        "o.orderTime," +
                        "cast(payTime as timestamp) as payTime " +
                        "from " +
                        "Orders AS o JOIN Payment AS p ON o.orderId = p.orderId AND " +
                        "p.payTime BETWEEN orderTime AND orderTime + INTERVAL '1' HOUR";


                   Table result= tEnv.sqlQuery(sqlQuery);

//--------------------------------------------------------------------------------------
       DataStream<Row> Result = tEnv.toAppendStream(result, Row.class);
       Result.print();


        env.execute();



    }
}