Table API

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

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

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

Table orders = tableEnv.scan("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.scan("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.scan("Orders");
Table result = orders.where("b === 'red'");

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

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

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

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

val orders: Table = tableEnv.scan("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.scan("Orders")
val result = orders.select('*)

Renames fields.

val orders: Table = tableEnv.scan("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.scan("Orders")
val result = orders.filter('a % 2 === 0)

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

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

orders = table_env.scan("Orders")

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

orders = table_env.scan("Orders")
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.

result = orders.select("*")

Renames fields.

orders = table_env.scan("Orders")
result = orders.alias("x, y, z, t")

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

orders = table_env.scan("Orders")
result = orders.where("b === 'red'")

orders = table_env.scan("Orders")
result = orders.filter("a % 2 === 0")

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

Table orders = tableEnv.scan("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.scan("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.scan("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.scan("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.scan("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.scan("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.scan("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.scan("Orders");
val 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.

orders = table_env.scan("Orders")
result = orders.add_columns("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.

orders = table_env.scan("Orders")
result = orders.add_or_replace_columns("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.

orders = table_env.scan("Orders")
result = orders.drop_columns("b, c")

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

orders = table_env.scan("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.scan("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.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

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

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.scan("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("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.scan("Orders");

// Use distinct aggregation for user-defined aggregate functions
tEnv.registerFunction("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.

Table orders = tableEnv.scan("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.scan("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.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

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

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.scan("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.scan("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.scan("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. 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.

orders = table_env.scan("Orders")
result = orders.group_by("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.

orders = table_env.scan("Orders")
result = orders.window(Tumble.over("5.minutes").on("rowtime").alias("w")) \ 
               .group_by("a, w") \
               .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.

orders = table_env.scan("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")

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.

orders = table_env.scan("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")).groupBy("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")

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.

orders = table_env.scan("Orders")
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.

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

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: Time-windowed joins are a subset of regular joins that can be processed in a streaming fashion.

A time-windowed 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 window 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("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.

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

// join
Table orders = tableEnv.scan("Orders");
Table result = orders
    .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.

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

// join
Table orders = tableEnv.scan("Orders");
Table result = orders
    .leftOuterJoinLateral("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.scan("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.scan("Orders");
Table result = orders
    .joinLateral("rates(o_proctime)", "o_currency = 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: Time-windowed joins are a subset of regular joins that can be processed in a streaming fashion.

A time-windowed 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 window 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.scan("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.scan("Orders")
val result = orders
    .joinLateral(rates('o_rowtime), 'r_currency === 'o_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.

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("Source2").select("d, e, f")
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.

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("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")

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.

Currently not supported in python API.

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 Java User-Defined Table Function
table_env.register_java_function("split", "com.my.udf.MySplitUDTF")

# join
orders = table_env.scan("Orders")
result = orders.join_lateral("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 Java User-Defined Table Function
table_env.register_java_function("split", "com.my.udf.MySplitUDTF")

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

Currently not supported in python 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");

// using implicit registration
Table result = left.select("a, b, c").where("a.in(" + right + ")");

// using explicit registration
tableEnv.registerTable("RightTable", right);
Table result = left.select("a, b, c").where("a.in(RightTable)");

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.

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

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

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

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("Source2").select("a, b, c")
result = left.union_all(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.

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("Source2").select("a, b, c")
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.

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("Source2").select("a, b, c")
result = left.intersect_all(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.

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("Source2").select("a, b, c")
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.

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("Source2").select("a, b, c")
result = left.minus_all(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.

left = table_env.scan("Source1").select("a, b, c")
right = table_env.scan("Source2").select("a")

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

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

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

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

in = table_env.scan("Source1").select("a, b, c")
result = in.order_by("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.

in = table_env.scan("Source1").select("a, b, c")

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

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

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

Similar to the INSERT INTO clause in a SQL query. Performs a insertion into a registered output table.

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.scan("Orders");
orders.insertInto("OutOrders");

Similar to the INSERT INTO clause in a SQL query. Performs a insertion into a registered output table.

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.

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

Similar to the INSERT INTO clause in a SQL query. Performs a insertion into a registered output table.

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.

orders = table_env.scan("Orders");
orders.insert_into("OutOrders");

Performs a map operation with a user-defined scalar function or built-in scalar function. The output will be flattened if the output type is a composite type.

public class MyMapFunction extends ScalarFunction {
    public Row eval(String a) {
        return Row.of(a, "pre-" + a);
    }

    @Override
    public TypeInformation<?> getResultType(Class<?>[] signature) {
        return Types.ROW(Types.STRING(), Types.STRING());
    }
}

ScalarFunction func = new MyMapFunction();
tableEnv.registerFunction("func", func);

Table table = input
  .map("func(c)").as("a, b")

Performs a flatMap operation with a table function.

public class MyFlatMapFunction extends TableFunction<Row> {

    public void eval(String str) {
        if (str.contains("#")) {
            String[] array = str.split("#");
            for (int i = 0; i < array.length; ++i) {
                collect(Row.of(array[i], array[i].length()));
            }
        }
    }

    @Override
    public TypeInformation<Row> getResultType() {
        return Types.ROW(Types.STRING(), Types.INT());
    }
}

TableFunction func = new MyFlatMapFunction();
tableEnv.registerFunction("func", func);

Table table = input
  .flatMap("func(c)").as("a, b")

Performs an aggregate operation with an aggregate function. You have to close the "aggregate" with a select statement and the select statement does not support aggregate functions. The output of aggregate will be flattened if the output type is a composite type.

public class MyMinMaxAcc {
    public int min = 0;
    public int max = 0;
}

public class MyMinMax extends AggregateFunction<Row, MyMinMaxAcc> {

    public void accumulate(MyMinMaxAcc acc, int value) {
        if (value < acc.min) {
            acc.min = value;
        }
        if (value > acc.max) {
            acc.max = value;
        }
    }

    @Override
    public MyMinMaxAcc createAccumulator() {
        return new MyMinMaxAcc();
    }
    
    public void resetAccumulator(MyMinMaxAcc acc) {
        acc.min = 0;
        acc.max = 0;
    }

    @Override
    public Row getValue(MyMinMaxAcc acc) {
        return Row.of(acc.min, acc.max);
    }

    @Override
    public TypeInformation<Row> getResultType() {
        return new RowTypeInfo(Types.INT, Types.INT);
    }
}
    
AggregateFunction myAggFunc = new MyMinMax();
tableEnv.registerFunction("myAggFunc", myAggFunc);
Table table = input
  .groupBy("key")
  .aggregate("myAggFunc(a) as (x, y)")
  .select("key, x, y")

Groups and aggregates a table on a group window and possibly one or more grouping keys. You have to close the "aggregate" with a select statement. And the select statement does not support "*" or aggregate functions.

AggregateFunction myAggFunc = new MyMinMax();
tableEnv.registerFunction("myAggFunc", myAggFunc);

Table table = input
    .window(Tumble.over("5.minutes").on("rowtime").as("w")) // define window
    .groupBy("key, w") // group by key and window
    .aggregate("myAggFunc(a) as (x, y)")
    .select("key, x, y, w.start, w.end"); // access window properties and aggregate results

Similar to a GroupBy Aggregation. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.

Instead of using emitValue to output results, you can also use the emitUpdateWithRetract method. Different from emitValue, emitUpdateWithRetract is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The emitUpdateWithRetract method will be used in preference to the emitValue method if both methods are defined in the table aggregate function, because the method is treated to be more efficient than emitValue as it can output values incrementally. See Table Aggregation Functions for details.

/**
 * Accumulator for Top2.
 */
public class Top2Accum {
    public Integer first;
    public Integer second;
}

/**
 * The top2 user-defined table aggregate function.
 */
public class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {

    @Override
    public Top2Accum createAccumulator() {
        Top2Accum acc = new Top2Accum();
        acc.first = Integer.MIN_VALUE;
        acc.second = Integer.MIN_VALUE;
        return acc;
    }


    public void accumulate(Top2Accum acc, Integer v) {
        if (v > acc.first) {
            acc.second = acc.first;
            acc.first = v;
        } else if (v > acc.second) {
            acc.second = v;
        }
    }

    public void merge(Top2Accum acc, java.lang.Iterable<Top2Accum> iterable) {
        for (Top2Accum otherAcc : iterable) {
            accumulate(acc, otherAcc.first);
            accumulate(acc, otherAcc.second);
        }
    }

    public void emitValue(Top2Accum acc, Collector<Tuple2<Integer, Integer>> out) {
        // emit the value and rank
        if (acc.first != Integer.MIN_VALUE) {
            out.collect(Tuple2.of(acc.first, 1));
        }
        if (acc.second != Integer.MIN_VALUE) {
            out.collect(Tuple2.of(acc.second, 2));
        }
    }
}
    
tEnv.registerFunction("top2", new Top2());
Table orders = tableEnv.scan("Orders");
Table result = orders
    .groupBy("key")
    .flatAggregate("top2(a) as (v, rank)")
    .select("key, v, rank");

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 a 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. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.

tableEnv.registerFunction("top2", new Top2());
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
    .flatAggregate("top2(b) as (v, rank)")
    .select("a, w.start, w.end, w.rowtime, v, rank"); // access window properties and aggregate results

Performs a map operation with a user-defined scalar function or built-in scalar function. The output will be flattened if the output type is a composite type.

class MyMapFunction extends ScalarFunction {
  def eval(a: String): Row = {
    Row.of(a, "pre-" + a)
  }

  override def getResultType(signature: Array[Class[_]]): TypeInformation[_] =
    Types.ROW(Types.STRING, Types.STRING)
}

val func = new MyMapFunction()
val table = input
  .map(func('c)).as('a, 'b)

Performs a flatMap operation with a table function.

class MyFlatMapFunction extends TableFunction[Row] {
  def eval(str: String): Unit = {
    if (str.contains("#")) {
      str.split("#").foreach({ s =>
        val row = new Row(2)
        row.setField(0, s)
        row.setField(1, s.length)
        collect(row)
      })
    }
  }

  override def getResultType: TypeInformation[Row] = {
    Types.ROW(Types.STRING, Types.INT)
  }
}

val func = new MyFlatMapFunction
val table = input
  .flatMap(func('c)).as('a, 'b)

Performs an aggregate operation with an aggregate function. You have to close the "aggregate" with a select statement and the select statement does not support aggregate functions. The output of aggregate will be flattened if the output type is a composite type.

case class MyMinMaxAcc(var min: Int, var max: Int)

class MyMinMax extends AggregateFunction[Row, MyMinMaxAcc] {

  def accumulate(acc: MyMinMaxAcc, value: Int): Unit = {
    if (value < acc.min) {
      acc.min = value
    }
    if (value > acc.max) {
      acc.max = value
    }
  }

  override def createAccumulator(): MyMinMaxAcc = MyMinMaxAcc(0, 0)
  
  def resetAccumulator(acc: MyMinMaxAcc): Unit = {
    acc.min = 0
    acc.max = 0
  }

  override def getValue(acc: MyMinMaxAcc): Row = {
    Row.of(Integer.valueOf(acc.min), Integer.valueOf(acc.max))
  }

  override def getResultType: TypeInformation[Row] = {
    new RowTypeInfo(Types.INT, Types.INT)
  }
}

val myAggFunc = new MyMinMax
val table = input
  .groupBy('key)
  .aggregate(myAggFunc('a) as ('x, 'y))
  .select('key, 'x, 'y)