/opt/flink/bin/flink run -m yarn-cluster -ynm KafkaSourceHbaseSinkCaseTestSql \
-c ${mainClassName} \
${jarPath} \
--flink.parallelism 40 \
--mode stream \
--sql.file.path ${sqlFile}

create table kafka_table(
  ts  bigint,
  username string,
  num bigint
) with (
  'connector' = 'kafka',
  'topic' = 'test',
  'properties.bootstrap.servers' = '',
  'properties.group.id' = 'test-consumer001',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'csv'
)
create table sink_table(
  ts  bigint,
  username string,
  num bigint
) with (
  'connector' = 'kafka',
  'topic' = 'test2',
  'properties.bootstrap.servers' = '',
  'format' = 'json'
)
insert into sink_table select * from kafka_table

* {{{-
*  val sourceTable = tEnv.scan("test_table").select('a, 'b, 'c)
*  val leftTable = sourceTable.filter('a > 0).select('a as 'a1, 'b as 'b1)
*  val rightTable = sourceTable.filter('c.isNotNull).select('b as 'b2, 'c as 'c2)
*  val joinTable = leftTable.join(rightTable, 'a1 === 'b2)
*  joinTable.where('a1 >= 70).select('a1, 'b1).writeToSink(sink1)
*  joinTable.where('a1 < 70 ).select('a1, 'c2).writeToSink(sink2)
* }}}
*
* the RelNode DAG is:
*
* {{{-
* Sink(sink1)     Sink(sink2)
*    |               |
* Project(a1,b1)  Project(a1,c2)
*    |               |
* Filter(a1>=70)  Filter(a1<70)
*       \          /
*        Join(a1=b2)
*       /           \
* Project(a1,b1)  Project(b2,c2)
*      |             |
* Filter(a>0)     Filter(c is not null)
*      \           /
*      Project(a,b,c)
*          |
*       TableScan
* }}}
* This [[RelNode]] DAG will be decomposed into three [[RelNodeBlock]]s, the break-point
* is the [[RelNode]](`Join(a1=b2)`) which data outputs to multiple [[LegacySink]]s.
* <p>Notes: Although `Project(a,b,c)` has two parents (outputs),
* they eventually merged at `Join(a1=b2)`. So `Project(a,b,c)` is not a break-point.
* <p>the first [[RelNodeBlock]] includes TableScan, Project(a,b,c), Filter(a>0),
* Filter(c is not null), Project(a1,b1), Project(b2,c2) and Join(a1=b2)
* <p>the second one includes Filter(a1>=70), Project(a1,b1) and Sink(sink1)
* <p>the third one includes Filter(a1<70), Project(a1,c2) and Sink(sink2)
* <p>And the first [[RelNodeBlock]] is the child of another two.
*
* The [[RelNodeBlock]] plan is:
* {{{-
* RelNodeBlock2  RelNodeBlock3
*        \            /
*        RelNodeBlock1
* }}}

create table source_table(
  data string,
  topic string,
) with (
 'connector' = 'kafka',
 'topic' = 'topic',
 'properties.bootstrap.servers' = '',
 'format' = 'csv'
)
create table sink_table1(
  data string,
) with (
 'connector' = 'kafka',
 'topic' = 'topic2',
 'properties.bootstrap.servers' = '',
 'format' = 'csv'
)
create table sink_table2(
  data string,
) with (
 'connector' = 'kafka',
 'topic' = 'topic2',
 'properties.bootstrap.servers' = '',
 'format' = 'csv'
)
insert into sink_table1 select * from (select SBSTR(data, 0, 6) data,topic from source_table) where topic='topic1'
insert into sink_table2 select * from (select SBSTR(data, 0, 6) data,topic from source_table) where topic='topic2'

/**
    * Reuse common sub-plan in different RelNode tree, generate a RelNode dag
    *
    * @param relNodes RelNode trees
    * @return RelNode dag which reuse common subPlan in each tree
    */
  private def reuseRelNodes(relNodes: Seq[RelNode], tableConfig: TableConfig): Seq[RelNode] = {
    val findOpBlockWithDigest = tableConfig.getConfiguration.getBoolean(
      RelNodeBlockPlanBuilder.TABLE_OPTIMIZER_REUSE_OPTIMIZE_BLOCK_WITH_DIGEST_ENABLED)
    if (!findOpBlockWithDigest) {
      return relNodes
    }
    // reuse sub-plan with same digest in input RelNode trees.
    val context = new SubplanReuseContext(true, relNodes: _*)
    val reuseShuttle = new SubplanReuseShuttle(context)
    relNodes.map(_.accept(reuseShuttle))
  }

// 创建视图
StreamTableEnvironment.executeSql("create view as select SBSTR(data, 0, 6) data,topic from source_table);
// 注册表
Table table = StreamTableEnvironment.executeSql("select SBSTR(data, 0, 6) data,topic from source_table");
StreamTableEnvironment.registerTable("tmp", table);

REGISTER TABLE TABLE_NAME AS SELECT_QUERY
// 示例
register table tmp as select SBSTR(data, 0, 6) data,topic from source_table

package org.apache.flink.sql.parser.ddl;
import org.apache.calcite.sql.*;
import org.apache.calcite.sql.parser.SqlParserPos;
import javax.annotation.Nonnull;
import java.util.Collections;
import java.util.List;
public class SqlRegisterTable extends SqlCall {
public static final SqlSpecialOperator OPERATOR = new SqlSpecialOperator("REGISTER TABLE", SqlKind.OTHER_DDL);
private SqlIdentifier tableName;
private SqlNode query;
     // 部分代码省略
 }

SqlRegisterTable SqlRegisterTable() :
  {
    SqlIdentifier tableName = null;
    SqlNode query = null;
    SqlParserPos pos;
    }
  {
    <REGISTER> <TABLE> { pos = getPos();}
    tableName = CompoundIdentifier()
    <AS>
    query = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)
    {
        return new SqlRegisterTable(pos, tableName, query);
        }
  }

package org.apache.flink.table.operations.ddl;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
public class RegisterTableOperation implements Operation{
  private final ObjectIdentifier tableIdentifier;
  private final QueryOperation query;
    // 省略部分代码
}

private Operation convertRegisterTable(SqlRegisterTable sqlRegisterTable) {
    UnresolvedIdentifier unresolvedIdentifier = UnresolvedIdentifier.of(sqlRegisterTable.fullTableName());
    ObjectIdentifier identifier = catalogManager.qualifyIdentifier(unresolvedIdentifier);
    PlannerQueryOperation operation = toQueryOperation(flinkPlanner, validateQuery);
    return new RegisterTableOperation(identifier, operation);
  }

@Override
  public TableResult executeSql(String statement) {
    List<Operation> operations = parser.parse(statement);
    if (operations.size() != 1) {
      throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG);
    }
    return executeOperation(operations.get(0));
  }