flink sql 知其是以然（六）| flink sql 約會 calcite（看這篇就夠了）

1.序篇-本文結構

全網第一個 flink sql 實戰，本文主要介紹 flink sql 與 calcite 之間的關系。flink sql 的解析主要依賴 calcite。

而部落客通過此文抛磚引玉幫助大家了解 flink sql 在解析中是怎樣依賴 calcite 的，以及 flink sql 解析的流程，sql parser 相關内容。希望對大家有所幫助。

本文通過以下幾節進行介紹，對某個章節感興趣的可以直接劃到對應章節。

1. 背景篇-一條 flink sql 的執行過程
   • 發揮自己的想象力
   • 看看 flink 的實作
2. 簡介篇-calcite 扮演的角色
   • calcite 是啥？
   • flink sql 為啥選擇 calcite？
3. 案例篇-calcite 的能力、案例
   • 先用用 calcite
   • 關系代數
   • calcite 必知的基礎 model
   • calcite 的處理流程（以 flink sql 為例）
   • calcite 怎麼做到這麼通用？
4. 原理剖析篇-calcite 在 flink sql 中大展身手
   • FlinkSqlParserImpl
   • FlinkSqlParserImpl 的生成
5. 總結與展望篇

2.背景篇-一條 flink sql 的執行過程

本節先給大家大緻描述

一條 flink sql 的執行過程

，不了解詳細内容不要緊，主要先了解整個流程，有了全局視角之後，後續會詳述細節。

在介紹一條 flink sql 的執行過程之前，先來看看 flink datastream 任務的執行過程，這對了解一條 flink sql 的執行過程有很大的幫助。

• datastream：datastream 在使用時要在 flink datastream api 提供的各種 udf（比如 flatmap，keyedprocessfunction 等）中自定義處理邏輯，具體的業務執行邏輯都是敲代碼、 java 檔案寫的，然後編譯在 jvm 中執行，就和一個普通的 main 函數應用一模一樣的流程。因為代碼執行邏輯都是自己寫的，是以這一部分相對好了解。
• sql：java 編譯器不能識别和編譯一條 sql 進行執行，那麼一條 SQL 是咋執行的呢？

2.1.先發揮自己的想象力

我們逆向思維進行考慮，如果想讓一條 flink sql 按照我們的預期在 jvm 中執行，需要哪些過程。

1. 整體來說：參考 datastream，如果 jvm 能執行 datastream java code 編譯後的 class 檔案，那麼加一個 sql 解析層，能将 sql 邏輯解析為 datastream 的各種算子，然後編譯執行不就 vans 了。
2. sql parser：首先得有一個 sql parser 吧，得先能識别 sql 文法，将 sql 文法轉化為 AST、具體的關系代數。
3. 關系代數到 datastream 算子的映射：sql 邏輯解析為 datastream，需要有一個解析的映射邏輯吧。sql 是基于關系代數的，可以維護一個 sql 中的每個關系代數到具體 datastream 接口的映射關系，有了這些映射關系我們就可以将 sql 映射成一段可執行的 datastream 代碼。舉個例子：其可以将：
   • sql select xxx 解析為類似 datastream 中的 map
   • where xxx 解析為 filter
   • group by 解析成 keyby
   • sum（xx），count（xxx）可以解析為 datastream 中的 aggregate function
   • etc...
4. 代碼生成：有了 sql AST，sql 到 datasretam 算子的映射關系之後，就要進行具體的代碼生成了。比如去解析 sql AST 中具體哪些字段用作 where 邏輯，哪些字段用作 group by，都需要生成對應具體的 datastream 代碼。
5. 運作：經過上述流程之後，就可以将一個 sql 翻譯成一個 datastream 作業了，happy 的執行。

如下圖所示，描繪了上述邏輯：

12

那麼這個和 flink 實際實作有啥異同呢？

flink 大緻是這樣做的，雖在 flink 本身的中間還有一些其他的流程，後來的版本也不是基于 datastream，但是整體的處理邏輯還是和上述一緻的。

是以不了解整體流程的同學可以先按照上述流程進行了解。

按照

部落客的腦洞

來總結一條 sql 的使命就是：

sql -> AST -> codegen(java code) -> 讓我們 run 起來好嗎

2.2.看看 flink 的實作

26

上面手繪可能看不清，下面這張圖更清楚。

28

标準的一條 flink sql 運作起來的流程如下：

Notes：剛開始對其中的 SqlNode，RelNode 概念可能比較模糊。先了解整個流程，後續會詳細介紹這些概念。

1. sql 解析階段：calcite parser 解析（sql -> AST，AST 即 SqlNode Tree）
2. SqlNode 驗證階段：calcite validator 校驗（SqlNode -> SqlNode，文法、表達式、表資訊）
3. 語義分析階段：SqlNode 轉換為 RelNode，RelNode 即 Logical Plan（SqlNode -> RelNode）
4. 優化階段：calcite optimizer 優化（RelNode -> RelNode，剪枝、謂詞下推等）
5. 實體計劃生成階段：Logical Plan 轉換為 Physical Plan（等同于 RelNode 轉換成 DataSet\DataStream API）
6. 後續的運作邏輯與 datastream 一緻

可以發現

flink 的實作

比

部落客的腦洞

整體主要架構上面是一緻的。多出來的部分主要是 SqlNode 驗證階段，優化階段。

3.簡介篇-calcite 在 flink sql 中的角色

大緻了解了

一條 flink sql 的運作流程

之後，我們來看看 calcite 這玩意到底在 flink 裡幹了些啥。

根據上文總結來說 calcite 在 flink sql 中擔當了

sql 解析、驗證、優化

功能。

30

看着 calcite 幹了這麼多事，那 calcite 是個啥東東，它的定位是啥？

3.1.calcite 是啥？

calcite 是一個動态資料的管理架構，它可以用來建構資料庫系統的不同的解析的子產品，但是它不包含資料存儲資料處理等功能。

calcite 的目标是一種方案，适應所有的需求場景，希望能為不同計算平台和資料源提供統一的 sql 解析引擎，但是它隻是提供查詢引擎，而沒有真正的去存儲這些資料。

61

下圖是目前使用了 calcite 能力的其他元件，也可見官網 https://calcite.apache.org/docs/powered_by.html 。

4

簡單來說的話，可以先了解為 calcite 具有這幾個功能（當然還有其他很牛逼的功能，感興趣可以自查官網）。

1. 自定義 sql 解析器：比如說我們新發明了一個引擎，然後我們要在這個引擎上來創造一套基于 sql 的接口，那麼我們就可以使用直接 calcite，不用自己去寫一套專門的 sql 的解析器，以及執行以及優化引擎，calcite 人都有。
2. sql parser（extends SqlAbstractParserImpl）：将 sql 的各種關系代數解析為具體的 AST，這些 AST 都能對應到具體的 java model，在 java 的世界裡面，對象很重要，有了這些對象（

   SqlSelect

   、

   SqlNode

   ），就可以根據這些對象做具體邏輯處理了。舉個例子，如下圖，一條簡單的

   select c,d from source where a = '6'

   sql，經過 calcite 的解析之後，就可以得到 AST model（SqlNode）。可以看到有

   SqlSelect

   、

   SqlIdentifier

   、

   SqlIdentifier

   、

   SqlCharStringLiteral

   。
3. sql validator（extends SqlValidatorImpl）：根據文法、表達式、表資訊進行 SqlNode 正确性校驗。
4. sql optimizer：剪枝、謂詞下推等優化

上面的這些能力整體組成如下圖所示：

29

實際使用 calcite 解析一條 sql，跑起來看看。

2

3.2.flink sql 為什麼選擇 calcite？

1. 不用重複造輪子。有限的精力應該放在有價值的事情上。
2. calcite 有針對 stream 表的解決方案。具體可見 https://calcite.apache.org/docs/stream.html。

4.案例篇-calcite 的能力、案例

4.1.先用用 calcite

重中之重，在了解原理之前，先跑起來是王道，也會幫助我們逐漸了解。

官網已經有一個 csv 的案例了。感興趣的可以直達 https://calcite.apache.org/docs/tutorial.html 。

跑完一個 csv demo，在詳細了解 calcite 之前還需要了解下 sql，calcite 的支柱：關系代數。

4.2.關系代數

sql 是基于關系代數的查詢語言，是關系代數在工程上的一種很好的實作方案。在工程中，關系代數難表達，但是 sql 就易于了解。關系代數和 sql 的關系如下。

1. 可以将一條 sql 解析為一個關系代數表達式的組合。在 sql 中的操作都可以轉化成關系代數的表達式。
2. sql 的執行優化（所有的優化的前提都是優化前和優化後最終執行結果相同，即等價交換）是基于關系代數運算的。

4.2.1.常用關系代數

總結下，有哪些常用的關系代數：

50

4.2.2.sql 優化支柱之關系代數等價變換

關系代數等價變換是 calcite optimizer 的基礎理論。

下面是一些等價變換的例子。

1.連接配接（

⋈

），笛卡爾積（

×

）的交換律

51

2.連接配接（

⋈

），笛卡爾積（

×

）的結合律

3.投影（

Π

）的串接定律

4.選擇（

σ

）的串接定律

5.選擇（

σ

）與投影（

Π

）的交換

6.選擇（

σ

）與笛卡爾積（

×

）的交換

7.選擇（

σ

）與并（

∪

）的交換

8.選擇（

σ

）與差（

-

）的交換

9.投影（

Π

）與笛卡爾積（

×

）的交換

10.投影（

Π

）與并（

∪

）的交換

然後看一個基于關系代數優化的實際 sql 案例：

有三個關系

A（a1,a2,a3,…）

、

B（b1,b2,b3, … ）

、

C（a1,b1,c1,c2, … ）

有一個查詢請求如下：

SELECT A.a1 FROM A，B，C WHERE A.a1 = C.a1 AND B.b1 = C.b1 AND f(c1)

1.首先将 sql 轉為關系代數的文法樹。

36

2.優化：選擇（

σ

）的串接定律。

47

37

3.優化：選擇（

σ

）與笛卡爾積（

×

）的交換。

48

38

4.優化：投影（

π

）與笛卡爾積（

×

）的交換。

49

關于關系代數我們就有了大緻的了解。

除此之外，對于更深入了解 flink sql，calcite 而言，我們還需要了解一下在 calcite 代碼體系中有哪些重要 model。

4.3.calcite 必知的基礎 model

calcite 中有兩個最最基礎、重要的 model 在我們了解 flink sql 解析流程時需要知道的。

1. SqlNode：sql 轉化而成，可以了解為直覺表達 sql 層次結構的的 model
2. RelNode：SqlNode 轉化而成，可以了解為将 SqlNode 轉化為關系代數，表達關系代數層次結構的 model

舉個例子來說明下，下面這條 flink sql，經過解析之後的 SqlNode，RelNode 如下圖：

SELECT
  sum(part_pv) as pv,
  window_start
FROM (
 SELECT
   count(1) as part_pv,
   cast(tumble_start(rowtime, INTERVAL '60' SECOND) as bigint) * 1000 as window_start
 FROM
   source_db.source_table
 GROUP BY
   tumble(rowtime, INTERVAL '60' SECOND)
   , mod(id, 1024)
)
GROUP BY
  window_start
           

複制

62

可以看到 SqlNode 包含的内容是 sql 的層次結構，包括

selectList

，

from

，

where

，

group by

等。

RelNode 包含的是關系代數的層次結構，每一層都有一個 input 來承接。結合上面優化案例的樹狀結構一樣。

63

4.4.calcite 的處理流程（以 flink sql 為例）

29

如上圖所示，此處我們結合上節介紹的 calcite 的 model，以及 flink sql 的實作來走一遍其處理流程：

1. sql 解析階段（sql –> SqlNode）
2. SqlNode 驗證（SqlNode –> SqlNode）
3. 語義分析（SqlNode –> RelNode）
4. 優化階段（RelNode –> RelNode）

4.4.1.flink sql demo

SELECT
  sum(part_pv) as pv,
  window_start
FROM (
 SELECT
   count(1) as part_pv,
   cast(tumble_start(rowtime, INTERVAL '60' SECOND) as bigint) * 1000 as window_start
 FROM
   source_db.source_table
 GROUP BY
   tumble(rowtime, INTERVAL '60' SECOND)
   , mod(id, 1024)
)
GROUP BY
  window_start
           

複制

其中前三步解析和轉化，都在 在執行

TableEnvironment#sqlQuery

進行。

最後一步優化，在執行 sink 操作時進行，即在這個例子中是

tEnv.toRetractStream(result, Row.class)

。

源碼公衆号背景回複flink sql 知其是以然（六）| flink sql 約會 calcite擷取。

4.4.2.sql 解析階段（sql –> SqlNode）

sql 解析階段使用 Sql Parser 将 sql 解析為

SqlNode

。這一步在執行

TableEnvironment#sqlQuery

進行。

可以從上圖看到 flink sql 具體實作類是

FlinkSqlParserImpl

。

68

具體 parse 得到 SqlNode 如上圖。

4.4.3.SqlNode 驗證（SqlNode –> SqlNode）

上面的第一步生産的 SqlNode 對象是一個未經驗證的，這一步就是文法檢查階段，文法檢查前需要知道中繼資料資訊，這個檢查會包括表名、字段名、函數名、資料類型的檢查。進行文法檢查的實作如下：

可以從上圖看到 flink sql 校驗器的具體實作類是

FlinkCalciteSqlValidator

，其中包含了中繼資料資訊，進而可以進行中繼資料資訊檢查。

4.4.4.語義分析（SqlNode –> RelNode）

這一步就是将 SqlNode 轉換成 RelNode，也就是生成相應的關系代數層面的邏輯（這裡一般都叫做邏輯計劃：Logical Plan）。

4.4.5.優化階段（RelNode –> RelNode）

這一步就是優化階段。詳細内容可以自己 debug 代碼檢視，此處不贅述。

4.5.calcite 怎麼做到這麼通用？

此處以 calcite parser 舉例說明，其子產品為什麼這通用？其他的子產品都是類似的方式。

先說結論

：因為 calcite parser 子產品提供了接口，具體的 parse 邏輯、規則是可以根據使用者自定義進行配置的。大家可以看下圖，部落客畫出了一張圖進行詳述。

5

如上圖，引擎 sql 解析器的生成是有一個輸入的，就是

使用者自定義文法分析規則變量

，具體引擎的 sql 解析器其實也是根據使用者自定義的

解析規則

去生成的

解析器

。其

解析器

的動态生成依賴

javacc

這樣的元件。calcite 提供的是統一的 sql AST 模型、優化模型接口等，而具體的解析實作交給了使用者自己去決定。

javacc

會根據 calcite 中定義的

Parser.jj

檔案，生成具體的 sql parser 代碼（如上圖），這個 sql parser 的能力就是将 sql 轉換成 AST （SqlNode）。關于 calcite 能力的更詳細内容見 https://matt33.com/2019/03/07/apache-calcite-process-flow/ 。

上圖涉及到的檔案大家可以下載下傳

calcite 源碼

https://github.com/apache/calcite.git 之後，切換到

core

module 之後檢視。

31

4.5.1.javacc 是啥？

javacc

是一個用 java 開發的最受歡迎的文法分析生成器。這個分析生成器工具可以讀取上下文無關且有着特殊意義的文法并把它轉換成可以識别且比對該文法的 java 程式。它是 100% 的純 java 代碼，可以在多種平台上運作。

簡單解釋

javacc

就是它是一個通用的文法分析生産器，使用者可以使用

javacc

任意定義一套 DSL 及解析器。

舉個例子，如果哪天你覺得 sql 也不夠簡潔通用，你可以使用

javacc

自己定義一套更簡潔的

user-define-ql

。然後使用 javacc 作為你的

user-define-ql

的解析器。是不是很流批，可以自己去搞編譯器了。

4.5.2.跑跑 javacc

這裡不介紹具體的 javacc 文法，直接以官網的

Simple1.jj

為案例。詳細文法和功能可以參考官網（https://javacc.github.io/javacc/） 或者一下部落格。

1. https://www.cnblogs.com/Gavin_Liu/archive/2009/03/07/1405029.html
2. https://www.yangguo.info/2014/12/13/%E7%BC%96%E8%AF%91%E5%8E%9F%E7%90%86-Javacc%E4%BD%BF%E7%94%A8/
3. https://www.engr.mun.ca/~theo/JavaCC-Tutorial/javacc-tutorial.pdf

Simple1.jj

是用于識别一系列的

{相同數量的花括号}

，之後跟着 0 個或多個行終結符。

7

下面是合法的字元串例子：

{}

，

{{{{{}}}}}

，etc.

下面是不合法的字元串例子：

{{{{

，

{}{}

，

{}}

，

{{}{}}

，etc.

接下來讓我們實際将

Simple1.jj

編譯生成具體的規則代碼。

在 pom 中加入 javacc build 插件：

<plugin>
    <!-- This must be run AFTER the fmpp-maven-plugin -->
    <groupId>org.codehaus.mojo</groupId>
    <artifactId>javacc-maven-plugin</artifactId>
    <version>2.4</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <id>javacc</id>
            <goals>
                <goal>javacc</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.build.directory}/generated-sources/</sourceDirectory>
                <includes>
                    <include>**/Simple1.jj</include>
                </includes>
                <!-- This must be kept synced with Apache Calcite. -->
                <lookAhead>1</lookAhead>
                <isStatic>false</isStatic>
                <outputDirectory>${project.build.directory}/generated-sources/</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>
           

複制

在 compile 之後，就會在

generated-sources

下生成代碼：

8

然後把代碼 copy 到 Sources 路徑下：

33

執行下代碼，可以看到

{}

，

{{}}

都可以校驗通過，一旦出現不符合規則的

{{

輸入，就會抛出異常。

關于 javacc 基本上就了解個大概了。

感興趣的可以嘗試自定義一個編譯器。

4.5.3.fmpp 是啥？

5

fmpp 就是一個基于 freemarker 的模闆生産器。使用者可以統一管理自己的變量，然後用

ftl 模闆 + 變量

生成對應的最終檔案。在 calcite 中使用 fmpp 作為變量 + 模闆的統一管理器。然後基于 fmpp 來生成對應的

Parser.jj

檔案。

5.原理剖析篇-calcite 在 flink sql 中大展身手

部落客畫了一張圖，包含了其中重要元件之間的依賴關系。

3

你沒猜錯，還是上面那些流程，

fmpp（Parser.jj 模闆生成）

->

javacc（Parser 生成）

->

calcite

。

在介紹 Parser 生成流程之前，先看看 flink 最終生成的 Parser：

FlinkSqlParserImpl

（此處使用 Blink Planner）。

5.1.FlinkSqlParserImpl

以下面這個案例出發（代碼基于 flink 1.13.1 版本）：

public class ParserTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(10);

        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        DataStream<Tuple3<String, Long, Long>> tuple3DataStream =
                env.fromCollection(Arrays.asList(
                        Tuple3.of("2", 1L, 1627254000000L),
                        Tuple3.of("2", 1L, 1627218000000L + 5000L),
                        Tuple3.of("2", 101L, 1627218000000L + 6000L),
                        Tuple3.of("2", 201L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 86400000 + 7000L)))
                        .assignTimestampsAndWatermarks(
                                new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Long>>(Time.seconds(0L)) {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, Long, Long> element) {
                                        return element.f2;
                                    }
                                });

        tEnv.registerFunction("mod", new Mod_UDF());

        tEnv.registerFunction("status_mapper", new StatusMapper_UDF());

        tEnv.createTemporaryView("source_db.source_table", tuple3DataStream,
                "status, id, timestamp, rowtime.rowtime");

        String sql = "SELECT\n"
                + "  count(1),\n"
                + "  cast(tumble_start(rowtime, INTERVAL '1' DAY) as string)\n"
                + "FROM\n"
                + "  source_db.source_table\n"
                + "GROUP BY\n"
                + "  tumble(rowtime, INTERVAL '1' DAY)";

        Table result = tEnv.sqlQuery(sql);

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

        env.execute();

    }

}
           

複制

debug 過程如之前分析 sql -> SqlNode 過程所示，如下圖直接定位到 SqlParser：

21

如上圖可以看到具體的 Parser 就是

FlinkSqlParserImpl

。

定位到具體的代碼如下圖所示（

flink-table-palnner-blink-2.11-1.13.1.jar

）。

34

最終 parse 的結果 SqlNode 如下圖。

22

再來看看

FlinkSqlParserImpl

是怎麼使用 calcite 生成的。

具體到 flink 中的實作，位于源碼中的

flink-table

.

flink-sql-parser

子產品（源碼基于 flink 1.13.1）。

flink 是依賴 maven 插件實作的上面的整體流程。

5.2.FlinkSqlParserImpl 的生成

14

接下來看看整個 Parser 生成流程。

5.2.1.flink 引入 calcite

使用

maven-dependency-plugin

将 calcite 解壓到 flink 項目 build 目錄下。

<plugin>
    <!-- Extract parser grammar template from calcite-core.jar and put
         it under ${project.build.directory} where all freemarker templates are. -->
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-dependency-plugin</artifactId>
    <executions>
        <execution>
            <id>unpack-parser-template</id>
            <phase>initialize</phase>
            <goals>
                <goal>unpack</goal>
            </goals>
            <configuration>
                <artifactItems>
                    <artifactItem>
                        <groupId>org.apache.calcite</groupId>
                        <artifactId>calcite-core</artifactId>
                        <type>jar</type>
                        <overWrite>true</overWrite>
                        <outputDirectory>${project.build.directory}/</outputDirectory>
                        <includes>**/Parser.jj</includes>
                    </artifactItem>
                </artifactItems>
            </configuration>
        </execution>
    </executions>
</plugin>
           

複制

15

5.2.2.fmpp 生成 Parser.jj

使用

maven-resources-plugin

将

Parser.jj

代碼生成。

<plugin>
    <artifactId>maven-resources-plugin</artifactId>
    <executions>
        <execution>
            <id>copy-fmpp-resources</id>
            <phase>initialize</phase>
            <goals>
                <goal>copy-resources</goal>
            </goals>
            <configuration>
                <outputDirectory>${project.build.directory}/codegen</outputDirectory>
                <resources>
                    <resource>
                        <directory>src/main/codegen</directory>
                        <filtering>false</filtering>
                    </resource>
                </resources>
            </configuration>
        </execution>
    </executions>
</plugin>
<plugin>
    <groupId>com.googlecode.fmpp-maven-plugin</groupId>
    <artifactId>fmpp-maven-plugin</artifactId>
    <version>1.0</version>
    <dependencies>
        <dependency>
            <groupId>org.freemarker</groupId>
            <artifactId>freemarker</artifactId>
            <version>2.3.28</version>
        </dependency>
    </dependencies>
    <executions>
        <execution>
            <id>generate-fmpp-sources</id>
            <phase>generate-sources</phase>
            <goals>
                <goal>generate</goal>
            </goals>
            <configuration>
                <cfgFile>${project.build.directory}/codegen/config.fmpp</cfgFile>
                <outputDirectory>target/generated-sources</outputDirectory>
                <templateDirectory>${project.build.directory}/codegen/templates</templateDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>
           

複制

16

5.2.3.javacc 生成 parser

使用 javacc 将根據

Parser.jj

檔案生成 Parser。

<plugin>
    <!-- This must be run AFTER the fmpp-maven-plugin -->
    <groupId>org.codehaus.mojo</groupId>
    <artifactId>javacc-maven-plugin</artifactId>
    <version>2.4</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <id>javacc</id>
            <goals>
                <goal>javacc</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.build.directory}/generated-sources/</sourceDirectory>
                <includes>
                    <include>**/Parser.jj</include>
                </includes>
                <!-- This must be kept synced with Apache Calcite. -->
                <lookAhead>1</lookAhead>
                <isStatic>false</isStatic>
                <outputDirectory>${project.build.directory}/generated-sources/</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>
           

複制

17

5.2.4.看看 Parser

最終生成的

Parser

就是

FlinkSqlParserImpl

。

18

5.2.5.blink planner 引入 flink-sql-parser

blink planner（

flink-table-planner-blink

） 在打包時将

flink-sql-parser

、

flink-sql-parser-hive

打包進去。

35

6.總結與展望篇

本文主要介紹了 flink sql 與 calcite 之間的依賴關系，以及 flink sql parser 的生成過程。

7.參考文獻

https://www.slideshare.net/JordanHalterman/introduction-to-apache-calcite

https://arxiv.org/pdf/1802.10233.pdf

https://changbo.tech/blog/7dec2e4.html

http://www.liaojiayi.com/calcite/

https://www.zhihu.com/column/c_1110245426124554240

https://blog.csdn.net/QuinnNorris/article/details/70739094

https://www.pianshen.com/article/72171186489/

https://matt33.com/2019/03/07/apache-calcite-process-flow/

https://www.jianshu.com/p/edf503a2a1e7

https://blog.csdn.net/u013007900/article/details/78978271

https://blog.csdn.net/u013007900/article/details/78993101

http://www.ptbird.cn/optimization-of-relational-algebraic-expression.html

https://book.51cto.com/art/201306/400084.htm

https://book.51cto.com/art/201306/400085.htm

https://miaowenting.site/2019/11/10/Flink-SQL-with-Calcite/