python/ray/data/impl/simple_block.py
def aggregate_combined_blocks(
        blocks: List[Block[Tuple[KeyType, AggType]]],
        key: KeyFn,
        aggs: Tuple[AggregateFn],
    ) -> Tuple[Block[Tuple[KeyType, U]], BlockMetadata]:
        """Aggregate sorted, partially combined blocks with the same key range.

        This assumes blocks are already sorted by key in ascending order,
        so we can do merge sort to get all the rows with the same key.

        Args:
            blocks: A list of partially combined and sorted blocks.
            key: The key function that returns the key from the row
                or None for global aggregation.
            aggs: The aggregations to do.

        Returns:
            A block of (k, v_1, ..., v_n) tuples and its metadata where k is
            the groupby key and v_i is the corresponding aggregation result for
            the ith given aggregation.
            If key is None then the k element of tuple is omitted.
        """

        stats = BlockExecStats.builder()
        key_fn = (lambda r: r[0]) if key else (lambda r: 0)

        iter = heapq.merge(
            *[SimpleBlockAccessor(block).iter_rows() for block in blocks], key=key_fn
        )
        next_row = None
        ret = []
        while True:
            try:
                if next_row is None:
                    next_row = next(iter)
                next_key = key_fn(next_row)

                def gen():
                    nonlocal iter
                    nonlocal next_row
                    while key_fn(next_row) == next_key:
                        yield next_row
                        try:
                            next_row = next(iter)
                        except StopIteration:
                            next_row = None
                            break

                first = True
                accumulators = [None] * len(aggs)
                for r in gen():
                    if first:
                        for i in range(len(aggs)):
                            accumulators[i] = r[i + 1] if key else r[i]
                        first = False
                    else:
                        for i in range(len(aggs)):
                            accumulators[i] = aggs[i].merge(
                                accumulators[i], r[i + 1] if key else r[i]
                            )
                if key is None:
                    ret.append(
                        tuple(
                            agg.finalize(accumulator)
                            for agg, accumulator in zip(aggs, accumulators)
                        )
                    )
                else:
                    ret.append(
                        (next_key,)
                        + tuple(
                            agg.finalize(accumulator)
                            for agg, accumulator in zip(aggs, accumulators)
                        )
                    )
            except StopIteration:
                break

        return ret, SimpleBlockAccessor(ret).get_metadata(
            None, exec_stats=stats.build()
        )
Aggregate sorted, partially combined blocks with the same key range.

        This assumes blocks are already sorted by key in ascending order,
        so we can do merge sort to get all the rows with the same key.

        Args:
            blocks: A list of partially combined and sorted blocks.
            key: The key function that returns the key from the row
                or None for global aggregation.
            aggs: The aggregations to do.

        Returns:
            A block of (k, v_1, ..., v_n) tuples and its metadata where k is
            the groupby key and v_i is the corresponding aggregation result for
            the ith given aggregation.
            If key is None then the k element of tuple is omitted.
        
Aggregate sorted, partially combined blocks with the same key range.
This assumes blocks are already sorted by key in ascending order,
so we can do merge sort to get all the rows with the same key.
Aggregate sorted partially combined blocks with the same key range . This assumes blocks are already sorted by key in ascending order so we can do merge sort to get all the rows with the same key .
def aggregate_combined_blocks(
        blocks: List[Block[Tuple[KeyType, AggType]]],
        key: KeyFn,
        aggs: Tuple[AggregateFn],
    ) -> Tuple[Block[Tuple[KeyType, U]], BlockMetadata]:
        stats = BlockExecStats.builder()
        key_fn = (lambda r: r[0]) if key else (lambda r: 0)
        iter = heapq.merge(
            *[SimpleBlockAccessor(block).iter_rows() for block in blocks], key=key_fn
        )
        next_row = None
        ret = []
        while True:
            try:
                if next_row is None:
                    next_row = next(iter)
                next_key = key_fn(next_row)
                def gen():
                    nonlocal iter
                    nonlocal next_row
                    while key_fn(next_row) == next_key:
                        yield next_row
                        try:
                            next_row = next(iter)
                        except StopIteration:
                            next_row = None
                            break
                first = True
                accumulators = [None] * len(aggs)
                for r in gen():
                    if first:
                        for i in range(len(aggs)):
                            accumulators[i] = r[i + 1] if key else r[i]
                        first = False
                    else:
                        for i in range(len(aggs)):
                            accumulators[i] = aggs[i].merge(
                                accumulators[i], r[i + 1] if key else r[i]
                            )
                if key is None:
                    ret.append(
                        tuple(
                            agg.finalize(accumulator)
                            for agg, accumulator in zip(aggs, accumulators)
                        )
                    )
                else:
                    ret.append(
                        (next_key,)
                        + tuple(
                            agg.finalize(accumulator)
                            for agg, accumulator in zip(aggs, accumulators)
                        )
                    )
            except StopIteration:
                break
        return ret, SimpleBlockAccessor(ret).get_metadata(
            None, exec_stats=stats.build()
        )
def aggregate_combined_blocks ( blocks : List [ Block [ Tuple [ KeyType , AggType ] ] ] , key : KeyFn , aggs : Tuple [ AggregateFn ] , ) -> Tuple [ Block [ Tuple [ KeyType , U ] ] , BlockMetadata ] : stats = BlockExecStats . builder ( ) key_fn = ( lambda r : r [ 0 ] ) if key else ( lambda r : 0 ) iter = heapq . merge ( * [ SimpleBlockAccessor ( block ) . iter_rows ( ) for block in blocks ] , key = key_fn ) next_row = None ret = [ ] while True : try : if next_row is None : next_row = next ( iter ) next_key = key_fn ( next_row ) def gen ( ) : nonlocal iter nonlocal next_row while key_fn ( next_row ) == next_key : yield next_row try : next_row = next ( iter ) except StopIteration : next_row = None break first = True accumulators = [ None ] * len ( aggs ) for r in gen ( ) : if first : for i in range ( len ( aggs ) ) : accumulators [ i ] = r [ i + 1 ] if key else r [ i ] first = False else : for i in range ( len ( aggs ) ) : accumulators [ i ] = aggs [ i ] . merge ( accumulators [ i ] , r [ i + 1 ] if key else r [ i ] ) if key is None : ret . append ( tuple ( agg . finalize ( accumulator ) for agg , accumulator in zip ( aggs , accumulators ) ) ) else : ret . append ( ( next_key , ) + tuple ( agg . finalize ( accumulator ) for agg , accumulator in zip ( aggs , accumulators ) ) ) except StopIteration : break return ret , SimpleBlockAccessor ( ret ) . get_metadata ( None , exec_stats = stats . build ( ) )
Aggregate sorted, partially combined blocks with the same key range.
Aggregate sorted partially combined blocks with the same key range .
"""Aggregate sorted, partially combined blocks with the same key range.

        This assumes blocks are already sorted by key in ascending order,
        so we can do merge sort to get all the rows with the same key.

        Args:
            blocks: A list of partially combined and sorted blocks.
            key: The key function that returns the key from the row
                or None for global aggregation.
            aggs: The aggregations to do.

        Returns:
            A block of (k, v_1, ..., v_n) tuples and its metadata where k is
            the groupby key and v_i is the corresponding aggregation result for
            the ith given aggregation.
            If key is None then the k element of tuple is omitted.
        """
A block of (k, v_1, ..., v_n) tuples and its metadata where k is
the groupby key and v_i is the corresponding aggregation result for
the ith given aggregation.
If key is None then the k element of tuple is omitted.
A block of ( k v_1 ... v_n ) tuples and its metadata where k is the groupby key and v_i is the corresponding aggregation result for the ith given aggregation . If key is None then the k element of tuple is omitted .
blocks
List[Block[Tuple[KeyType, AggType]]]
A list of partially combined and sorted blocks.
A list of partially combined and sorted blocks .
key
KeyFn
The key function that returns the key from the row
or None for global aggregation.
The key function that returns the key from the row or None for global aggregation .
aggs
Tuple[AggregateFn]
The aggregations to do.
The aggregations to do .