下面的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");
}
}