2. sql解析原理

Apache Calcite面向Hadoop新的sql引擎，它提供了标准的SQL语言、多种查询优化和连接各种数据源的能力。除此之外，Calcite还提供了OLAP和流处理的查询引擎。它2013年成为了Apache孵化项目以来，在Hadoop中越来越引人注目，并被众多项目集成。比如Flink/Storm/Drill/Phoenix都依赖它做sql解析和优化。

先从demo跑起来，看看sql 解析都经历了什么工程？

(1) select 

package org.apache.flink.table.examples.java;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.Table;import org.apache.flink.table.api.java.StreamTableEnvironment;import java.util.Arrays;/** * Simple example for demonstrating the use of SQL on a Stream Table in Java.

*

* <p>This example shows how to:

* - Convert DataStreams to Tables

* - Register a Table under a name

* - Run a StreamSQL query on the registered Table

*

*/publicclass StreamSQLExample {

// *************************************************************************

// PROGRAM

// *************************************************************************publicstaticvoidmain(String[] args)throws Exception {

// set up execution environmentStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

DataStream orderA = env.fromCollection(Arrays.asList(

newOrder(1L, "beer", 3),

newOrder(1L, "diaper", 4),

newOrder(3L, "rubber", 2)));

DataStream orderB = env.fromCollection(Arrays.asList(

newOrder(2L, "pen", 3),

newOrder(2L, "rubber", 3),

newOrder(4L, "beer", 1)));

// convert DataStream to TableTable tableA = tEnv.fromDataStream(orderA, "user, product, amount");

// register DataStream as TabletEnv.registerDataStream("OrderB", orderB, "user, product, amount");

// union the two tablesTable result = tEnv.sqlQuery("SELECT * FROM " + tableA + " WHERE amount > 2 UNION ALL " + "SELECT * FROM OrderB WHERE amount < 2");

tEnv.toAppendStream(result, Order.class).print();

env.execute();

}

// *************************************************************************

// USER DATA TYPES

// *************************************************************************/** * Simple POJO.

*/publicstaticclass Order {

public Long user;

public String product;

publicint amount;

public Order() {

}

publicOrder(Long user, String product,int amount) {

this.user = user;

this.product = product;

this.amount = amount;

}

@Override

public String toString() {

return"Order{" + "user=" + user + ", product='" + product + '\'' + ", amount=" + amount + '}';

}

}

}

实现代码如下

override def sqlQuery(query: String): Table = {

val planner = getFlinkPlanner

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

if(null!= parsed && parsed.getKind.belongsTo(SqlKind.QUERY)) {

// validate the sql queryval validated = planner.validate(parsed)

// transform to a relational treeval relational = planner.rel(validated)

newTableImpl(this,new PlannerQueryOperation(relational.rel))

} else {

thrownew TableException(

"Unsupported SQL query! sqlQuery() only accepts SQL queries of type " + "SELECT, UNION, INTERSECT, EXCEPT, VALUES, and ORDER_BY.")

}

}

>>parse the sql query

在calcite中用SqlNode表示

public SqlSelect(SqlParserPos pos,

SqlNodeList keywordList,

SqlNodeList selectList,

SqlNode from,

SqlNode where,

SqlNodeList groupBy,

SqlNode having,

SqlNodeList windowDecls,

SqlNodeList orderBy,

SqlNode offset,

SqlNode fetch) {

super(pos);

this.keywordList = Objects.requireNonNull(keywordList !=null? keywordList :new SqlNodeList(pos));

this.selectList = selectList;

this.from = from;

this.where = where;

this.groupBy = groupBy;

this.having = having;

this.windowDecls = Objects.requireNonNull(windowDecls !=null? windowDecls :new SqlNodeList(pos));

this.orderBy = orderBy;

this.offset = offset;

this.fetch = fetch;

}

>>validate the sql query

SqlValidatorImpl验证sqlNode