Hi
We're using multi sink in sql with view, the TestCase is
"""java
@Test
def testJoinTemporalTableWithViewWithFilterPushDown(): Unit = {
createLookupTable("LookupTableAsync1", new AsyncTableFunction1)
util.addTable(
"""
|CREATE TEMPORARY VIEW v_vvv AS
|SELECT * FROM MyTable AS T
|JOIN LookupTableAsync1 FOR SYSTEM_TIME AS OF T.proctime AS D
|ON T.a = D.id
|""".stripMargin)
val stmtSet = util.tableEnv.createStatementSet()
val appendSink1 = util.createRetractTableSink(
Array("a", "b","id","name"),
Array(INT, STRING, INT, STRING))
util.tableEnv.asInstanceOf[TableEnvironmentInternal].registerTableSinkInternal(
"appendSink1", appendSink1)
stmtSet.addInsert("appendSink1", util.tableEnv.sqlQuery("""
|SELECT
a,b,id,name FROM v_vvv
|WHERE age =
10
""".stripMargin))
stmtSet.addInsert("appendSink1", util.tableEnv.sqlQuery("""
|SELECT
a,b,id,name FROM v_vvv
|WHERE age =
30
""".stripMargin))
// util.verifyPlan(stmtSet)
util.verifyExecPlan(stmtSet)
}
class AsyncTableFunction1 extends AsyncTableFunction[RowData] {
def eval(resultFuture: CompletableFuture[JCollection[RowData]], a:
Integer): Unit = {
}
}
"""
the optimized exec plan is
"""
Calc(select=[a, b])(reuse_id=[1])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
LegacySink(name=[`default_catalog`.`default_database`.`appendSink1`],
fields=[a, b, id, name])
+- LookupJoin(table=[default_catalog.default_database.LookupTableAsync1],
joinType=[InnerJoin], async=[true], lookup=[age=10, id=a], where=[(age =
10)], select=[a, b, id, name])
+- Reused(reference_id=[1])
LegacySink(name=[`default_catalog`.`default_database`.`appendSink2`],
fields=[a, b, id, name])
+- LookupJoin(table=[default_catalog.default_database.LookupTableAsync1],
joinType=[InnerJoin], async=[true], lookup=[age=30, id=a], where=[(age =
30)], select=[a, b, id, name])
+- Reused(reference_id=[1])
"""
*I have 2 questions*
*1. The lookup table function execute twice, which is very expensive*
*2. the age filter is push down to LookupJoin with lookup=[age=10, id=a],
which result to function signature mismatch (exception follows blow)*
org.apache.flink.table.api.ValidationException: Could not find an
implementation method 'eval' in class
'org.apache.flink.table.planner.plan.utils.AsyncTableFunction1' for
function 'default_catalog.default_database.LookupTableAsync1, source:
[TestInvalidTemporalTable(id, name, age, ts)]' that matches the following
signature:
void eval(java.util.concurrent.CompletableFuture, java.lang.Integer,
java.lang.Integer)
Is the optimizer wrong or I'm wrong ?
Cooper.Luan
