Почему ListView работает лучше, чем ArrayList

Я столкнулся с проблемой при настройке UDAF с использованием Flink (версия 1.20). Я хотел реализовать UDAF, который рассчитывает медиану, и использовал следующие два метода:

public class MedianUDAF2 extends AggregateFunction<Double, MedianUDAF2.State> {

    public static class State {

        public int scale = 2;

        public ListView<Double> numbers;

        public State() {}
    }

    @Override
    public State createAccumulator() {
        State state = new State();
        state.numbers = new ListView<>();
        return state;
    }

    public void accumulate(State acc, Double val, Integer scale) throws Exception {
        acc.numbers.add(val);
        if (scale != null && scale > 0) acc.scale = scale;
    }

    public void merge(State acc, Iterable<State> it) throws Exception {
        for (State a : it) {
            acc.numbers.addAll(a.numbers.getList());
        }
    }

    @Override
    public Double getValue(State acc) {
        try {
            List<Double> numbers = acc.numbers.getList();
            numbers.sort(Double::compareTo);
            double n = numbers.size() - 1;
            double index = n * 0.5;

            int low = (int) Math.floor(index);
            int high = (int) Math.ceil(index);

            double value = low == high ? (numbers.get(low) + numbers.get(high)) * 0.5 : numbers.get(high);
            BigDecimal decimal = new BigDecimal(value);
            return decimal.setScale(acc.scale, BigDecimal.ROUND_HALF_UP).doubleValue();
        } catch (Exception ignored) {
        }
        return 0.0;
    }
}

public class MedianUDAF extends AggregateFunction<Double, MedianUDAF.State> {

    public static class State {

        public int scale = 2;

        @DataTypeHint(value = "ARRAY<DOUBLE>")
        public ArrayList<Double> numbers;

        public State() {}
    }

    @Override
    public State createAccumulator() {
        State state = new State();
        state.numbers = new ArrayList<>();
        return state;
    }

    public void accumulate(State acc, Double val, Integer scale) throws Exception {
        acc.numbers.add(val);
        if (scale != null && scale > 0) acc.scale = scale;
    }

    public void merge(State acc, Iterable<State> it) throws Exception {
        for (State a : it) {
            acc.numbers.addAll(a.numbers);
        }
    }

    @Override
    public Double getValue(State acc) {
        try {
            List<Double> numbers = acc.numbers;
            numbers.sort(Double::compareTo);
            double n = numbers.size() - 1;
            double index = n * 0.5;

            int low = (int) Math.floor(index);
            int high = (int) Math.ceil(index);

            double value = low == high ? (numbers.get(low) + numbers.get(high)) * 0.5 : numbers.get(high);
            BigDecimal decimal = new BigDecimal(value);
            return decimal.setScale(acc.scale, BigDecimal.ROUND_HALF_UP).doubleValue();
        } catch (Exception ignored) {
        }
        return 0.0;
    }
}

tableEnvironment.createTemporarySystemFunction("median", new MedianUDAF()); // Или new MedianUDAF2()
Table table = tableEnvironment.sqlQuery("select median(l_linenumber, 2) from lineitem");

Их отличие заключается в том, что один использует ArrayList, а другой – ListView в State, и разница в производительности очень большая. Почему?

В комментариях к ListView говорится, что он будет использовать бэкенд состояния при больших объемах данных. До Flink-table-planner 1.14 в addAccumulatorDataViews в AggregationCodeGenerator можно было видеть этот процесс конверсии, но в версии 1.20 он больше не виден. Я пытался отладить этот процесс конверсии в классе AggsHandlerCodeGenerator, но не смог добиться успеха. Могу ли я спросить, где произошел этот процесс конверсии и как я должен наблюдать за этим явлением? Большое спасибо!!!