如何在flink中将int列聚合到array<int>？

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

// 创建TableEnvironment
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 注册输入表
tableEnv.createTemporaryView("inputTable", inputDataStream, "intColumn");

// 执行聚合操作
Table resultTable = tableEnv.sqlQuery("SELECT COLLECT(intColumn) AS intArray FROM inputTable");

// 将结果转换为DataStream
DataStream<Row> resultDataStream = tableEnv.toAppendStream(resultTable, Row.class);

// 打印结果
resultDataStream.print();

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 创建输入数据流
DataStream<Tuple2<String, Integer>> inputDataStream = env.fromElements(
    Tuple2.of("key", 1),
    Tuple2.of("key", 2),
    Tuple2.of("key", 3)
);

// 按照key进行分组，并在5秒的时间窗口内进行聚合
DataStream<Tuple2<String, Integer[]>> resultDataStream = inputDataStream
    .keyBy(0)
    .timeWindow(Time.seconds(5))
    .aggregate(new IntArrayAggregator(), new IntArrayWindowFunction());

// 打印结果
resultDataStream.print();

// 定义聚合函数
public class IntArrayAggregator implements AggregateFunction<Tuple2<String, Integer>, List<Integer>, Integer[]> {
    @Override
    public List<Integer> createAccumulator() {
        return new ArrayList<>();
    }

    @Override
    public List<Integer> add(Tuple2<String, Integer> value, List<Integer> accumulator) {
        accumulator.add(value.f1);
        return accumulator;
    }

    @Override
    public Integer[] getResult(List<Integer> accumulator) {
        return accumulator.toArray(new Integer[0]);
    }

    @Override
    public List<Integer> merge(List<Integer> a, List<Integer> b) {
        a.addAll(b);
        return a;
    }
}

// 定义WindowFunction
public class IntArrayWindowFunction implements WindowFunction<Integer[], Tuple2<String, Integer[]>, String, TimeWindow> {
    @Override
    public void apply(String key, TimeWindow window, Iterable<Integer[]> input, Collector<Tuple2<String, Integer[]>> out) {
        Integer[] result = input.iterator().next();
        out.collect(Tuple2.of(key, result));
    }
}