Function processGroupBy

packages/db/src/query/compiler/group-by.ts:127–474

(
  pipeline: NamespacedAndKeyedStream,
  groupByClause: GroupBy,
  havingClauses?: Array<Having>,
  selectClause?: Select,
  fnHavingClauses?: Array<(row: any) => any>,
  aggregateCollectionId?: string,
  mainSource?: string,
)

 * Works with the new $selected structure from early SELECT processing
 */
export function processGroupBy(
  pipeline: NamespacedAndKeyedStream,
  groupByClause: GroupBy,
  havingClauses?: Array<Having>,
  selectClause?: Select,
  fnHavingClauses?: Array<(row: any) => any>,
  aggregateCollectionId?: string,
  mainSource?: string,
): NamespacedAndKeyedStream {
  const virtualAggregates: Record<string, any> = {
    [VIRTUAL_SYNCED_KEY]: {
      preMap: ([, row]: [string, NamespacedRow]) =>
        getRowVirtualMetadata(row).synced,
      reduce: (values: Array<[boolean, number]>) => {
        for (const [isSynced, multiplicity] of values) {
          if (!isSynced && multiplicity > 0) {
            return false
          }
        }
        return true
      },
    },
    [VIRTUAL_HAS_LOCAL_KEY]: {
      preMap: ([, row]: [string, NamespacedRow]) =>
        getRowVirtualMetadata(row).hasLocal,
      reduce: (values: Array<[boolean, number]>) => {
        for (const [isLocal, multiplicity] of values) {
          if (isLocal && multiplicity > 0) {
            return true
          }
        }
        return false
      },
    },
  }

  // Handle empty GROUP BY (single-group aggregation)
  if (groupByClause.length === 0) {
    // For single-group aggregation, create a single group with all data
    const aggregates: Record<string, any> = virtualAggregates

    // Expressions that wrap aggregates (e.g. coalesce(count(...), 0)).
    // Keys are the original SELECT aliases; values are pre-compiled evaluators
    // over the transformed (aggregate-free) expression.
    const wrappedAggExprs: Record<string, (data: any) => any> = {}
    const aggCounter = { value: 0 }

    if (selectClause) {
      // Scan the SELECT clause for aggregate functions
      for (const [alias, expr] of Object.entries(selectClause)) {
        if (expr.type === `agg`) {
          aggregates[alias] = getAggregateFunction(expr)
        } else if (containsAggregate(expr)) {
          const { transformed, extracted } = extractAndReplaceAggregates(
            expr as SelectValueExpression,
            aggCounter,
          )
          for (const [syntheticAlias, aggExpr] of Object.entries(extracted)) {