Package org.apache.flink.table.runtime.operators.over

Class NonTimeRowsUnboundedPrecedingFunction<K>

  • All Implemented Interfaces:
    Serializable, org.apache.flink.api.common.functions.Function, org.apache.flink.api.common.functions.RichFunction
    public class NonTimeRowsUnboundedPrecedingFunction<K>
extends AbstractNonTimeUnboundedPrecedingOver<K>
    The NonTimeRowsUnboundedPrecedingFunction class is a specialized implementation for processing unbounded OVER window aggregations, particularly for non-time-based rows queries in Apache Flink. It maintains strict ordering of rows within partitions and handles the full changelog lifecycle (inserts, updates, deletes).

    Key Components and Assumptions

    Data Structure Design: (1) Maintains a sorted list of tuples containing sort keys and lists of IDs for each key (2) Each incoming row is assigned a unique Long ID (starting from Long.MIN_VALUE) (3) Uses multiple state types to track rows, sort orders, and aggregations

    State Management: (1) idState: Counter for generating unique row IDs (2) sortedListState: Ordered list of sort keys with their associated row IDs (3) valueMapState: Maps IDs to their corresponding input rows (4) accMapState: Maps RowData(ID)s to their accumulated values

    Processing Model: (1) For inserts/updates: Adds rows to the appropriate position based on sort key (2) For deletes: Removes rows by matching both sort key and row content (3) Recalculates aggregates for affected rows and emits the appropriate events (4) Skips redundant events when accumulators haven't changed to reduce network traffic

    Optimization Assumptions: (1) Skip emitting updates when accumulators haven't changed to reduce network traffic (2) Uses state TTL for automatic cleanup of stale data (3) Carefully manages row state to support incremental calculations

    Retraction Handling: (1) Handles retraction mode (DELETE/UPDATE_BEFORE) events properly (2) Supports the proper processing of changelog streams

    Limitations

    Linear search performance: - The current implementation uses a linear search to find the correct position for each sort key. This can be optimized using a binary search for large state sizes.

    State size and performance: - The implementation maintains multiple state types that could grow large with high cardinality data

    Linear recalculation: - When processing updates, all subsequent elements need to be recalculated, which could be inefficient for large windows

    See Also:
    Serialized Form

    • Method Detail

      • insertIntoSortedList

        protected void insertIntoSortedList​(org.apache.flink.table.data.RowData insRow,
                                    org.apache.flink.util.Collector<org.apache.flink.table.data.RowData> out)
                             throws Exception
        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.
        Parameters:
        insRow - The input value.
        out - The collector for returning result values.
        Throws:
        Exception

      • processRemainingElements

        protected void processRemainingElements​(List<org.apache.flink.api.java.tuple.Tuple2<org.apache.flink.table.data.RowData,​List<Long>>> sortedList,
                                        int startPos,
                                        org.apache.flink.table.data.RowData currAcc,
                                        org.apache.flink.util.Collector<org.apache.flink.table.data.RowData> out)
                                 throws Exception
        Process all sort keys starting from startPos location. Calculates the new agg value for all ids and emits the old and new aggregated values for the ids if the newly aggregated value is different from the previously aggregated values.
        Parameters:
        sortedList -
        startPos -
        currAcc -
        out -
        Throws:
        Exception