Apache-Flink深度解析-JOIN-LATERAL

CREATE TABLE Customers (
  customerid char(5) NOT NULL,
  city varchar (10) NOT NULL
)

insert into Customers values('C001','Beijing');
insert into Customers values('C002','Beijing');
insert into Customers values('C003','Beijing');
insert into Customers values('C004','HangZhou');

CREATE TABLE Orders(
  orderid char(5) NOT NULL,
  customerid char(5) NULL
)

insert into Orders values('O001','C001');
insert into Orders values('O002','C001');
insert into Orders values('O003','C003');
insert into Orders values('O004','C001');

SELECT 
    c.customerid, c.city, o.orderid 
FROM Customers c JOIN Orders o 
    ON o.customerid = c.customerid

SELECT 
    c.customerid, c.city
FROM Customers c WHERE c.customerid IN (
    SELECT 
      o.customerid, o.orderid
    FROM Orders o
    WHERE o.customerid = c.customerid
)

SELECT 
    c.customerid, c.city
FROM Customers c WHERE c.customerid IN (
    SELECT 
      o.customerid
    FROM Orders o
    WHERE o.customerid = c.customerid
)

SELECT 
    c.customerid, c.city, o.orderid 
FROM Customers c, (
    SELECT 
      o.orderid, o.customerid 
    FROM Orders o
    WHERE o.customerid = c.customerid
) as o

SELECT 
    c.customerid, c.city, o.orderid 
FROM Customers c CROSS APPLY (
    SELECT 
     TOP(1) o.orderid, o.customerid 
    FROM Orders o 
    WHERE o.customerid = c.customerid
    ORDER BY o.customerid, o.orderid
) as o

SELECT  c.customerid, c.city, o.orderid
 FROM    Customers c
  JOIN (
   SELECT 
     o2.*, 
     ROW_NUMBER() OVER (
        PARTITION BY customerid 
        ORDER BY orderid
      ) AS rn
   FROM    Orders o2
 ) o
ON c.customerid = o.customerid AND o.rn = 1

SELECT 
   name, log_backup_time 
FROM sys.databases AS s
 CROSS APPLY sys.dm_db_log_stats(s.database_id); 

SELECT 
    e.NAME, e.DEPTNO, d.NAME 
FROM EMPS e, LATERAL (
    SELECT 
    *
    FORM DEPTS d 
    WHERE e.DEPTNO=d.DEPTNO
 ) as d;

/**
  * Base class for all user-defined functions such as scalar functions, table functions,
  * or aggregation functions.
  */
abstract class UserDefinedFunction extends Serializable {
  // 关键是FunctionContext中提供了若干高级属性（在UDX篇会详细介绍）
  def open(context: FunctionContext): Unit = {}
  def close(): Unit = {}
}

/**
  * Base class for a user-defined table function (UDTF). A user-defined table functions works on
  * zero, one, or multiple scalar values as input and returns multiple rows as output.
  *
  * The behavior of a [[TableFunction]] can be defined by implementing a custom evaluation
  * method. An evaluation method must be declared publicly, not static and named "eval".
  * Evaluation methods can also be overloaded by implementing multiple methods named "eval".
  *
  * User-defined functions must have a default constructor and must be instantiable during runtime.
  *
  * By default the result type of an evaluation method is determined by Flink's type extraction
  * facilities. This is sufficient for basic types or simple POJOs but might be wrong for more
  * complex, custom, or composite types. In these cases [[TypeInformation]] of the result type
  * can be manually defined by overriding [[getResultType()]].
  */
abstract class TableFunction[T] extends UserDefinedFunction {

  // 对于泛型T，如果是基础类型那么Flink框架可以自动识别，
  // 对于用户自定义的复杂对象，需要用户overwrite这个实现。
  def getResultType: TypeInformation[T] = null
}

// 定义一个简单的UDTF返回类型，对应接口上的 T 
case class SimpleUser(name: String, age: Int)
// 继承TableFunction，并实现evale方法
// 核心功能是解析以#分割的字符串
class SplitTVF extends TableFunction[SimpleUser] {
  // make sure input element's format is "<string>#<int>"
  def eval(user: String): Unit = {
    if (user.contains("#")) {
      val splits = user.split("#")
      collect(SimpleUser(splits(0), splits(1).toInt))
    }
  }
}

@Test
  def testLateralTVF(): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tEnv = TableEnvironment.getTableEnvironment(env)
    env.setStateBackend(getStateBackend)
    StreamITCase.clear

    val userData = new mutable.MutableList[(String)]
    userData.+=(("Sunny#8"))
    userData.+=(("Kevin#36"))
    userData.+=(("Panpan#36"))

    val SQLQuery = "SELECT data, name, age FROM userTab, LATERAL TABLE(splitTVF(data)) AS T(name, age)"

    val users = env.fromCollection(userData).toTable(tEnv, 'data)

    val tvf = new SplitTVF()
    tEnv.registerTable("userTab", users)
    tEnv.registerFunction("splitTVF", tvf)

    val result = tEnv.SQLQuery(SQLQuery).toAppendStream[Row]
    result.addSink(new StreamITCase.StringSink[Row])
    env.execute()
    StreamITCase.testResults.foreach(println(_))
  }

val SQLQuery = "SELECT data, name, age FROM userTab, LATERAL TABLE(splitTVF(data)) AS T(name, age)"