Table API

Table API是用于流和批处理的统一关系API。 Table API查询可以在批量或流式输入上运行而无需修改。 Table API是SQL语言的超级集合，专门用于与Apache Flink一起使用。 Table API是Scala和Java语言集成API。 Table API查询不是像SQL一样将字符串值指定为SQL，而是在Java或Scala中以嵌入语言的样式定义，并支持自动完成和语法验证等IDE支持。

Table API与Flink的SQL集成共享其API的许多概念和部分。查看Common Concepts＆API以了解如何注册表或创建Table对象。该流概念页讨论流如动态表和时间属性，具体的概念。

以下示例假定Orders使用属性调用的已注册表(a, b, c, rowtime)。该rowtime字段是流式传输中的逻辑时间属性或批处理中的常规时间戳字段。

概述和示例

Table API可用于Scala和Java。Scala Table API利用Scala表达式，Java Table API基于字符串，这些字符串被解析并转换为等效表达式。

以下示例显示了Scala和Java Table API之间的差异。表程序在批处理环境中执行。它Orders按字段Scan表，a并按组计算结果行。表程序的结果转换为DataSet类型Row并打印。

• Java
• Scala

通过导入启用Java Table API org.apache.flink.table.api.java.*。以下示例显示如何构造Java Table API程序以及如何将表达式指定为字符串。

// environment configuration
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
// register Orders table in table environment
// ...
// specify table program
Table orders = tEnv.scan("Orders"); // schema (a, b, c, rowtime)
Table counts = orders
        .groupBy("a")
        .select("a, b.count as cnt");
// conversion to DataSet
DataSet<Row> result = tableEnv.toDataSet(counts, Row.class);
result.print();

The Scala Table API is enabled by importing org.apache.flink.api.scala. and org.apache.flink.table.api.scala..

The following example shows how a Scala Table API program is constructed. Table attributes are referenced using Scala Symbols, which start with an apostrophe character (').

import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala._
// environment configuration
val env = ExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env)
// register Orders table in table environment
// ...
// specify table program
val orders = tEnv.scan("Orders") // schema (a, b, c, rowtime)
val result = orders
               .groupBy('a)
               .select('a, 'b.count as 'cnt)
               .toDataSet[Row] // conversion to DataSet
               .print()

下一个示例显示了一个更复杂的 Table API程序。程序再次扫描Orders表格。它过滤空值，规范化aString类型的字段，并计算每小时和产品a的平均计费金额b。

• Java
• Scala

// environment configuration
// ...
// specify table program
Table orders = tEnv.scan("Orders"); // schema (a, b, c, rowtime)
Table result = orders
        .filter("a.isNotNull && b.isNotNull && c.isNotNull")
        .select("a.lowerCase(), b, rowtime")
        .window(Tumble.over("1.hour").on("rowtime").as("hourlyWindow"))
        .groupBy("hourlyWindow, a")
        .select("a, hourlyWindow.end as hour, b.avg as avgBillingAmount");

// environment configuration
// ...
// specify table program
val orders: Table = tEnv.scan("Orders") // schema (a, b, c, rowtime)
val result: Table = orders
        .filter('a.isNotNull && 'b.isNotNull && 'c.isNotNull)
        .select('a.lowerCase(), 'b, 'rowtime)
        .window(Tumble over 1.hour on 'rowtime as 'hourlyWindow)
        .groupBy('hourlyWindow, 'a)
        .select('a, 'hourlyWindow.end as 'hour, 'b.avg as 'avgBillingAmount)

由于 Table API是批处理和流数据的统一API，因此两个示例程序都可以在批处理和流输入上执行，而无需对表程序本身进行任何修改。在这两种情况下，程序产生相同的结果，因为流记录不会延迟（有关详细信息，请参阅流式概念）。

算子操作

Table API支持以下 算子操作。请注意，并非所有 算子操作都可用于批处理和流式传输; 他们被相应地标记。

Scan，Projection和过滤

• Java
• Scala

Table orders = tableEnv.scan("Orders");

Table orders = tableEnv.scan("Orders");Table result = orders.select("a, c as d");

Table result = orders.select("");

Table orders = tableEnv.scan("Orders");Table result = orders.as("x, y, z, t");

Table orders = tableEnv.scan("Orders");Table result = orders.where("b === 'red'");

Table orders = tableEnv.scan("Orders");Table result = orders.filter("a % 2 === 0");

val orders: Table = tableEnv.scan("Orders")

val orders: Table = tableEnv.scan("Orders")val result = orders.select('a, 'c as 'd)

val orders: Table = tableEnv.scan("Orders")val result = orders.select(')

val orders: Table = tableEnv.scan("Orders").as('x, 'y, 'z, 't)

val orders: Table = tableEnv.scan("Orders")val result = orders.filter('a % 2 === 0)

val orders: Table = tableEnv.scan("Orders")val result = orders.where('b === "red")

聚合

• Java
• Scala

Table orders = tableEnv.scan("Orders");Table result = orders.groupBy("a").select("a, b.sum as d");

Table orders = tableEnv.scan("Orders");Table result = orders    .window(Tumble.over("5.minutes").on("rowtime").as("w")) // define window    .groupBy("a, w") // group by key and window    .select("a, w.start, w.end, w.rowtime, b.sum as d"); // access window properties and aggregate

Table orders = tableEnv.scan("Orders");Table result = orders    // define window    .window(Over      .partitionBy("a")      .orderBy("rowtime")      .preceding("UNBOUNDED_RANGE")      .following("CURRENT_RANGE")      .as("w"))    .select("a, b.avg over w, b.max over w, b.min over w"); // sliding aggregate

Table orders = tableEnv.scan("Orders");Table result = orders.distinct();

val orders: Table = tableEnv.scan("Orders")val result = orders.groupBy('a).select('a, 'b.sum as 'd)

val orders: Table = tableEnv.scan("Orders")val result: Table = orders    .window(Tumble over 5.minutes on 'rowtime as 'w) // define window    .groupBy('a, 'w) // group by key and window    .select('a, w.start, 'w.end, 'w.rowtime, 'b.sum as 'd) // access window properties and aggregate

val orders: Table = tableEnv.scan("Orders")val result: Table = orders    // define window    .window(Over      partitionBy 'a      orderBy 'rowtime      preceding UNBOUNDED_RANGE      following CURRENT_RANGE      as 'w)    .select('a, 'b.avg over 'w, 'b.max over 'w, 'b.min over 'w) // sliding aggregate

val orders: Table = tableEnv.scan("Orders")val result = orders.distinct()

Join

• Java
• Scala

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "d, e, f");Table result = left.join(right).where("a = d").select("a, b, e");

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "d, e, f");Table leftOuterResult = left.leftOuterJoin(right, "a = d").select("a, b, e");Table rightOuterResult = left.rightOuterJoin(right, "a = d").select("a, b, e");Table fullOuterResult = left.fullOuterJoin(right, "a = d").select("a, b, e");

Table left = tableEnv.fromDataSet(ds1, "a, b, c, ltime.rowtime");Table right = tableEnv.fromDataSet(ds2, "d, e, f, rtime.rowtime");Table result = left.join(right)  .where("a = d && ltime >= rtime - 5.minutes && ltime < rtime + 10.minutes")  .select("a, b, e, ltime");

// register functionTableFunction<String> split = new MySplitUDTF();tEnv.registerFunction("split", split);// joinTable orders = tableEnv.scan("Orders");Table result = orders    .join(new Table(tEnv, "split(c)").as("s", "t", "v"))    .select("a, b, s, t, v");

// register functionTableFunction<String> split = new MySplitUDTF();tEnv.registerFunction("split", split);// joinTable orders = tableEnv.scan("Orders");Table result = orders    .leftOuterJoin(new Table(tEnv, "split(c)").as("s", "t", "v"))    .select("a, b, s, t, v");

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'd, 'e, 'f)val result = left.join(right).where('a === 'd).select('a, 'b, 'e)

val left = tableEnv.fromDataSet(ds1, 'a, 'b, 'c)val right = tableEnv.fromDataSet(ds2, 'd, 'e, 'f)val leftOuterResult = left.leftOuterJoin(right, 'a === 'd).select('a, 'b, 'e)val rightOuterResult = left.rightOuterJoin(right, 'a === 'd).select('a, 'b, 'e)val fullOuterResult = left.fullOuterJoin(right, 'a === 'd).select('a, 'b, 'e)

val left = ds1.toTable(tableEnv, 'a, 'b, 'c, 'ltime.rowtime)val right = ds2.toTable(tableEnv, 'd, 'e, 'f, 'rtime.rowtime)val result = left.join(right)  .where('a === 'd && 'ltime >= 'rtime - 5.minutes && 'ltime < 'rtime + 10.minutes)  .select('a, 'b, 'e, 'ltime)

// instantiate functionval split: TableFunction[] = new MySplitUDTF()// joinval result: Table = table    .join(split('c) as ('s, 't, 'v))    .select('a, 'b, 's, 't, 'v)

// instantiate functionval split: TableFunction[] = new MySplitUDTF()// joinval result: Table = table    .leftOuterJoin(split('c) as ('s, 't, 'v))    .select('a, 'b, 's, 't, 'v)

设置 算子操作

• Java
• Scala

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "a, b, c");Table result = left.union(right);

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "a, b, c");Table result = left.unionAll(right);

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "d, e, f");Table result = left.intersect(right);

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "d, e, f");Table result = left.intersectAll(right);

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "a, b, c");Table result = left.minus(right);

Table left = tableEnv.fromDataSet(ds1, "a, b, c");Table right = tableEnv.fromDataSet(ds2, "a, b, c");Table result = left.minusAll(right);

Table left = ds1.toTable(tableEnv, "a, b, c");Table right = ds2.toTable(tableEnv, "a");// using implicit registrationTable result = left.select("a, b, c").where("a.in(" + right + ")");// using explicit registrationtableEnv.registerTable("RightTable", right);Table result = left.select("a, b, c").where("a.in(RightTable)");

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'a, 'b, 'c)val result = left.union(right)

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'a, 'b, 'c)val result = left.unionAll(right)

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'e, 'f, 'g)val result = left.intersect(right)

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'e, 'f, 'g)val result = left.intersectAll(right)

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'a, 'b, 'c)val result = left.minus(right)

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'a, 'b, 'c)val result = left.minusAll(right)

val left = ds1.toTable(tableEnv, 'a, 'b, 'c)val right = ds2.toTable(tableEnv, 'a)val result = left.select('a, 'b, 'c).where('a.in(right))

OrderBy，Offset＆Fetch

• Java
• Scala

Table in = tableEnv.fromDataSet(ds, "a, b, c");Table result = in.orderBy("a.asc");

Table in = tableEnv.fromDataSet(ds, "a, b, c");// returns the first 5 records from the sorted resultTable result1 = in.orderBy("a.asc").fetch(5); // skips the first 3 records and returns all following records from the sorted resultTable result2 = in.orderBy("a.asc").offset(3);// skips the first 10 records and returns the next 5 records from the sorted resultTable result3 = in.orderBy("a.asc").offset(10).fetch(5);

val in = ds.toTable(tableEnv, 'a, 'b, 'c)val result = in.orderBy('a.asc)

val in = ds.toTable(tableEnv, 'a, 'b, 'c)// returns the first 5 records from the sorted resultval result1: Table = in.orderBy('a.asc).fetch(5)// skips the first 3 records and returns all following records from the sorted resultval result2: Table = in.orderBy('a.asc).offset(3)// skips the first 10 records and returns the next 5 records from the sorted resultval result3: Table = in.orderBy('a.asc).offset(10).fetch(5)

Insert

• Java
• Scala

Table orders = tableEnv.scan("Orders");orders.insertInto("OutOrders");

val orders: Table = tableEnv.scan("Orders")orders.insertInto("OutOrders")

GroupWindows

组窗口根据时间或行计数间隔将行组聚合为有限组，并按组评估聚合函数。对于批处理表，窗口是按时间间隔对记录进行分组的便捷快捷方式。

Windows是使用该window(w: Window)子句定义的，需要使用该as子句指定的别名。为了按窗口对表进行分组，必须在groupBy(…)子句中引用窗口别名，就像常规分组属性一样。以下示例显示如何在表上定义窗口聚合。

• Java
• Scala

Table table = input
  .window([Window w].as("w"))  // define window with alias w
  .groupBy("w")  // group the table by window w
  .select("b.sum");  // aggregate

val table = input
  .window([w: Window] as 'w)  // define window with alias w
  .groupBy('w)   // group the table by window w
  .select('b.sum)  // aggregate

在流式传输环境中，如果窗口聚合除了窗口之外还在一个或多个属性上进行分组，则它们只能并行计算，即groupBy(…)子句引用窗口别名和至少一个附加属性。一个groupBy(…)仅引用一个窗口别名（如在上面的例子）子句只能由一个单一的，非平行的任务进行评估。以下示例显示如何使用其他分组属性定义窗口聚合。

• Java
• Scala

Table table = input
  .window([Window w].as("w"))  // define window with alias w
  .groupBy("w, a")  // group the table by attribute a and window w 
  .select("a, b.sum");  // aggregate

val table = input
  .window([w: Window] as 'w) // define window with alias w
  .groupBy('w, 'a)  // group the table by attribute a and window w
  .select('a, 'b.sum)  // aggregate

窗口属性诸如开始，结束，或一个时间窗口的rowtime时间戳可以在选择语句被添加为窗口别名作为的一个属性w.start，w.end以及w.rowtime分别。窗口开始和行时间戳是包含的低窗口和超窗口边界。相反，窗口结束时间戳是独占的上窗口边界。例如，从下午2点开始的30分钟的翻滚窗口将具有14:00:00.000开始时间戳，14:29:59.999行时间戳和14:30:00.000结束时间戳。

• Java
• Scala

Table table = input
  .window([Window w].as("w"))  // define window with alias w
  .groupBy("w, a")  // group the table by attribute a and window w 
  .select("a, w.start, w.end, w.rowtime, b.count"); // aggregate and add window start, end, and rowtime timestamps

val table = input
  .window([w: Window] as 'w)  // define window with alias w
  .groupBy('w, 'a)  // group the table by attribute a and window w
  .select('a, 'w.start, 'w.end, 'w.rowtime, 'b.count) // aggregate and add window start, end, and rowtime timestamps

该Window参数定义行如何映射到窗口。Window不是用户可以实现的接口。相反， Table API提供了一组Window具有特定语义的预定义类，这些类被转换为底层DataStream或DataSet 算子操作。支持的窗口定义如下所示。

翻滚（翻滚的Windows）

翻滚窗口将行分配给固定长度的非重叠连续窗口。例如，5分钟的翻滚窗口以5分钟为间隔对行进行分组。可以在事件时间，处理时间或行数上定义翻滚窗口。

使用Tumble类定义翻滚窗口如下：

• Java
• Scala

// Tumbling Event-time Window
.window(Tumble.over("10.minutes").on("rowtime").as("w"));
// Tumbling Processing-time Window (assuming a processing-time attribute "proctime")
.window(Tumble.over("10.minutes").on("proctime").as("w"));
// Tumbling Row-count Window (assuming a processing-time attribute "proctime")
.window(Tumble.over("10.rows").on("proctime").as("w"));

// Tumbling Event-time Window
.window(Tumble over 10.minutes on 'rowtime as 'w)
// Tumbling Processing-time Window (assuming a processing-time attribute "proctime")
.window(Tumble over 10.minutes on 'proctime as 'w)
// Tumbling Row-count Window (assuming a processing-time attribute "proctime")
.window(Tumble over 10.rows on 'proctime as 'w)

滑动（滑动窗口）

滑动窗口具有固定大小，并按指定的滑动间隔滑动。如果滑动间隔小于窗口大小，则滑动窗口重叠。因此，可以将行分配给多个窗口。例如，15分钟大小和5分钟滑动间隔的滑动窗口将每行分配给3个不同的15分钟大小的窗口，这些窗口以5分钟的间隔进行评估。可以在事件时间，处理时间或行数上定义滑动窗口。

滑动窗口使用Slide如下类定义：

• Java
• Scala

// Sliding Event-time Window
.window(Slide.over("10.minutes").every("5.minutes").on("rowtime").as("w"));
// Sliding Processing-time window (assuming a processing-time attribute "proctime")
.window(Slide.over("10.minutes").every("5.minutes").on("proctime").as("w"));
// Sliding Row-count window (assuming a processing-time attribute "proctime")
.window(Slide.over("10.rows").every("5.rows").on("proctime").as("w"));

// Sliding Event-time Window
.window(Slide over 10.minutes every 5.minutes on 'rowtime as 'w)
// Sliding Processing-time window (assuming a processing-time attribute "proctime")
.window(Slide over 10.minutes every 5.minutes on 'proctime as 'w)
// Sliding Row-count window (assuming a processing-time attribute "proctime")
.window(Slide over 10.rows every 5.rows on 'proctime as 'w)

会话（会话窗口）

会话窗口没有固定的大小，但它们的边界由不活动的间隔定义，即如果在定义的间隙期间没有出现事件，则会话窗口关闭。例如，如果在30分钟不活动后观察到一行，则会开始一个30分钟间隙的会话窗口（否则该行将被添加到现有窗口中），如果在30分钟内未添加任何行，则会关闭。会话窗口可以在事件时间或处理时间上工作。

使用Session类定义会话窗口，如下所示：

• Java
• Scala

// Session Event-time Window
.window(Session.withGap("10.minutes").on("rowtime").as("w"));
// Session Processing-time Window (assuming a processing-time attribute "proctime")
.window(Session.withGap("10.minutes").on("proctime").as("w"));

// Session Event-time Window
.window(Session withGap 10.minutes on 'rowtime as 'w)
// Session Processing-time Window (assuming a processing-time attribute "proctime")
.window(Session withGap 10.minutes on 'proctime as 'w)

OverWindows

从标准SQL（OVER子句）已知过窗口聚合，并在SELECT查询的子句中定义。与GROUP BY子句中指定的组窗口不同，在窗口上不会折叠行。而是通过窗口聚合计算每个输入行在其相邻行的范围内的聚合。

OverWindows是使用window(w: OverWindow*)子句定义的，并通过select()方法中的别名引用。以下示例显示如何在表上定义过窗口聚合。

• Java
• Scala

Table table = input
  .window([OverWindow w].as("w"))           // define over window with alias w
  .select("a, b.sum over w, c.min over w"); // aggregate over the over window w

val table = input
  .window([w: OverWindow] as 'w)              // define over window with alias w
  .select('a, 'b.sum over 'w, 'c.min over 'w) // aggregate over the over window w

的OverWindow限定范围在其聚集计算的行。OverWindow不是用户可以实现的接口。相反， Table API提供了Over用于配置覆盖窗口属性的类。可以在事件时间或处理时间以及指定为时间间隔或行计数的范围上定义窗口。支持的窗口定义在Over（和其他类）上公开为方法，如下所示：

注意：目前，同一select()调用中的所有聚合函数必须计算相同的窗口。

在Windows上无限制

• Java
• Scala

// Unbounded Event-time over window (assuming an event-time attribute "rowtime")
.window(Over.partitionBy("a").orderBy("rowtime").preceding("unbounded_range").as("w"));
// Unbounded Processing-time over window (assuming a processing-time attribute "proctime")
.window(Over.partitionBy("a").orderBy("proctime").preceding("unbounded_range").as("w"));
// Unbounded Event-time Row-count over window (assuming an event-time attribute "rowtime")
.window(Over.partitionBy("a").orderBy("rowtime").preceding("unbounded_row").as("w"));
 
// Unbounded Processing-time Row-count over window (assuming a processing-time attribute "proctime")
.window(Over.partitionBy("a").orderBy("proctime").preceding("unbounded_row").as("w"));

// Unbounded Event-time over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_RANGE as 'w)
// Unbounded Processing-time over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_RANGE as 'w)
// Unbounded Event-time Row-count over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy 'a orderBy 'rowtime preceding UNBOUNDED_ROW as 'w)
 
// Unbounded Processing-time Row-count over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy 'a orderBy 'proctime preceding UNBOUNDED_ROW as 'w)

限制在Windows上

• Java
• Scala

// Bounded Event-time over window (assuming an event-time attribute "rowtime")
.window(Over.partitionBy("a").orderBy("rowtime").preceding("1.minutes").as("w"))
// Bounded Processing-time over window (assuming a processing-time attribute "proctime")
.window(Over.partitionBy("a").orderBy("proctime").preceding("1.minutes").as("w"))
// Bounded Event-time Row-count over window (assuming an event-time attribute "rowtime")
.window(Over.partitionBy("a").orderBy("rowtime").preceding("10.rows").as("w"))
 
// Bounded Processing-time Row-count over window (assuming a processing-time attribute "proctime")
.window(Over.partitionBy("a").orderBy("proctime").preceding("10.rows").as("w"))

// Bounded Event-time over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy 'a orderBy 'rowtime preceding 1.minutes as 'w)
// Bounded Processing-time over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy 'a orderBy 'proctime preceding 1.minutes as 'w)
// Bounded Event-time Row-count over window (assuming an event-time attribute "rowtime")
.window(Over partitionBy 'a orderBy 'rowtime preceding 10.rows as 'w)
  
// Bounded Processing-time Row-count over window (assuming a processing-time attribute "proctime")
.window(Over partitionBy 'a orderBy 'proctime preceding 10.rows as 'w)

数据类型

Table API建立在Flink的DataSet和DataStream API之上。在内部，它还使用Flink TypeInformation来定义数据类型。完全支持的类型列在org.apache.flink.table.api.Types。下表总结了 Table API类型，SQL类型和生成的Java类之间的关系。

通用类型和（嵌套）复合类型（例如，POJO，元组，Row，Scala案例类）也可以是行的字段。

可以使用值访问函数访问具有任意嵌套的复合类型的字段。

通用类型被视为黑盒子，可以由用户定义的函数传递或处理。

表达式语法

前面部分中的一些 算子需要一个或多个表达式。可以使用嵌入式Scala DSL或字符串指定表达式。请参阅上面的示例以了解如何指定表达式。

这是表达式的EBNF语法：

expressionList = expression , { "," , expression } ;
expression = timeIndicator | overConstant | alias ;
alias = logic | ( logic , "as" , fieldReference ) | ( logic , "as" , "(" , fieldReference , { "," , fieldReference } , ")" ) ;
logic = comparison , [ ( "&&" | "||" ) , comparison ] ;
comparison = term , [ ( "=" | "==" | "===" | "!=" | "!==" | ">" | ">=" | "<" | "<=" ) , term ] ;
term = product , [ ( "+" | "-" ) , product ] ;
product = unary , [ ( "*" | "/" | "%") , unary ] ;
unary = [ "!" | "-" ] , composite ;
composite = over | nullLiteral | suffixed | atom ;
suffixed = interval | cast | as | if | functionCall ;
interval = timeInterval | rowInterval ;
timeInterval = composite , "." , ("year" | "years" | "quarter" | "quarters" | "month" | "months" | "week" | "weeks" | "day" | "days" | "hour" | "hours" | "minute" | "minutes" | "second" | "seconds" | "milli" | "millis") ;
rowInterval = composite , "." , "rows" ;
cast = composite , ".cast(" , dataType , ")" ;
dataType = "BYTE" | "SHORT" | "INT" | "LONG" | "FLOAT" | "DOUBLE" | "BOOLEAN" | "STRING" | "DECIMAL" | "SQL_DATE" | "SQL_TIME" | "SQL_TIMESTAMP" | "INTERVAL_MONTHS" | "INTERVAL_MILLIS" | ( "MAP" , "(" , dataType , "," , dataType , ")" ) | ( "PRIMITIVE_ARRAY" , "(" , dataType , ")" ) | ( "OBJECT_ARRAY" , "(" , dataType , ")" ) ;
as = composite , ".as(" , fieldReference , ")" ;
if = composite , ".?(" , expression , "," , expression , ")" ;
functionCall = composite , "." , functionIdentifier , [ "(" , [ expression , { "," , expression } ] , ")" ] ;
atom = ( "(" , expression , ")" ) | literal | fieldReference ;
fieldReference = "*" | identifier ;
nullLiteral = "Null(" , dataType , ")" ;
timeIntervalUnit = "YEAR" | "YEAR_TO_MONTH" | "MONTH" | "QUARTER" | "WEEK" | "DAY" | "DAY_TO_HOUR" | "DAY_TO_MINUTE" | "DAY_TO_SECOND" | "HOUR" | "HOUR_TO_MINUTE" | "HOUR_TO_SECOND" | "MINUTE" | "MINUTE_TO_SECOND" | "SECOND" ;
timePointUnit = "YEAR" | "MONTH" | "DAY" | "HOUR" | "MINUTE" | "SECOND" | "QUARTER" | "WEEK" | "MILLISECOND" | "MICROSECOND" ;
over = composite , "over" , fieldReference ;
overConstant = "current_row" | "current_range" | "unbounded_row" | "unbounded_row" ;
timeIndicator = fieldReference , "." , ( "proctime" | "rowtime" ) ;

这里literal是一个有效的Java文字，fieldReference指定数据中的列（或使用的所有列*），并functionIdentifier指定支持的标量函数。列名和函数名遵循Java标识符语法。指定为字符串的表达式也可以使用前缀表示法而不是后缀表示法来调用 算子和函数。

如果需要使用精确数值或大小数， Table API也支持JavaBigDecimal类型。在Scala Table API中，小数可以BigDecimal("123456")通过附加“p” 来定义，也可以在Java中定义，例如123456p。

为了使用时态值， Table API支持Java SQL的Date，Time和Timestamp类型。在Scala的 Table API文本可以通过使用来定义java.sql.Date.valueOf("2016-06-27")，java.sql.Time.valueOf("10:10:42")或java.sql.Timestamp.valueOf("2016-06-27 10:10:42.123")。在Java和Scala Table API还支持调用"2016-06-27".toDate()，"10:10:42".toTime()以及"2016-06-27 10:10:42.123".toTimestamp()对String转化成时间类型。注意：由于Java临时SQL类型与时区有关，因此请确保Flink Client和所有TaskManagers使用相同的时区。

时间间隔可以表示为月数（Types.INTERVAL_MONTHS）或毫秒数（Types.INTERVAL_MILLIS）。可以添加或减去相同类型的间隔（例如1.hour + 10.minutes）。可以将毫秒的间隔添加到时间点（例如"2016-08-10".toDate + 5.days）。