Flink Table API中添加新列的正确姿势与常见陷阱

import org.apache.flink.table.api.*;
import static org.apache.flink.table.api.Expressions.*;

// 假设 tEnv 已经初始化，并且 orders 是一个已存在的 Table 对象
// orders 的 schema: [productName: STRING, quantity: INT, price: DECIMAL]
Table orders = tEnv.fromValues(
    row("Laptop", 2, 1200.00),
    row("Mouse", 5, 25.00),
    row("Keyboard", 1, 75.00)
).as("productName", "quantity", "price");

// 错误示例 (会抛出 ValidationException)
// Table errorTable = orders.addColumns($("orderDescription"));

// 正确示例：使用 concat 表达式拼接字符串，并使用 .as() 指定新列名
Table result = orders.addColumns(
    concat($("productName"), lit(" x "), $("quantity")).as("orderDescription")
);

// 打印结果表的 schema 和内容
result.printSchema();
// root
//  |-- productName: STRING
//  |-- quantity: INT
//  |-- price: DECIMAL(5,2)
//  |-- orderDescription: STRING

tEnv.toDataStream(result).print();
// (Laptop,2,1200.00,Laptop x 2)
// (Mouse,5,25.00,Mouse x 5)
// (Keyboard,1,75.00,Keyboard x 1)

import org.apache.flink.table.api.*;
import static org.apache.flink.table.api.Expressions.*;

Table orders = tEnv.fromValues(
    row("Laptop", 2),
    row("Mouse", 5)
).as("productName", "quantity");

// 添加一个名为 "source" 的常量列，值为 "OnlineStore"
Table resultWithConstant = orders.addColumns(
    lit("OnlineStore").as("source")
);

resultWithConstant.printSchema();
// root
//  |-- productName: STRING
//  |-- quantity: INT
//  |-- source: STRING

tEnv.toDataStream(resultWithConstant).print();
// (Laptop,2,OnlineStore)
// (Mouse,5,OnlineStore)

import org.apache.flink.table.api.*;
import static org.apache.flink.table.api.Expressions.*;

Table orders = tEnv.fromValues(
    row("Laptop", 2),
    row("Mouse", 10)
).as("productName", "quantity");

// 根据 quantity 判断 orderType
Table resultWithConditionalColumn = orders.addColumns(
    when($("quantity").isGreater(5))
        .then(lit("BulkOrder"))
        .otherwise(lit("StandardOrder"))
        .as("orderType")
);

resultWithConditionalColumn.printSchema();
// root
//  |-- productName: STRING
//  |-- quantity: INT
//  |-- orderType: STRING

tEnv.toDataStream(resultWithConditionalColumn).print();
// (Laptop,2,StandardOrder)
// (Mouse,10,BulkOrder)