Flink Table API 中添加新列的常见误区与正确实践

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

public class AddColumnLiteralExample {

    public static void main(String[] args) throws Exception {
        // 1. 设置 TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tEnv = TableEnvironment.create(settings);

        // 2. 创建一个示例表（模拟现有数据）
        // 假设原始表有 id 和 name 列
        Table inputTable = tEnv.fromValues(
            row(1, "Alice"),
            row(2, "Bob"),
            row(3, "Charlie")
        ).as("id", "name");

        System.out.println("原始表 Schema:");
        inputTable.printSchema();
        // 原始表 Schema:
        // root
        //  |-- id: INT
        //  |-- name: STRING

        // 3. 正确添加一个新列 "Status"，其值为字面量 "Active"
        Table tableWithNewColumn = inputTable.addColumns(
            lit("Active").as("Status") // 使用 lit() 定义字面量值，并用 .as() 命名
        );

        System.out.println("\n添加新列后的表 Schema:");
        tableWithNewColumn.printSchema();
        // 添加新列后的表 Schema:
        // root
        //  |-- id: INT
        //  |-- name: STRING
        //  |-- Status: STRING

        // 4. 验证数据 (可选)
        // tableWithNewColumn.execute().print();
        // +----+---------+--------+
        // | id |    name | Status |
        // +----+---------+--------+
        // |  1 |   Alice | Active |
        // |  2 |     Bob | Active |
        // |  3 | Charlie | Active |
        // +----+---------+--------+
    }
}

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

public class AddColumnComputedExample {

    public static void main(String[] args) throws Exception {
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tEnv = TableEnvironment.create(settings);

        Table inputTable = tEnv.fromValues(
            row(1, "John", "Doe"),
            row(2, "Jane", "Smith")
        ).as("id", "firstName", "lastName");

        System.out.println("原始表 Schema:");
        inputTable.printSchema();
        // 原始表 Schema:
        // root
        //  |-- id: INT
        //  |-- firstName: STRING
        //  |-- lastName: STRING

        // 3. 正确添加一个新列 "fullName"，它是 firstName 和 lastName 的拼接
        Table tableWithFullName = inputTable.addColumns(
            concat($("firstName"), lit(" "), $("lastName")).as("fullName") // 使用 concat() 拼接，并用 .as() 命名
        );

        System.out.println("\n添加新列后的表 Schema:");
        tableWithFullName.printSchema();
        // 添加新列后的表 Schema:
        // root
        //  |-- id: INT
        //  |-- firstName: STRING
        //  |-- lastName: STRING
        //  |-- fullName: STRING

        // 4. 验证数据 (可选)
        // tableWithFullName.execute().print();
        // +----+-----------+----------+-----------+
        // | id | firstName | lastName |  fullName |
        // +----+-----------+----------+-----------+
        // |  1 |      John |      Doe |   John Doe |
        // |  2 |      Jane |    Smith | Jane Smith |
        // +----+-----------+----------+-----------+
    }
}

// 假设 inputTable 已经有 "id" 和 "name" 列
Table inputTable = tEnv.fromValues(
    row(1, "Alice"),
    row(2, "Bob")
).as("id", "name");

// 使用 addOrReplaceColumns 替换 "name" 列
Table replacedTable = inputTable.addOrReplaceColumns(
    concat(lit("User_"), $("id")).as("name") // 替换 name 列
);
System.out.println("\n替换 'name' 列后的表 Schema:");
replacedTable.printSchema();
// Schema 相同，但 'name' 列的值已改变
// replacedTable.execute().print();
// +----+--------+
// | id |   name |
// +----+--------+
// |  1 | User_1 |
// |  2 | User_2 |
// +----+--------+