Repository: apache/flink
File: flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/buffers/RecordsWindowBuffer.java (Java)

Code before:

    private long minSliceEnd = Long.MAX_VALUE;

    public RecordsWindowBuffer(
            Object operatorOwner,
            MemoryManager memoryManager,
            long memorySize,
            RecordsCombiner combineFunction,
            PagedTypeSerializer<RowData> keySer,
            AbstractRowDataSerializer<RowData> inputSer,
            boolean requiresCopy,
            ZoneId shiftTimeZone) {
        this.combineFunction = combineFunction;
        this.recordsBuffer =
                new WindowBytesMultiMap(
                        operatorOwner, memoryManager, memorySize, keySer, inputSer.getArity());
        this.recordSerializer = inputSer;
        this.reuseWindowKey = new WindowKeySerializer(keySer).createInstance();
        this.requiresCopy = requiresCopy;
        this.shiftTimeZone = shiftTimeZone;
    }

    @Override
    public void addElement(RowData key, long sliceEnd, RowData element) throws Exception {
        // track the lowest trigger time, if watermark exceeds the trigger time,
        // it means there are some elements in the buffer belong to a window going to be fired,
        // and we need to flush the buffer into state for firing.
        minSliceEnd = Math.min(sliceEnd, minSliceEnd);

        reuseWindowKey.replace(sliceEnd, key);
        while (true) {
            LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
            try {
                recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
                break;
            } catch (EOFException e) {
                if (recordsBuffer.getNumKeys() == 0) {
                    // Buffer is empty, retry won't help (record is too large for the buffer)
                    throw e;
                }
                // Buffer has data, flush and retry
                flush();
                checkState(
                        recordsBuffer.getNumKeys() == 0,
                        "The recordsBuffer should be empty after flushing.");
            }
        }
    }

    @Override
    public void advanceProgress(long progress) throws Exception {

Review comment (bug):
Do we also need to move the above line
`minSliceEnd = Math.min(sliceEnd, minSliceEnd);`
inside the loop?

In `flush()`, it might be reset to `Long.MAX_VALUE`, so we'll end up with the wrong `minSliceEnd` field

Code after:

    private long minSliceEnd = Long.MAX_VALUE;

    public RecordsWindowBuffer(
            Object operatorOwner,
            MemoryManager memoryManager,
            long memorySize,
            RecordsCombiner combineFunction,
            PagedTypeSerializer<RowData> keySer,
            AbstractRowDataSerializer<RowData> inputSer,
            boolean requiresCopy,
            ZoneId shiftTimeZone) {
        this.combineFunction = combineFunction;
        this.recordsBuffer =
                new WindowBytesMultiMap(
                        operatorOwner, memoryManager, memorySize, keySer, inputSer.getArity());
        this.recordSerializer = inputSer;
        this.reuseWindowKey = new WindowKeySerializer(keySer).createInstance();
        this.requiresCopy = requiresCopy;
        this.shiftTimeZone = shiftTimeZone;
    }

    @Override
    public void addElement(RowData key, long sliceEnd, RowData element) throws Exception {
        reuseWindowKey.replace(sliceEnd, key);
        while (true) {
            LookupInfo<WindowKey, Iterator<RowData>> lookup = recordsBuffer.lookup(reuseWindowKey);
            try {
                recordsBuffer.append(lookup, recordSerializer.toBinaryRow(element));
                // Track the lowest trigger time. If watermark exceeds the trigger time,
                // it means there are some elements in the buffer belong to a window going
                // to be fired, and we need to flush the buffer into state for firing.
                minSliceEnd = Math.min(sliceEnd, minSliceEnd);
                break;
            } catch (EOFException e) {
                if (recordsBuffer.getNumKeys() == 0) {
                    // Buffer is empty, retry won't help (record is too large for the buffer)
                    throw e;
                }
                // Buffer has data, flush and retry
                flush();
                checkState(
                        recordsBuffer.getNumKeys() == 0,
                        "The recordsBuffer should be empty after flushing.");
            }
        }
    }

    @Override
    public void advanceProgress(long progress) throws Exception {
        if (isWindowFired(minSliceEnd, progress, shiftTimeZone)) {