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

Code before:

    @Override
    public void open(OpenContext openContext) throws Exception {
        // Initialize agg functions
        aggFuncs = generatedAggsHandler.newInstance(getRuntimeContext().getUserCodeClassLoader());
        aggFuncs.open(new PerKeyStateDataViewStore(getRuntimeContext()));

        // Initialize output record
        output = new JoinedRowData();

        // Initialize value/row equaliser
        valueEqualiser =
                generatedRecordEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());

        // Initialize sortKey equaliser
        sortKeyEqualiser =
                generatedSortKeyEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());

        // Initialize sort comparator
        sortKeyComparator =
                generatedSortKeyComparator.newInstance(
                        getRuntimeContext().getUserCodeClassLoader());

        StateTtlConfig ttlConfig = createTtlConfig(stateRetentionTime);

        // Initialize state to maintain id counter
        idStateDescriptor = new ValueStateDescriptor<Long>("idState", Long.class);
        idState = getRuntimeContext().getState(idStateDescriptor);

        // Input elements are all binary rows as they came from network
        InternalTypeInfo<RowData> inputRowTypeInfo = InternalTypeInfo.ofFields(inputFieldTypes);
        InternalTypeInfo<RowData> sortKeyRowTypeInfo = InternalTypeInfo.ofFields(this.sortKeyTypes);

        // Initialize state which maintains a sorted list of tuples(sortKey, List of IDs)
        ListTypeInfo<Long> idListTypeInfo = new ListTypeInfo<Long>(Types.LONG);
        ListTypeInfo<Tuple2<RowData, List<Long>>> listTypeInfo =
                new ListTypeInfo<>(new TupleTypeInfo<>(sortKeyRowTypeInfo, idListTypeInfo));
        sortedListStateDescriptor =
                new ValueStateDescriptor<List<Tuple2<RowData, List<Long>>>>(
                        "sortedListState", listTypeInfo);
        sortedListState = getRuntimeContext().getState(sortedListStateDescriptor);

        // Initialize state which maintains the actual row
        valueStateDescriptor =
                new MapStateDescriptor<Long, RowData>(
                        "valueMapState", Types.LONG, inputRowTypeInfo);
        valueMapState = getRuntimeContext().getMapState(valueStateDescriptor);

        // Initialize accumulator state per row
        InternalTypeInfo<RowData> accTypeInfo = InternalTypeInfo.ofFields(accTypes);
        accStateDescriptor =

Review comment (suggestion):
In addition to the fix could we have a validation check when defining a state ttl on NonTime Over Windows, and error in this case, so the user is not led to think that ttl might be valid in this case.

Code after:

        // Initialize output record
        output = new JoinedRowData();

        // Initialize value/row equaliser
        valueEqualiser =
                generatedRecordEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());

        // Initialize sortKey equaliser
        sortKeyEqualiser =
                generatedSortKeyEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());

        // Initialize sort comparator
        sortKeyComparator =
                generatedSortKeyComparator.newInstance(
                        getRuntimeContext().getUserCodeClassLoader());

        // Initialize state to maintain id counter
        idStateDescriptor = new ValueStateDescriptor<Long>("idState", Long.class);
        idState = getRuntimeContext().getState(idStateDescriptor);

        // Input elements are all binary rows as they came from network
        InternalTypeInfo<RowData> inputRowTypeInfo = InternalTypeInfo.ofFields(inputFieldTypes);
        InternalTypeInfo<RowData> sortKeyRowTypeInfo = InternalTypeInfo.ofFields(this.sortKeyTypes);

        // Initialize state which maintains a sorted list of tuples(sortKey, List of IDs)
        ListTypeInfo<Long> idListTypeInfo = new ListTypeInfo<Long>(Types.LONG);
        ListTypeInfo<Tuple2<RowData, List<Long>>> listTypeInfo =
                new ListTypeInfo<>(new TupleTypeInfo<>(sortKeyRowTypeInfo, idListTypeInfo));
        sortedListStateDescriptor =
                new ValueStateDescriptor<List<Tuple2<RowData, List<Long>>>>(
                        "sortedListState", listTypeInfo);
        sortedListState = getRuntimeContext().getState(sortedListStateDescriptor);

        // Initialize state which maintains the actual row
        valueStateDescriptor =
                new MapStateDescriptor<Long, RowData>(
                        "valueMapState", Types.LONG, inputRowTypeInfo);
        valueMapState = getRuntimeContext().getMapState(valueStateDescriptor);

        // Initialize accumulator state per row
        InternalTypeInfo<RowData> accTypeInfo = InternalTypeInfo.ofFields(accTypes);
        accStateDescriptor =
                new MapStateDescriptor<RowData, RowData>(
                        "accMapState", accKeyRowTypeInfo, accTypeInfo);

        accMapState = getRuntimeContext().getMapState(accStateDescriptor);

        initCleanupTimeState("NonTimeUnboundedPrecedingOverCleanupTime");

        // metrics