Repository: apache/flink
File: flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/AbstractNonTimeUnboundedPrecedingOver.java (Java)

Code before:
        // register state-cleanup timer
        registerProcessingCleanupTimer(ctx, ctx.timerService().currentProcessingTime());

        RowKind rowKind = input.getRowKind();

        switch (rowKind) {
            case INSERT:
            case UPDATE_AFTER:
                insertIntoSortedList(input, out);
                break;

            case DELETE:
            case UPDATE_BEFORE:
                removeFromSortedList(input, out);
                break;
        }

        // Reset acc state since we can have out of order inserts in the ordered list
        resetAndCleanupAggFuncs();
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
            throws Exception {
        if (stateCleaningEnabled) {
            cleanupState(idState, valueMapState, accMapState, sortedListState);
            resetAndCleanupAggFuncs();
        }
    }

    private void resetAndCleanupAggFuncs() throws Exception {
        aggFuncs.resetAccumulators();
        aggFuncs.cleanup();
    }

    /**
     * Adds a new element(insRow) to a sortedList. The sortedList contains a list of Tuple(SortKey,
     * List[Long Ids]>). Extracts the inputSortKey from the insRow and compares it with every
     * element in the sortedList. If a sortKey already exists in the sortedList for the input, add
     * the id to the list of ids and update the sortedList, otherwise find the right position in the
     * sortedList and add a new entry in the sortedList. After the insRow is successfully inserted,
     * an INSERT/UPDATE_AFTER event is emitted for the newly inserted element, and for all
     * subsequent elements an UPDATE_BEFORE and UPDATE_AFTER event is emitted based on the previous
     * and newly aggregated values. Some updates are skipped if the previously accumulated value is
     * the same as the newly accumulated value to save on network bandwidth and downstream
     * processing including writing the result to the sink system. Implementation differs for rows
     * vs range. To comply with the sql RANGE syntax, emits the same aggregated value for all
     * elements with the same sort key. To comply with the sql ROWS syntax, emits different
     * aggregated values for all elements with the same sort key.
     *
     * @param insRow The input value.

Review comment (refactor):
I think we should check that the firing timer is actually a clean up timer, even though this operator does not register any other timers. This makes it a bit safer and the intent more clear.
* `KeyedProcessFunctionWithCleanupState.isProcessingTimeTimer()` 
* `KeyedProcessFunctionWithCleanupState.needToCleanupState()`

Code after:
                insertIntoSortedList(input, out);
                break;

            case DELETE:
            case UPDATE_BEFORE:
                removeFromSortedList(input, out);
                break;
        }

        // Reset acc state since we can have out of order inserts in the ordered list
        resetAndCleanupAggFuncs();
    }

    /**
     * Called when the cleanup timer fires.
     *
     * <p>This method clears all associated state for the current key (idState, valueMapState,
     * accMapState, sortedListState) atomically to implement state TTL behavior. Using a single
     * timer ensures consistent cleanup of all related states.
     */
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out)
            throws Exception {
        if (stateCleaningEnabled && isProcessingTimeTimer(ctx) && needToCleanupState(timestamp)) {
            cleanupState(idState, valueMapState, accMapState, sortedListState);
            resetAndCleanupAggFuncs();
        }
    }

    private void resetAndCleanupAggFuncs() throws Exception {
        aggFuncs.resetAccumulators();
        aggFuncs.cleanup();
    }

    /**
     * Adds a new element(insRow) to a sortedList. The sortedList contains a list of Tuple(SortKey,
     * List[Long Ids]>). Extracts the inputSortKey from the insRow and compares it with every
     * element in the sortedList. If a sortKey already exists in the sortedList for the input, add
     * the id to the list of ids and update the sortedList, otherwise find the right position in the
     * sortedList and add a new entry in the sortedList. After the insRow is successfully inserted,
     * an INSERT/UPDATE_AFTER event is emitted for the newly inserted element, and for all
     * subsequent elements an UPDATE_BEFORE and UPDATE_AFTER event is emitted based on the previous
     * and newly aggregated values. Some updates are skipped if the previously accumulated value is
     * the same as the newly accumulated value to save on network bandwidth and downstream
     * processing including writing the result to the sink system. Implementation differs for rows
     * vs range. To comply with the sql RANGE syntax, emits the same aggregated value for all
     * elements with the same sort key. To comply with the sql ROWS syntax, emits different
     * aggregated values for all elements with the same sort key.
     *
     * @param insRow The input value.
     * @param out The collector for returning result values.