下面的flink代码：

String sqlQuery = “SELECT MAX(val) AS max_val, MIN(val) AS min_val FROM dataT GROUP BY pid”;

Table resultTable = tableEnv.sqlQuery(sqlQuery);

DataStream resultStream = tableEnv.toAppendStream(resultTable, Row.class);

运行报错：

org.apache.flink.table.api.TableException: toAppendStream doesn’t support consuming update changes which is produced by node GroupAggregate(groupBy=[pid], select=[pid, MAX(val) AS max_val, MIN(val) AS min_val])

原因分析

报错信息提示 toAppendStream doesn't support consuming update changes which is produced by node GroupAggregate，这意味着在当前代码中使用 toAppendStream 方法去转换结果表为 DataStream 时出现了不兼容的情况。
在 Flink 中，当执行包含聚合操作（比如这里的 GROUP BY 以及 MAX、MIN 聚合函数计算）的 SQL 查询时，查询结果可能会产生更新（update）类型的变更，而 toAppendStream 方法只适用于那种仅追加（append-only）类型的结果，也就是结果表中数据只会新增而不会有更新、删除等变更的情况。这里的聚合操作导致了结果存在更新变化，所以调用 toAppendStream 就抛出了异常，它无法处理这种带有更新的数据变更情况。

正确代码修改思路及示例

要解决这个问题，可以使用 toRetractStream 方法来替代 toAppendStream 方法，toRetractStream 方法可以处理包含更新、删除等多种变更类型的数据，它返回的 DataStream 中元素是包含了一个布尔值标志（表示是新增还是撤回操作）以及实际的数据行（对应查询结果行）的二元组形式。
以下是修改后的代码示例：

        // 2. 添加 Source
        DataStream<RunData> dataSource = env.addSource(new TDengineSourceFunction(jdbcUrl, user, password, query));

        // 3. 注册临时表
        tableEnv.createTemporaryView("rundata", dataSource, "pid, val"); // 根据实际字段调整

        // 4. 执行 SQL 查询以计算最大值和最小值
        String sqlQuery = "SELECT MAX(val) AS max_val, MIN(val) AS min_val FROM dataT GROUP BY pid";
        Table resultTable = tableEnv.sqlQuery(sqlQuery);

        // 5. 将结果转换为 DataStream 并打印，这里使用 toRetractStream 替代 toAppendStream
        DataStream<Tuple2<Boolean, Row>> resultStream = tableEnv.toRetractStream(resultTable, Row.class);
        resultStream.print();

        // 6. 触发执行
        env.execute("Flink SQL Max and Min Calculation");
    }
}