Operations

Similar to the FROM clause in a SQL query. Performs a scan of a registered table.

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

Similar to the VALUES clause in a SQL query. Produces an inline table out of the provided rows.

You can use a `row(...)` expression to create composite rows:

Table table = tEnv.fromValues(
   row(1, "ABC"),
   row(2L, "ABCDE")
);

will produce a Table with a schema as follows:

root
|-- f0: BIGINT NOT NULL     // original types INT and BIGINT are generalized to BIGINT
|-- f1: VARCHAR(5) NOT NULL // original types CHAR(3) and CHAR(5) are generalized
                            // to VARCHAR(5). VARCHAR is used instead of CHAR so that
                            // no padding is applied

The method will derive the types automatically from the input expressions. If types at a certain position differ, the method will try to find a common super type for all types. If a common super type does not exist, an exception will be thrown.

You can also specify the requested type explicitly. It might be helpful for assigning more generic types like e.g. DECIMAL or naming the columns.

Table table = tEnv.fromValues(
    DataTypes.ROW(
        DataTypes.FIELD("id", DataTypes.DECIMAL(10, 2)),
        DataTypes.FIELD("name", DataTypes.STRING())
    ),
    row(1, "ABC"),
    row(2L, "ABCDE")
);

will produce a Table with a schema as follows:

root
|-- id: DECIMAL(10, 2)
|-- name: STRING

Similar to a SQL SELECT statement. Performs a select operation.

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

You can use star (*) to act as a wild card, selecting all of the columns in the table.

Table result = orders.select($("*"));

Renames fields.

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

Similar to a SQL WHERE clause. Filters out rows that do not pass the filter predicate.

Table orders = tableEnv.from("Orders");
Table result = orders.where($("b").isEqual("red"));

Table orders = tableEnv.from("Orders");
Table result = orders.filter($("a").mod(2).isEqual(0));

Similar to the FROM clause in a SQL query. Performs a scan of a registered table.

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

Similar to the VALUES clause in a SQL query. Produces an inline table out of the provided rows.

You can use a `row(...)` expression to create composite rows:

val table = tEnv.fromValues(
   row(1, "ABC"),
   row(2L, "ABCDE")
)

will produce a Table with a schema as follows:

root
|-- f0: BIGINT NOT NULL     // original types INT and BIGINT are generalized to BIGINT
|-- f1: VARCHAR(5) NOT NULL // original types CHAR(3) and CHAR(5) are generalized
                            // to VARCHAR(5). VARCHAR is used instead of CHAR so that
                            // no padding is applied

The method will derive the types automatically from the input expressions. If types at a certain position differ, the method will try to find a common super type for all types. If a common super type does not exist, an exception will be thrown.

You can also specify the requested type explicitly. It might be helpful for assigning more generic types like e.g. DECIMAL or naming the columns.

val table = tEnv.fromValues(
    DataTypes.ROW(
        DataTypes.FIELD("id", DataTypes.DECIMAL(10, 2)),
        DataTypes.FIELD("name", DataTypes.STRING())
    ),
    row(1, "ABC"),
    row(2L, "ABCDE")
)

will produce a Table with a schema as follows:

root
|-- id: DECIMAL(10, 2)
|-- name: STRING

Similar to a SQL SELECT statement. Performs a select operation.

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

You can use star (*) to act as a wild card, selecting all of the columns in the table.

val orders: Table = tableEnv.from("Orders")
val result = orders.select($"*")

Renames fields.

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

Similar to a SQL WHERE clause. Filters out rows that do not pass the filter predicate.

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

val orders: Table = tableEnv.from("Orders")
val result = orders.where($"b" === "red")

类似于SQL请求中的FROM子句，将一个环境中已注册的表转换成Table对象。

orders = t_env.from_path("Orders")

Similar to the VALUES clause in a SQL query. Produces an inline table out of the provided rows.

table = t_env.from_elements([(1, 'ABC'), (2, 'ABCDE')])

will produce a Table with a schema as follows:

root
|-- _1: BIGINT
|-- _2: STRING

The method will derive the types automatically from the input data. The elements types must be acceptable atomic types or acceptable composite types. All elements must be of the same type. If the elements types are composite types, the composite types must be strictly equal, and its subtypes must also be acceptable types, e.g. if the elements are tuples, the length of the tuples must be equal, the element types of the tuples must be equal in order.

The built-in acceptable atomic element types contains:

int, long, str, unicode, bool, float, bytearray, datetime.date, datetime.time, datetime.datetime, datetime.timedelta, decimal.Decimal

The built-in acceptable composite element types contains:

list, tuple, dict, array, pyflink.table.Row

You can also specify the requested type explicitly. It might be helpful for assigning more generic types like e.g. DECIMAL or naming the columns.

table = t_env.from_elements(
    [(1, 'ABC'), (2, 'ABCDE')],
    schema=DataTypes.ROW([DataTypes.FIELD('id', DataTypes.INT()),
                          DataTypes.FIELD('name', DataTypes.STRING())]))
)

will produce a Table with a schema as follows:

root
|-- id: INT
|-- name: STRING

类似于SQL请求中的SELECT子句，执行一个select操作。

orders = t_env.from_path("Orders")
result = orders.select("a, c as d")

您可以使用星号 (*) 表示选择表中的所有列。

result = orders.select("*")

重命名字段。

orders = t_env.from_path("Orders")
result = orders.alias("x, y, z, t")

类似于SQL请求中的WHERE子句，过滤掉表中不满足条件的行。

orders = t_env.from_path("Orders")
result = orders.where("a === 'red'")

orders = t_env.from_path("Orders")
result = orders.filter("b % 2 === 0")

Performs a field add operation. It will throw an exception if the added fields already exist.

Table orders = tableEnv.from("Orders");
Table result = orders.addColumns(concat($("c"), "sunny"));

Performs a field add operation. Existing fields will be replaced if add columns name is the same as the existing column name. Moreover, if the added fields have duplicate field name, then the last one is used.

Table orders = tableEnv.from("Orders");
Table result = orders.addOrReplaceColumns(concat($("c"), "sunny").as("desc"));

Performs a field drop operation. The field expressions should be field reference expressions, and only existing fields can be dropped.

Table orders = tableEnv.from("Orders");
Table result = orders.dropColumns($("b"), $("c"));

Performs a field rename operation. The field expressions should be alias expressions, and only the existing fields can be renamed.

Table orders = tableEnv.from("Orders");
Table result = orders.renameColumns($("b").as("b2"), $("c").as("c2"));

Performs a field add operation. It will throw an exception if the added fields already exist.

val orders = tableEnv.from("Orders");
val result = orders.addColumns(concat($"c", "Sunny"))

Performs a field add operation. Existing fields will be replaced if add columns name is the same as the existing column name. Moreover, if the added fields have duplicate field name, then the last one is used.

val orders = tableEnv.from("Orders");
val result = orders.addOrReplaceColumns(concat($"c", "Sunny") as "desc")

Performs a field drop operation. The field expressions should be field reference expressions, and only existing fields can be dropped.

val orders = tableEnv.from("Orders");
val result = orders.dropColumns($"b", $"c")

Performs a field rename operation. The field expressions should be alias expressions, and only the existing fields can be renamed.

val orders = tableEnv.from("Orders");
val result = orders.renameColumns($"b" as "b2", $"c" as "c2")

执行新增字段操作。如果欲添加字段已经存在，将会抛出异常。

orders = t_env.from_path("Orders")
result = orders.add_columns("concat(c, 'sunny')")

执行新增字段操作。如果欲添加字段已经存在，将会替换该字段。如果新增字段列表中有同名字段，取最靠后的为有效字段。

orders = t_env.from_path("Orders")
result = orders.add_or_replace_columns("concat(c, 'sunny') as desc")

执行删除字段操作。参数必须是字段列表，并且必须是已经存在的字段才能被删除。

orders = t_env.from_path("Orders")
result = orders.drop_columns("b, c")

执行重命名字段操作。参数必须是字段别名(例：b as b2)列表，并且必须是已经存在的字段才能被重命名。

orders = t_env.from_path("Orders")
result = orders.rename_columns("b as b2, c as c2")

Similar to a SQL GROUP BY clause. Groups the rows on the grouping keys with a following running aggregation operator to aggregate rows group-wise.

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

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the type of aggregation and the number of distinct grouping keys. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Groups and aggregates a table on a group window and possibly one or more grouping keys.

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

Similar to a SQL OVER clause. Over window aggregates are computed for each row, based on a window (range) of preceding and succeeding rows. See the over windows section for more details.

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

Note: All aggregates must be defined over the same window, i.e., same partitioning, sorting, and range. Currently, only windows with PRECEDING (UNBOUNDED and bounded) to CURRENT ROW range are supported. Ranges with FOLLOWING are not supported yet. ORDER BY must be specified on a single time attribute.

Similar to a SQL DISTINCT aggregation clause such as COUNT(DISTINCT a). Distinct aggregation declares that an aggregation function (built-in or user-defined) is only applied on distinct input values. Distinct can be applied to GroupBy Aggregation, GroupBy Window Aggregation and Over Window Aggregation.

Table orders = tableEnv.from("Orders");
// Distinct aggregation on group by
Table groupByDistinctResult = orders
    .groupBy($("a"))
    .select($("a"), $("b").sum().distinct().as("d"));
// Distinct aggregation on time window group by
Table groupByWindowDistinctResult = orders
    .window(Tumble
            .over(lit(5).minutes())
            .on($("rowtime"))
            .as("w")
    )
    .groupBy($("a"), $("w"))
    .select($("a"), $("b").sum().distinct().as("d"));
// Distinct aggregation on over window
Table result = orders
    .window(Over
        .partitionBy($("a"))
        .orderBy($("rowtime"))
        .preceding(UNBOUNDED_RANGE)
        .as("w"))
    .select(
        $("a"), $("b").avg().distinct().over($("w")),
        $("b").max().over($("w")),
        $("b").min().over($("w"))
    );

User-defined aggregation function can also be used with DISTINCT modifiers. To calculate the aggregate results only for distinct values, simply add the distinct modifier towards the aggregation function.

Table orders = tEnv.from("Orders");

// Use distinct aggregation for user-defined aggregate functions
tEnv.registerFunction("myUdagg", new MyUdagg());
orders.groupBy("users")
    .select(
        $("users"),
        call("myUdagg", $("points")).distinct().as("myDistinctResult")
    );

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct fields. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Similar to a SQL DISTINCT clause. Returns records with distinct value combinations.

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

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct fields. Please provide a query configuration with valid retention interval to prevent excessive state size. If state cleaning is enabled, distinct have to emit messages to prevent too early state eviction of downstream operators which makes distinct contains result updating. See Query Configuration for details.

Similar to a SQL GROUP BY clause. Groups the rows on the grouping keys with a following running aggregation operator to aggregate rows group-wise.

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

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the type of aggregation and the number of distinct grouping keys. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Groups and aggregates a table on a group window and possibly one or more grouping keys.

val orders: Table = tableEnv.from("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

Similar to a SQL OVER clause. Over window aggregates are computed for each row, based on a window (range) of preceding and succeeding rows. See the over windows section for more details.

val orders: Table = tableEnv.from("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

Note: All aggregates must be defined over the same window, i.e., same partitioning, sorting, and range. Currently, only windows with PRECEDING (UNBOUNDED and bounded) to CURRENT ROW range are supported. Ranges with FOLLOWING are not supported yet. ORDER BY must be specified on a single time attribute.

Similar to a SQL DISTINCT AGGREGATION clause such as COUNT(DISTINCT a). Distinct aggregation declares that an aggregation function (built-in or user-defined) is only applied on distinct input values. Distinct can be applied to GroupBy Aggregation, GroupBy Window Aggregation and Over Window Aggregation.

val orders: Table = tableEnv.from("Orders");
// Distinct aggregation on group by
val groupByDistinctResult = orders
    .groupBy($"a")
    .select($"a", $"b".sum.distinct as "d")
// Distinct aggregation on time window group by
val groupByWindowDistinctResult = orders
    .window(Tumble over 5.minutes on $"rowtime" as "w").groupBy($"a", $"w")
    .select($"a", $"b".sum.distinct as "d")
// Distinct aggregation on over window
val result = orders
    .window(Over
        partitionBy $"a"
        orderBy $"rowtime"
        preceding UNBOUNDED_RANGE
        as $"w")
    .select($"a", $"b".avg.distinct over $"w", $"b".max over $"w", $"b".min over $"w")

User-defined aggregation function can also be used with DISTINCT modifiers. To calculate the aggregate results only for distinct values, simply add the distinct modifier towards the aggregation function.

val orders: Table = tEnv.from("Orders");

// Use distinct aggregation for user-defined aggregate functions
val myUdagg = new MyUdagg();
orders.groupBy($"users").select($"users", myUdagg.distinct($"points") as "myDistinctResult");

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct fields. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Similar to a SQL DISTINCT clause. Returns records with distinct value combinations.

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

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct fields. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

类似于SQL的GROUP BY子句。将数据按照指定字段进行分组，之后对各组内数据执行聚合操作。

orders = t_env.from_path("Orders")
result = orders.group_by("a").select("a, b.sum as d")

注意： 对于流式查询，计算查询结果所需的状态（state）可能会无限增长，具体情况取决于聚合操作的类型和分组的数量。您可能需要在查询配置中设置状态保留时间，以防止状态过大。详情请看查询配置。

在一个窗口上分组和聚合数据，可包含其它分组字段。

from pyflink.table.window import Tumble

orders = t_env.from_path("Orders")
result = orders.window(Tumble.over("5.minutes").on("rowtime").alias("w")) \ 
               .group_by("a, w") \
               .select("a, w.start, w.end, b.sum as d")

类似于SQL中的OVER开窗函数。Over窗口聚合对每一行都进行一次聚合计算，聚合的对象是以当前行的位置为基准，向前向后取一个区间范围内的所有数据。详情请见Over窗口一节。

from pyflink.table.window import Over

orders = t_env.from_path("Orders")
result = orders.over_window(Over.partition_by("a").order_by("rowtime")
                            .preceding("UNBOUNDED_RANGE").following("CURRENT_RANGE")
                            .alias("w")) \
    .select("a, b.avg over w, b.max over w, b.min over w")

注意： 所有的聚合操作必须在同一个窗口上定义，即分组，排序，范围等属性必须一致。目前，窗口区间范围的向前（PRECEDING）取值没有限制，可以为无界（UNBOUNDED），但是向后（FOLLOWING）只支持当前行（CURRENT ROW），其它向后范围取值暂不支持。排序（ORDER BY）属性必须指定单个时间属性。

类似于SQL聚合函数中的的DISTINCT关键字比如COUNT(DISTINCT a)。带有distinct标记的聚合函数只会接受不重复的输入，重复输入将被丢弃。这个去重特性可以在分组聚合（GroupBy Aggregation），分组窗口聚合（GroupBy Window Aggregation）以及Over窗口聚合（Over Window Aggregation）上使用。

orders = t_env.from_path("Orders")
# Distinct aggregation on group by
group_by_distinct_result = orders.group_by("a") \
                                 .select("a, b.sum.distinct as d")
# Distinct aggregation on time window group by
group_by_window_distinct_result = orders.window(
    Tumble.over("5.minutes").on("rowtime").alias("w")).group_by("a, w") \
    .select("a, b.sum.distinct as d")
# Distinct aggregation on over window
result = orders.over_window(Over
                       .partition_by("a")
                       .order_by("rowtime")
                       .preceding("UNBOUNDED_RANGE")
                       .alias("w")) \
                       .select(
                       "a, b.avg.distinct over w, b.max over w, b.min over w")

注意： 对于流式查询，计算查询结果所需的状态（state）可能会无限增长，具体情况取决于执行去重判断时参与判断的字段的数量。您可能需要在查询配置中设置状态保留时间，以防止状态过大。详情请看查询配置。

类似于SQL中的DISTINCT子句。返回去重后的数据。

orders = t_env.from_path("Orders")
result = orders.distinct()

注意： 对于流式查询，计算查询结果所需的状态（state）可能会无限增长，具体情况取决于执行去重判断时参与判断的字段的数量。您可能需要在查询配置中设置状态保留时间，以防止状态过大。详情请看查询配置。

Similar to a SQL JOIN clause. Joins two tables. Both tables must have distinct field names and at least one equality join predicate must be defined through join operator or using a where or filter operator.

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

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct input rows. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Similar to SQL LEFT/RIGHT/FULL OUTER JOIN clauses. Joins two tables. Both tables must have distinct field names and at least one equality join predicate must be defined.

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

Table leftOuterResult = left.leftOuterJoin(right, $("a").isEqual($("d")))
                            .select($("a"), $("b"), $("e"));
Table rightOuterResult = left.rightOuterJoin(right, $("a").isEqual($("d")))
                            .select($("a"), $("b"), $("e"));
Table fullOuterResult = left.fullOuterJoin(right, $("a").isEqual($("d")))
                            .select($("a"), $("b"), $("e"));

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct input rows. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Note: Interval joins are a subset of regular joins that can be processed in a streaming fashion.

A interval join requires at least one equi-join predicate and a join condition that bounds the time on both sides. Such a condition can be defined by two appropriate range predicates (<, <=, >=, >) or a single equality predicate that compares time attributes of the same type (i.e., processing time or event time) of both input tables.

For example, the following predicates are valid interval join conditions:

• ltime === rtime
• ltime >= rtime && ltime < rtime + 10.minutes

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(
    and(
        $("a").isEqual($("d")),
        $("ltime").isGreaterEqual($("rtime").minus(lit(5).minutes())),
        $("ltime").isLess($("rtime").plus(lit(10).minutes()))
    ))
  .select($("a"), $("b"), $("e"), $("ltime"));

Joins a table with the results of a table function. Each row of the left (outer) table is joined with all rows produced by the corresponding call of the table function. A row of the left (outer) table is dropped, if its table function call returns an empty result.

// register User-Defined Table Function
TableFunction<String> split = new MySplitUDTF();
tableEnv.registerFunction("split", split);

// join
Table orders = tableEnv.from("Orders");
Table result = orders
    .joinLateral(call("split", $("c")).as("s", "t", "v"))
    .select($("a"), $("b"), $("s"), $("t"), $("v"));

Joins a table with the results of a table function. Each row of the left (outer) table is joined with all rows produced by the corresponding call of the table function. If a table function call returns an empty result, the corresponding outer row is preserved and the result padded with null values.

Note: Currently, the predicate of a table function left outer join can only be empty or literal true.

// register User-Defined Table Function
TableFunction<String> split = new MySplitUDTF();
tableEnv.registerFunction("split", split);

// join
Table orders = tableEnv.from("Orders");
Table result = orders
    .leftOuterJoinLateral(call("split", $("c")).as("s", "t", "v"))
    .select($("a"), $("b"), $("s"), $("t"), $("v"));

Temporal tables are tables that track changes over time.

A temporal table function provides access to the state of a temporal table at a specific point in time. The syntax to join a table with a temporal table function is the same as in Inner Join with Table Function.

Currently only inner joins with temporal tables are supported.

Table ratesHistory = tableEnv.from("RatesHistory");

// register temporal table function with a time attribute and primary key
TemporalTableFunction rates = ratesHistory.createTemporalTableFunction(
    "r_proctime",
    "r_currency");
tableEnv.registerFunction("rates", rates);

// join with "Orders" based on the time attribute and key
Table orders = tableEnv.from("Orders");
Table result = orders
    .joinLateral(call("rates", $("o_proctime")), $("o_currency").isEqual($("r_currency")))

For more information please check the more detailed temporal tables concept description.

Similar to a SQL JOIN clause. Joins two tables. Both tables must have distinct field names and at least one equality join predicate must be defined through join operator or using a where or filter operator.

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")

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct input rows. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Similar to SQL LEFT/RIGHT/FULL OUTER JOIN clauses. Joins two tables. Both tables must have distinct field names and at least one equality join predicate must be defined.

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")

Note: For streaming queries the required state to compute the query result might grow infinitely depending on the number of distinct input rows. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Note: Interval joins are a subset of regular joins that can be processed in a streaming fashion.

A interval join requires at least one equi-join predicate and a join condition that bounds the time on both sides. Such a condition can be defined by two appropriate range predicates (<, <=, >=, >) or a single equality predicate that compares time attributes of the same type (i.e., processing time or event time) of both input tables.

For example, the following predicates are valid interval join conditions:

• $"ltime" === $"rtime"
• $"ltime" >= $"rtime" && $"ltime" < $"rtime" + 10.minutes

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")

Joins a table with the results of a table function. Each row of the left (outer) table is joined with all rows produced by the corresponding call of the table function. A row of the left (outer) table is dropped, if its table function call returns an empty result.

// instantiate User-Defined Table Function
val split: TableFunction[_] = new MySplitUDTF()

// join
val result: Table = table
    .joinLateral(split($"c") as ("s", "t", "v"))
    .select($"a", $"b", $"s", $"t", $"v")

Joins a table with the results of a table function. Each row of the left (outer) table is joined with all rows produced by the corresponding call of the table function. If a table function call returns an empty result, the corresponding outer row is preserved and the result padded with null values.

Note: Currently, the predicate of a table function left outer join can only be empty or literal true.

// instantiate User-Defined Table Function
val split: TableFunction[_] = new MySplitUDTF()

// join
val result: Table = table
    .leftOuterJoinLateral(split($"c") as ("s", "t", "v"))
    .select($"a", $"b", $"s", $"t", $"v")

Temporal tables are tables that track their changes over time.

A temporal table function provides access to the state of a temporal table at a specific point in time. The syntax to join a table with a temporal table function is the same as in Inner Join with Table Function.

Currently only inner joins with temporal tables are supported.

val ratesHistory = tableEnv.from("RatesHistory")

// register temporal table function with a time attribute and primary key
val rates = ratesHistory.createTemporalTableFunction($"r_proctime", $"r_currency")

// join with "Orders" based on the time attribute and key
val orders = tableEnv.from("Orders")
val result = orders
    .joinLateral(rates($"o_rowtime"), $"r_currency" === $"o_currency")

For more information please check the more detailed temporal tables concept description.

类似于SQL的JOIN子句。对两张表执行内连接操作。两张表必须具有不同的字段名称，并且必须在join方法或者随后的where或filter方法中定义至少一个等值连接条件。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("d, e, f")
result = left.join(right).where("a = d").select("a, b, e")

注意： 对于流式查询，计算查询结果所需的状态（state）可能会无限增长，具体取决于不重复的输入行的数量。您可能需要在查询配置中设置状态保留时间，以防止状态过大。详情请看查询配置。

类似于SQL的LEFT/RIGHT/FULL OUTER JOIN子句。对两张表执行外连接操作。两张表必须具有不同的字段名称，并且必须定义至少一个等值连接条件。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("d, e, f")

left_outer_result = left.left_outer_join(right, "a = d").select("a, b, e")
right_outer_result = left.right_outer_join(right, "a = d").select("a, b, e")
full_outer_result = left.full_outer_join(right, "a = d").select("a, b, e")

注意： 对于流式查询，计算查询结果所需的状态（state）可能会无限增长，具体取决于不重复的输入行的数量。您可能需要在查询配置中设置状态保留时间，以防止状态过大。详情请看查询配置。

注意： Interval Join 是所有常规流式数据处理join操作中的其中一种场景。

Interval Join 要求至少有一个等值连接条件以及一个用于划定时间间隔的条件。对两条数据流时间的划定可以通过两个范围比较确定一个合适时间区间(<, <=, >=, >)， 也可以简单的通过对相同时间属性（当前处理时间或事件时间）的时间值进行等值比较确定。

如下示例是合法的Interval Join条件：

• ltime = rtime
• ltime >= rtime && ltime < rtime + 2.second

left = t_env.from_path("Source1").select("a, b, c, rowtime1")
right = t_env.from_path("Source2").select("d, e, f, rowtime2")

result = left.join(right).where("a = d && rowtime1 >= rowtime2 - 1.second 
                     && rowtime1 <= rowtime2 + 2.second").select("a, b, e, rowtime1")
            

将一张表与一个表函数的执行结果执行内连接操作。左表的每一行都会进行一次表函数调用，调用将会返回0个，1个或多个结果，再与这些结果执行连接操作。如果一行数据对应的表函数调用返回了一个空的结果集，则这行数据会被丢弃。

# register User-Defined Table Function
@udtf(input_types=[DataTypes.BIGINT()],
      result_types=[DataTypes.BIGINT(), DataTypes.BIGINT(), DataTypes.BIGINT()])
def split(x):
    return [Row(1, 2, 3)]
t_env.register_function("split", split)

# join
orders = t_env.from_path("Orders")
result = orders.join_lateral("split(c).as(s, t, v)").select("a, b, s, t, v")

将一张表与一个表函数的执行结果执行左连接操作。左表的每一行都会进行一次表函数调用，调用将会返回0个，1个或多个结果，再与这些结果执行连接操作。如果一行数据对应的表函数调用返回了一个空的结果集，这行数据依然会被保留，对应的右表数值用null(python为None)填充。

注意：目前，表函数的左连接操作的连接条件(join predicate)只能为空或者为"true"常量。

# register User-Defined Table Function
@udtf(input_types=[DataTypes.BIGINT()],
      result_types=[DataTypes.BIGINT(), DataTypes.BIGINT(), DataTypes.BIGINT()])
def split(x):
    return [Row(1, 2, 3)]
t_env.register_function("split", split)

# join
orders = t_env.from_path("Orders")
result = orders.left_outer_join_lateral("split(c).as(s, t, v)").select("a, b, s, t, v")

Python Table API暂不支持。

Similar to a SQL UNION clause. Unions two tables with duplicate records removed. Both tables must have identical field types.

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

Similar to a SQL UNION ALL clause. Unions two tables. Both tables must have identical field types.

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

Similar to a SQL INTERSECT clause. Intersect returns records that exist in both tables. If a record is present one or both tables more than once, it is returned just once, i.e., the resulting table has no duplicate records. Both tables must have identical field types.

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

Similar to a SQL INTERSECT ALL clause. IntersectAll returns records that exist in both tables. If a record is present in both tables more than once, it is returned as many times as it is present in both tables, i.e., the resulting table might have duplicate records. Both tables must have identical field types.

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

Similar to a SQL EXCEPT clause. Minus returns records from the left table that do not exist in the right table. Duplicate records in the left table are returned exactly once, i.e., duplicates are removed. Both tables must have identical field types.

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

Similar to a SQL EXCEPT ALL clause. MinusAll returns the records that do not exist in the right table. A record that is present n times in the left table and m times in the right table is returned (n - m) times, i.e., as many duplicates as are present in the right table are removed. Both tables must have identical field types.

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

Similar to a SQL IN clause. In returns true if an expression exists in a given table sub-query. The sub-query table must consist of one column. This column must have the same data type as the expression.

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

Table result = left.select($("a"), $("b"), $("c")).where($("a").in(right));

Note: For streaming queries the operation is rewritten in a join and group operation. The required state to compute the query result might grow infinitely depending on the number of distinct input rows. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

Similar to a SQL UNION clause. Unions two tables with duplicate records removed, both tables must have identical field types.

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

Similar to a SQL UNION ALL clause. Unions two tables, both tables must have identical field types.

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

Similar to a SQL INTERSECT clause. Intersect returns records that exist in both tables. If a record is present in one or both tables more than once, it is returned just once, i.e., the resulting table has no duplicate records. Both tables must have identical field types.

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

Similar to a SQL INTERSECT ALL clause. IntersectAll returns records that exist in both tables. If a record is present in both tables more than once, it is returned as many times as it is present in both tables, i.e., the resulting table might have duplicate records. Both tables must have identical field types.

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

Similar to a SQL EXCEPT clause. Minus returns records from the left table that do not exist in the right table. Duplicate records in the left table are returned exactly once, i.e., duplicates are removed. Both tables must have identical field types.

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

Similar to a SQL EXCEPT ALL clause. MinusAll returns the records that do not exist in the right table. A record that is present n times in the left table and m times in the right table is returned (n - m) times, i.e., as many duplicates as are present in the right table are removed. Both tables must have identical field types.

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

Similar to a SQL IN clause. In returns true if an expression exists in a given table sub-query. The sub-query table must consist of one column. This column must have the same data type as the expression.

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))

Note: For streaming queries the operation is rewritten in a join and group operation. The required state to compute the query result might grow infinitely depending on the number of distinct input rows. Please provide a query configuration with valid retention interval to prevent excessive state size. See Query Configuration for details.

类似于SQL的UNION子句。将两张表组合成一张表，这张表拥有二者去除重复后的全部数据。两张表的字段和类型必须完全一致。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("a, b, c")
result = left.union(right)

类似于SQL的UNION ALL子句。将两张表组合成一张表，这张表拥有二者的全部数据。两张表的字段和类型必须完全一致。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("a, b, c")
result = left.union_all(right)

类似于SQL的INTERSECT子句。Intersect返回在两张表中都存在的数据。如果一个记录在两张表中不止出现一次，则只返回一次，即结果表没有重复记录。两张表的字段和类型必须完全一致。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("a, b, c")
result = left.intersect(right)

类似于SQL的INTERSECT ALL子句。IntersectAll返回在两张表中都存在的数据。如果一个记录在两张表中不止出现一次，则按照它在两张表中都出现的次数返回，即结果表可能包含重复数据。两张表的字段和类型必须完全一致。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("a, b, c")
result = left.intersect_all(right)

类似于SQL的EXCEPT子句。Minus返回仅存在于左表，不存在于右表中的数据。左表中的相同数据只会返回一次，即数据会被去重。两张表的字段和类型必须完全一致。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("a, b, c")
result = left.minus(right);

类似于SQL的EXCEPT ALL子句。MinusAll返回仅存在于左表，不存在于右表中的数据。如果一条数据在左表中出现了n次，在右表中出现了m次，最终这条数据将会被返回(n - m)次，即按右表中出现的次数来移除数据。两张表的字段和类型必须完全一致。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("a, b, c")
result = left.minus_all(right)

类似于SQL的IN子句。如果In左边表达式的值在给定的子查询结果中则返回true。子查询的结果必须为单列。此列数据类型必须和表达式一致。

left = t_env.from_path("Source1").select("a, b, c")
right = t_env.from_path("Source2").select("a")

# using implicit registration
result = left.select("a, b, c").where("a.in(%s)" % right)

# using explicit registration
t_env.create_temporary_view("RightTable", right)
result = left.select("a, b, c").where("a.in(RightTable)")

注意： 对于流式查询，这个操作会被替换成一个连接操作和一个分组操作。计算查询结果所需的状态（state）可能会无限增长，具体取决于不重复的输入行的数量。您可能需要在查询配置中设置状态保留时间，以防止状态过大。详情请看查询配置。

Similar to a SQL ORDER BY clause. Returns records globally sorted across all parallel partitions.

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

Similar to the SQL OFFSET and FETCH clauses. Offset and Fetch limit the number of records returned from a sorted result. Offset and Fetch are technically part of the Order By operator and thus must be preceded by it.

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

// returns the first 5 records from the sorted result
Table result1 = in.orderBy($("a").asc()).fetch(5);

// skips the first 3 records and returns all following records from the sorted result
Table result2 = in.orderBy($("a").asc()).offset(3);

// skips the first 10 records and returns the next 5 records from the sorted result
Table result3 = in.orderBy($("a").asc()).offset(10).fetch(5);

Similar to a SQL ORDER BY clause. Returns records globally sorted across all parallel partitions.

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

Similar to the SQL OFFSET and FETCH clauses. Offset and Fetch limit the number of records returned from a sorted result. Offset and Fetch are technically part of the Order By operator and thus must be preceded by it.

val in = ds.toTable(tableEnv, $"a", $"b", $"c")

// returns the first 5 records from the sorted result
val result1: Table = in.orderBy($"a".asc).fetch(5)

// skips the first 3 records and returns all following records from the sorted result
val result2: Table = in.orderBy($"a".asc).offset(3)

// skips the first 10 records and returns the next 5 records from the sorted result
val result3: Table = in.orderBy($"a".asc).offset(10).fetch(5)

类似于SQL的ORDER BY子句。返回包括所有子并发分区内所有数据的全局排序结果。

in = t_env.from_path("Source1").select("a, b, c")
result = in.order_by("a.asc")

类似于SQL的OFFSET和FETCH子句。Offset和Fetch从已排序的结果中返回指定数量的数据。Offset和Fetch在技术上是Order By操作的一部分，因此必须紧跟其后出现。

table = t_env.from_path("Source1").select("a, b, c")

# returns the first 5 records from the sorted result
result1 = table.order_by("a.asc").fetch(5)

# skips the first 3 records and returns all following records from the sorted result
result2 = table.order_by("a.asc").offset(3)

# skips the first 10 records and returns the next 5 records from the sorted result
result3 = table.order_by("a.asc").offset(10).fetch(5)

Similar to the `INSERT INTO` clause in a SQL query, the method performs an insertion into a registered output table. The `executeInsert()` method will immediately submit a Flink job which execute the insert operation.

Output tables must be registered in the TableEnvironment (see Register a TableSink). Moreover, the schema of the registered table must match the schema of the query.

Table orders = tableEnv.from("Orders");
orders.executeInsert("OutOrders");

Similar to the `INSERT INTO` clause in a SQL query, the method performs an insertion into a registered output table. The `executeInsert()` method will immediately submit a Flink job which execute the insert operation.

Output tables must be registered in the TableEnvironment (see Connector tables). Moreover, the schema of the registered table must match the schema of the query.

val orders: Table = tableEnv.from("Orders")
orders.executeInsert("OutOrders")

类似于SQL请求中的INSERT INTO子句。将数据输出到一个已注册的输出表中。`execute_insert` 方法会立即提交一个 Flink 作业，触发插入操作。

输出表必须先在TableEnvironment中注册（详见注册一个TableSink）。此外，注册的表的模式（schema）必须和请求的结果的模式（schema）相匹配。

orders = t_env.from_path("Orders")
orders.execute_insert("OutOrders")