>>validate the sql query

SqlValidatorImpl验证sqlNode

public SqlNode validate(SqlNode topNode) {

SqlValidatorScope scope =newEmptyScope(this);

scope =newCatalogScope(scope, ImmutableList.of("CATALOG"));

finalSqlNode topNode2 = validateScopedExpression(topNode, scope);

finalRelDataType type = getValidatedNodeType(topNode2);

Util.discard(type);

return topNode2;

}

>>transform to a relational tree

SqlToRelConverter.java

/** * Converts an unvalidated query's parse tree into a relational expression.

*

* @param query Query to convert

* @param needsValidation Whether to validate the query before converting;

* <code>false</code> if the query has already been

* validated.

* @param top Whether the query is top-level, say if its result

* will become a JDBC result set; <code>false</code> if

* the query will be part of a view.

*/public RelRoot convertQuery(

SqlNode query,

finalboolean needsValidation,

finalboolean top) {

if (needsValidation) {

query = validator.validate(query);

}

RelMetadataQuery.THREAD_PROVIDERS.set(

JaninoRelMetadataProvider.of(cluster.getMetadataProvider()));

RelNode result = convertQueryRecursive(query, top,null).rel;

if (top) {

if (isStream(query)) {

result =new LogicalDelta(cluster, result.getTraitSet(), result);

}

}

RelCollation collation = RelCollations.EMPTY;

if(!query.isA(SqlKind.DML)) {

if (isOrdered(query)) {

collation = requiredCollation(result);

}

}

checkConvertedType(query, result);

if (SQL2REL_LOGGER.isDebugEnabled()) {

SQL2REL_LOGGER.debug(

RelOptUtil.dumpPlan("Plan after converting SqlNode to RelNode",

result, SqlExplainFormat.TEXT,

SqlExplainLevel.EXPPLAN_ATTRIBUTES));

}

finalRelDataType validatedRowType = validator.getValidatedNodeType(query);

return RelRoot.of(result, validatedRowType, query.getKind())

.withCollation(collation);

}

(2)update

代码实现

override def sqlUpdate(stmt: String): Unit = {

sqlUpdate(stmt, this.queryConfig)

}

override def sqlUpdate(stmt: String, config: QueryConfig): Unit = {

val planner = getFlinkPlanner

// parse the sql queryval parsed = planner.parse(stmt)

parsed match {

caseinsert: SqlInsert =>// validate the SQL queryval query = insert.getSource

val validatedQuery = planner.validate(query)

// get query result as Tableval queryResult =newTableImpl(this,

new PlannerQueryOperation(planner.rel(validatedQuery).rel))

// get name of sink tableval targetTablePath = insert.getTargetTable.asInstanceOf[SqlIdentifier].names

// insert query result into sink tableinsertInto(queryResult, config, targetTablePath.asScala:_*)

case_ =>thrownew TableException(

"Unsupported SQL query! sqlUpdate() only accepts SQL statements of type INSERT.")

}

}

步骤类似，不再赘述。

3. 总结

Flink Table API&SQL 为流式数据和静态数据的关系查询保留统一的接口，而且利用了Calcite的查询优化框架和SQL parser。该设计是基于Flink已构建好的API构建的，DataStream API 提供低延时高吞吐的流处理能力而且就有exactly-once语义而且可以基于event-time进行处理。而且DataSet拥有稳定高效的内存算子和流水线式的数据交换。Flink的core API和引擎的所有改进都会自动应用到Table API和SQL上。