diff --git a/handwritten/firestore/api-report/firestore.api.md b/handwritten/firestore/api-report/firestore.api.md index 46360a5331b..e7219d0950c 100644 --- a/handwritten/firestore/api-report/firestore.api.md +++ b/handwritten/firestore/api-report/firestore.api.md @@ -68,6 +68,10 @@ class AggregateFunction implements AggregateFunction, HasUserData { _createdFromLiteral: boolean; // (undocumented) expressionType: firestore.Pipelines.ExpressionType; + over(frame: firestore.OneOf<{ + documents: firestore.Pipelines.DocumentWindowFrame; + range: firestore.Pipelines.RangeWindowFrame; + }>): WindowFunction; // (undocumented) _protoValueType: "ProtoValue"; // Warning: (tsdoc-undefined-tag) The TSDoc tag "@private" is not defined in this configuration @@ -170,6 +174,23 @@ class AliasedExpression implements firestore.Pipelines.Selectable, HasUserData { _validateUserData(ignoreUndefinedProperties: boolean): void; } +// @public +class AliasedWindowFunction implements firestore.Pipelines.AliasedWindowFunction, HasUserData { + constructor(_windowFunction: WindowFunction, _alias: string); + // (undocumented) + readonly _alias: string; + // Warning: (tsdoc-undefined-tag) The TSDoc tag "@private" is not defined in this configuration + // + // @internal + _createdFromLiteral: boolean; + // Warning: (tsdoc-undefined-tag) The TSDoc tag "@private" is not defined in this configuration + // + // @internal (undocumented) + _validateUserData(ignoreUndefinedProperties: boolean): void; + // (undocumented) + readonly _windowFunction: WindowFunction; +} + // @public function and(first: BooleanExpression, second: BooleanExpression, ...more: BooleanExpression[]): BooleanExpression; @@ -1998,6 +2019,8 @@ class Pipeline implements firestore.Pipelines.Pipeline { addFields(field: firestore.Pipelines.Selectable, ...additionalFields: firestore.Pipelines.Selectable[]): Pipeline; // Warning: (tsdoc-escape-right-brace) The "}" character should be escaped using a backslash to avoid confusion with a TSDoc inline tag addFields(options: firestore.Pipelines.AddFieldsStageOptions): Pipeline; + addWindowFields(window: firestore.Pipelines.WindowSpec, field: firestore.Pipelines.AliasedAggregate | firestore.Pipelines.AliasedWindowFunction, ...additionalFields: Array): Pipeline; + addWindowFields(options: firestore.Pipelines.AddWindowFieldsStageOptions): Pipeline; aggregate(accumulator: firestore.Pipelines.AliasedAggregate, ...additionalAccumulators: firestore.Pipelines.AliasedAggregate[]): Pipeline; aggregate(options: firestore.Pipelines.AggregateStageOptions): Pipeline; define(aliasedExpression: firestore.Pipelines.AliasedExpression, ...additionalExpressions: firestore.Pipelines.AliasedExpression[]): Pipeline; @@ -2274,7 +2297,10 @@ declare namespace Pipelines { coalesce, documentMatches, score, - geoDistance + geoDistance, + WindowFunction, + AliasedWindowFunction, + rank } } export { Pipelines } @@ -2616,6 +2642,9 @@ export class QuerySnapshot): WindowFunction; + // (undocumented) + _protoValueType: "ProtoValue"; + // Warning: (tsdoc-undefined-tag) The TSDoc tag "@private" is not defined in this configuration + // + // @internal (undocumented) + _toProto(serializer: Serializer): api.IValue; + // Warning: (tsdoc-undefined-tag) The TSDoc tag "@private" is not defined in this configuration + // + // @internal (undocumented) + _validateUserData(ignoreUndefinedProperties: boolean): void; +} + // Warning: (tsdoc-undefined-tag) The TSDoc tag "@class" is not defined in this configuration // // @public diff --git a/handwritten/firestore/dev/src/pipelines/expression.ts b/handwritten/firestore/dev/src/pipelines/expression.ts index fee0a107e96..b97f474d133 100644 --- a/handwritten/firestore/dev/src/pipelines/expression.ts +++ b/handwritten/firestore/dev/src/pipelines/expression.ts @@ -21,6 +21,7 @@ import {VectorValue} from '../field-value'; import {FieldPath} from '../path'; import { fieldOrExpression, + isExpr, isFirestoreValue, isString, toField, @@ -3374,6 +3375,41 @@ export class AggregateFunction implements AggregateFunction, HasUserData { return new AliasedAggregate(this, name); } + /** + * Evaluates this aggregate function over a specific window frame, rather than + * over the window frame defined by the enclosing + * `Pipeline.addWindowFields()` stage. + * + * @example + * ```typescript + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { sort: ascending('date') }, + * // Uses the stage's default frame. + * sum('amount').as('runningTotal'), + * // Overrides the stage frame with a 3 document moving window. + * average('amount') + * .over({ documents: { preceding: 1, following: 1 } }) + * .as('movingAverage') + * ); + * ``` + * + * @param frame - The `documents` or `range` frame to evaluate this aggregate over. + * @returns A new `WindowFunction`. + */ + over( + frame: firestore.OneOf<{ + documents: firestore.Pipelines.DocumentWindowFrame; + range: firestore.Pipelines.RangeWindowFrame; + }>, + ): WindowFunction { + return new WindowFunction( + this.name, + this.params, + new WindowSpecInternal(frame), + ); + } + /** * @private * @internal @@ -11214,3 +11250,330 @@ export class Ordering implements HasUserData { (this.expr as Expression)._validateUserData(ignoreUndefinedProperties); } } + +/** + * A bound of a window frame. Either a numeric offset, one of the sentinel + * strings `'current'` or `'unbounded'`, or an {@link Expression} that evaluates + * to one of those values. + * + * @internal + * @private + */ +export type WindowFrameBound = + number | 'current' | 'unbounded' | firestore.Pipelines.Expression; + +/** + * The normalized internal representation of a `WindowSpec`. + * + * This mirrors the wire format one-to-one: `partition` and `sort` are + * normalized to expressions and orderings, while `documents` and `range` are + * carried through verbatim. Combinations that the backend rejects (such as + * specifying both `documents` and `range`) are deliberately representable + * here so that the backend, rather than the SDK, owns validation. + * + * The same serialization is used for the first argument of the + * `add_window_fields` stage and for the second argument of an accumulator + * level `over()` clause. + * + * @internal + * @private + */ +export class WindowSpecInternal implements HasUserData { + readonly partition: Expression[]; + readonly sort: Ordering[]; + readonly documents?: firestore.Pipelines.DocumentWindowFrame; + readonly range?: firestore.Pipelines.RangeWindowFrame; + + constructor(window: firestore.Pipelines.WindowSpec) { + const sort = window.sort; + this.partition = (window.partition ?? []).map(fieldOrExpression); + this.sort = + sort === undefined + ? [] + : Array.isArray(sort) + ? (sort as Ordering[]) + : [sort as Ordering]; + this.documents = window.documents; + this.range = window.range; + } + + /** + * @private + * @internal + */ + _toProto(serializer: Serializer): api.IValue { + const fields: Record = {}; + + // Empty `partition` and `sort` keys are omitted rather than sent as empty + // arrays, matching what the backend emits. + if (this.partition.length > 0) { + fields.partition = { + arrayValue: {values: this.partition.map(p => p._toProto(serializer))}, + }; + } + if (this.sort.length > 0) { + fields.sort = { + arrayValue: {values: this.sort.map(o => o._toProto(serializer))}, + }; + } + if (this.documents !== undefined) { + fields.documents = this.frameToProto(serializer, this.documents); + } + if (this.range !== undefined) { + fields.range = this.frameToProto(serializer, this.range); + } + + return {mapValue: {fields}}; + } + + /** + * @private + * @internal + */ + _validateUserData(ignoreUndefinedProperties: boolean): void { + this.partition.forEach(p => p._validateUserData(ignoreUndefinedProperties)); + this.sort.forEach(o => o._validateUserData(ignoreUndefinedProperties)); + + for (const frame of [this.documents, this.range]) { + if (frame === undefined) { + continue; + } + if (isExpr(frame.preceding)) { + (frame.preceding as Expression)._validateUserData( + ignoreUndefinedProperties, + ); + } + if (isExpr(frame.following)) { + (frame.following as Expression)._validateUserData( + ignoreUndefinedProperties, + ); + } + const unit = (frame as firestore.Pipelines.RangeWindowFrame).unit; + if (unit !== undefined && isExpr(unit)) { + (unit as Expression)._validateUserData(ignoreUndefinedProperties); + } + } + } + + /** + * Serializes a window frame, i.e. the value keyed by `documents` or `range` + * in a window specification. + */ + private frameToProto( + serializer: Serializer, + frame: + | firestore.Pipelines.DocumentWindowFrame + | firestore.Pipelines.RangeWindowFrame, + ): api.IValue { + const fields: Record = {}; + + // Bounds are only sent when supplied. The backend defaults an absent bound + // to the current document. + if (frame.preceding !== undefined) { + fields.preceding = this.frameBoundToProto( + serializer, + frame.preceding, + 'preceding', + ); + } + if (frame.following !== undefined) { + fields.following = this.frameBoundToProto( + serializer, + frame.following, + 'following', + ); + } + + // `unit` is only meaningful on a `range` frame, but it is encoded wherever + // it is supplied so that the backend reports the error. + const unit = (frame as firestore.Pipelines.RangeWindowFrame).unit; + if (unit !== undefined) { + fields.unit = isString(unit) + ? serializer.encodeValue(unit)! + : (unit as Expression)._toProto(serializer); + } + + return {mapValue: {fields}}; + } + + /** + * Serializes a single window frame bound. + */ + private frameBoundToProto( + serializer: Serializer, + bound: WindowFrameBound, + boundName: 'preceding' | 'following', + ): api.IValue { + if (typeof bound === 'number') { + return serializer.encodeValue(bound)!; + } + // Sentinel strings are passed through as-is rather than being checked + // against a known set, so that the backend can add new sentinels without an + // SDK update. Unknown values are rejected by the backend. + if (isString(bound)) { + return serializer.encodeValue(bound)!; + } + if (isExpr(bound)) { + return (bound as Expression)._toProto(serializer); + } + // Validated client side only because there is no way to encode a value of + // an unknown type into a proto `Value`. + throw new Error( + `Invalid window frame '${boundName}' bound. Expected a number, ` + + `'current', 'unbounded', or an Expression.`, + ); + } + + _protoValueType = 'ProtoValue' as const; +} + +/** + * A function that is evaluated over a window frame of documents, as part of an + * `Pipeline.addWindowFields()` stage. + * + * A `WindowFunction` is created either by one of the dedicated window function + * factories (such as {@link rank}), or by calling + * `over()` on an {@link AggregateFunction}. + */ +export class WindowFunction + implements firestore.Pipelines.WindowFunction, HasUserData +{ + expressionType: firestore.Pipelines.ExpressionType = 'WindowFunction'; + + /** + * @internal + * @private + * Indicates if this expression was created from a literal value passed + * by the caller. + */ + _createdFromLiteral = false; + + constructor( + private name: string, + private params: Expression[] = [], + /** + * The accumulator level window this function is evaluated over. When + * `undefined`, the window of the enclosing stage is used. + * @internal + */ + private window?: WindowSpecInternal, + ) {} + + /** + * Evaluates this window function over a specific window frame, rather than + * over the window frame defined by the enclosing + * `Pipeline.addWindowFields()` stage. + * + * @param frame - The `documents` or `range` frame to evaluate this function over. + * @returns A new `WindowFunction`. + */ + over( + frame: firestore.OneOf<{ + documents: firestore.Pipelines.DocumentWindowFrame; + range: firestore.Pipelines.RangeWindowFrame; + }>, + ): WindowFunction { + return new WindowFunction( + this.name, + this.params, + new WindowSpecInternal(frame), + ); + } + + /** + * Assigns an alias to this `WindowFunction`. The alias specifies the name that + * the computed value will have in the output document. + * + * @param name - The alias to assign to this `WindowFunction`. + * @returns A new `AliasedWindowFunction`. + */ + as(name: string): AliasedWindowFunction { + return new AliasedWindowFunction(this, name); + } + + /** + * @private + * @internal + */ + _toProto(serializer: Serializer): api.IValue { + const functionValue: api.IValue = { + functionValue: { + name: this.name, + args: this.params.map(p => p._toProto(serializer)), + }, + }; + + if (this.window === undefined) { + return functionValue; + } + + return { + functionValue: { + name: 'over', + args: [functionValue, this.window._toProto(serializer)], + }, + }; + } + + _protoValueType = 'ProtoValue' as const; + + /** + * @private + * @internal + */ + _validateUserData(ignoreUndefinedProperties: boolean): void { + this.params.forEach(expr => { + return expr._validateUserData(ignoreUndefinedProperties); + }); + this.window?._validateUserData(ignoreUndefinedProperties); + } +} + +/** + * A `WindowFunction` with an alias. + */ +export class AliasedWindowFunction + implements firestore.Pipelines.AliasedWindowFunction, HasUserData +{ + constructor( + readonly _windowFunction: WindowFunction, + readonly _alias: string, + ) {} + + /** + * @internal + * @private + * Indicates if this expression was created from a literal value passed + * by the caller. + */ + _createdFromLiteral = false; + + /** + * @private + * @internal + */ + _validateUserData(ignoreUndefinedProperties: boolean): void { + this._windowFunction._validateUserData(ignoreUndefinedProperties); + } +} + +/** + * Creates a window function that computes the rank of the current document + * within its window frame. Documents that compare equal in the window sort + * order receive the same rank, and the next rank is offset by the number of + * tied documents. + * + * @example + * ```typescript + * firestore.pipeline().collection("employees") + * .addWindowFields( + * { partition: ['department'], sort: descending('salary') }, + * rank().as('salaryRank') + * ); + * ``` + * + * @returns A new `WindowFunction`. + */ +export function rank(): WindowFunction { + return new WindowFunction('rank', []); +} diff --git a/handwritten/firestore/dev/src/pipelines/index.ts b/handwritten/firestore/dev/src/pipelines/index.ts index 2a0ab092bf7..dc22992a122 100644 --- a/handwritten/firestore/dev/src/pipelines/index.ts +++ b/handwritten/firestore/dev/src/pipelines/index.ts @@ -172,5 +172,8 @@ export { documentMatches, score, geoDistance, + WindowFunction, + AliasedWindowFunction, + rank, // TODO(new-expression): Add new expression exports above this line } from './expression'; diff --git a/handwritten/firestore/dev/src/pipelines/pipeline-util.ts b/handwritten/firestore/dev/src/pipelines/pipeline-util.ts index ed820eadfb8..c35e1223ed9 100644 --- a/handwritten/firestore/dev/src/pipelines/pipeline-util.ts +++ b/handwritten/firestore/dev/src/pipelines/pipeline-util.ts @@ -65,6 +65,8 @@ import { lessThan, Field, AggregateFunction, + AliasedWindowFunction, + WindowFunction, pipelineValue, AliasedExpression, } from './expression'; @@ -602,6 +604,24 @@ export function isAliasedAggregate( ); } +/** + * Returns `true` if the given value is an `AliasedWindowFunction`. + * + * @internal + * @private + */ +export function isAliasedWindowFunction( + val: unknown, +): val is firestore.Pipelines.AliasedWindowFunction { + const candidate = val as firestore.Pipelines.AliasedWindowFunction; + return ( + candidate !== undefined && + candidate !== null && + isString(candidate._alias) && + candidate._windowFunction instanceof WindowFunction + ); +} + export function isExpr(val: unknown): val is firestore.Pipelines.Expression { return val instanceof Expression; } @@ -771,22 +791,70 @@ export function selectablesToObject( return result; } +/** + * Converts a list of aliased aggregate functions, and/or aliased window + * functions, into a map of output field name to the function that computes it. + */ export function aliasedAggregateToMap( - aliasedAggregatees: firestore.Pipelines.AliasedAggregate[], -): Map { - return aliasedAggregatees.reduce( + aliasedAggregates: firestore.Pipelines.AliasedAggregate[], +): Map; +export function aliasedAggregateToMap( + aliasedAggregates: Array< + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction + >, +): Map; +export function aliasedAggregateToMap( + aliasedAggregates: Array< + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction + >, +): Map { + return aliasedAggregates.reduce( ( - map: Map, - selectable: firestore.Pipelines.AliasedAggregate, + map: Map, + aliased: + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction, ) => { - if (map.get(selectable._alias) !== undefined) { - throw new Error(`Duplicate alias or field '${selectable._alias}'`); + const alias = aliased._alias; + // Validated client side because a duplicate alias cannot be encoded: the + // second entry would silently overwrite the first in the map, and the + // backend would never see it. + if (map.get(alias) !== undefined) { + throw new Error(`Duplicate alias or field '${alias}'`); } - map.set(selectable._alias, selectable._aggregate as AggregateFunction); + map.set( + alias, + isAliasedWindowFunction(aliased) + ? (aliased._windowFunction as WindowFunction) + : (aliased._aggregate as AggregateFunction), + ); return map; }, - new Map() as Map, + new Map() as Map, + ); +} + +/** + * Discriminates between the two `Pipeline.addWindowFields()` overloads. + * + * @private + * @internal + */ +export function isAddWindowFieldsStageOptions( + value: + | firestore.Pipelines.WindowSpec + | firestore.Pipelines.AddWindowFieldsStageOptions, +): value is firestore.Pipelines.AddWindowFieldsStageOptions { + const candidate = value as firestore.Pipelines.AddWindowFieldsStageOptions; + return ( + typeof candidate === 'object' && + candidate !== null && + typeof candidate.window === 'object' && + candidate.window !== null && + Array.isArray(candidate.fields) ); } diff --git a/handwritten/firestore/dev/src/pipelines/pipelines.ts b/handwritten/firestore/dev/src/pipelines/pipelines.ts index 48f0f8077d8..de92ad23bac 100644 --- a/handwritten/firestore/dev/src/pipelines/pipelines.ts +++ b/handwritten/firestore/dev/src/pipelines/pipelines.ts @@ -27,6 +27,7 @@ import { ExecutionUtil, aliasedAggregateToMap, fieldOrExpression, + isAddWindowFieldsStageOptions, isAliasedAggregate, isAliasedExpr, isBooleanExpr, @@ -60,6 +61,7 @@ import { Field, BooleanExpression, Ordering, + WindowSpecInternal, constant, _mapValue, field, @@ -68,6 +70,7 @@ import { } from './expression'; import { AddFields, + AddWindowFields, Aggregate, CollectionSource, CollectionGroupSource, @@ -88,6 +91,7 @@ import { Union, Unnest, DeleteStage, + InternalAddWindowFieldsStageOptions, InternalWhereStageOptions, InternalOffsetStageOptions, InternalLimitStageOptions, @@ -521,6 +525,154 @@ export class Pipeline implements firestore.Pipelines.Pipeline { return this._addStage(new AddFields(internalOptions)); } + /** + * Adds window function results to the output documents of the pipeline. + * + * Window functions evaluate expressions over a subset of documents (a "window frame") relative to the + * current document. + * + * @example + * ```typescript + * // 1. Unsorted partition/group aggregation (evaluates over the entire group) + * firestore.pipeline().collection("employees") + * .addWindowFields( + * { + * partition: ['department'] + * }, + * average(field('salary')).as('departmentAverageSalary') + * ); + * + * // 2. Document-based moving average with explicit boundaries + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { + * sort: ascending('date'), + * documents: { preceding: 1, following: 1 } + * }, + * average(field('amount')).as('movingAverageAmount') + * ); + * + * // 3. Document-based running total ('unbounded' preceding to 'current' document position, excluding later ties). + * // Note: Offsets are physical document counts, so no time unit is required or used even when sorting on 'date'. + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { + * sort: ascending('date'), + * documents: { preceding: 'unbounded', following: 'current' } + * }, + * sum(field('amount')).as('runningTotal') + * ); + * + * // 4. Range-based running average using default boundaries ('unbounded' preceding to 'current', including all peers with tied sort values) + * firestore.pipeline().collection("products") + * .addWindowFields( + * { + * sort: ascending('price'), + * range: { preceding: 'unbounded', following: 'current' } + * }, + * average(field('rating')).as('cumulativeAvgRating') + * ); + * + * // 5. Range-based date/time window with a time unit (cumulative sales over the last 30 days) + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { + * sort: ascending('date'), + * range: { preceding: 30, following: 'current', unit: 'day' } + * }, + * sum(field('amount')).as('thirtyDayCumulativeSales') + * ); + * ``` + * + * @param window - The specification defining how documents are partitioned, ordered, and bounded in the window frame. + * @param field - The first window field to add, specified as an `AliasedAggregate` or `AliasedWindowFunction`. + * @param additionalFields - Optional additional window fields to add to the documents. + * @returns A new Pipeline object with this stage appended to the stage list. + */ + addWindowFields( + window: firestore.Pipelines.WindowSpec, + field: + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction, + ...additionalFields: Array< + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction + > + ): Pipeline; + + /** + * Adds window function results to the output documents of the pipeline using options. + * + * @example + * ```typescript + * // 1. Unsorted partition/group aggregation using options + * firestore.pipeline().collection("employees") + * .addWindowFields({ + * window: { + * partition: ['department'] + * }, + * fields: [ + * average(field('salary')).as('departmentAverageSalary') + * ] + * }); + * + * // 2. Document-based moving average using options + * firestore.pipeline().collection("sales") + * .addWindowFields({ + * window: { + * sort: ascending('date'), + * documents: { preceding: 1, following: 1 } + * }, + * fields: [ + * average(field('amount')).as('movingAverageAmount') + * ] + * }); + * ``` + * + * @param options - An object specifying the window frame configuration and the fields to add. + * @returns A new Pipeline object with this stage appended to the stage list. + */ + addWindowFields( + options: firestore.Pipelines.AddWindowFieldsStageOptions, + ): Pipeline; + addWindowFields( + windowOrOptions: + | firestore.Pipelines.WindowSpec + | firestore.Pipelines.AddWindowFieldsStageOptions, + field?: + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction, + ...additionalFields: Array< + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction + > + ): Pipeline { + // Process argument union(s) from method overloads + const usesOptions = isAddWindowFieldsStageOptions(windowOrOptions); + // The whole options object is forwarded, including `window` and `fields`. + // That is safe because `OptionsUtil` only serializes keys declared in the + // stage's option definitions, and this stage declares none. + const options: firestore.Pipelines.StageOptions = usesOptions + ? windowOrOptions + : {}; + const window: firestore.Pipelines.WindowSpec = usesOptions + ? windowOrOptions.window + : windowOrOptions; + const fields: Array< + | firestore.Pipelines.AliasedAggregate + | firestore.Pipelines.AliasedWindowFunction + > = usesOptions ? windowOrOptions.fields : [field!, ...additionalFields]; + + const internalOptions: InternalAddWindowFieldsStageOptions = { + ...options, + window: new WindowSpecInternal(window), + fields: aliasedAggregateToMap(fields), + }; + + // Add stage to the pipeline + return this._addStage(new AddWindowFields(internalOptions)); + } + /** * Remove fields from outputs of previous stages. * diff --git a/handwritten/firestore/dev/src/pipelines/stage.ts b/handwritten/firestore/dev/src/pipelines/stage.ts index 521bc97205e..8afcfe607b3 100644 --- a/handwritten/firestore/dev/src/pipelines/stage.ts +++ b/handwritten/firestore/dev/src/pipelines/stage.ts @@ -30,6 +30,8 @@ import { field, MapValue, Ordering, + WindowFunction, + WindowSpecInternal, } from './expression'; import {OptionsUtil} from './options-util'; import {CollectionReference} from '../reference/collection-reference'; @@ -695,6 +697,55 @@ export class AddFields implements Stage { } } +/** + * Internal options for AddWindowFields stage. + */ +export type InternalAddWindowFieldsStageOptions = Omit< + firestore.Pipelines.AddWindowFieldsStageOptions, + 'window' | 'fields' +> & { + window: WindowSpecInternal; + fields: Map; +}; + +/** + * Implementation of the `add_window_fields` stage. + * + * Serializes to a stage with exactly two arguments: + * 1. The `window_spec` map, containing the optional `partition` array, the + * optional `sort` array, and at most one of the `documents` or `range` + * frame maps. + * 2. The `fields` map, mapping each output field name to either a bare + * aggregate function value (stage level or default framing), or an + * `over(aggregate, frame)` function value (accumulator level framing). + */ +export class AddWindowFields implements Stage { + name = 'add_window_fields'; + readonly optionsUtil = new OptionsUtil({}); + + constructor(private options: InternalAddWindowFieldsStageOptions) {} + + _toProto(serializer: Serializer): api.Pipeline.IStage { + return { + name: this.name, + args: [ + this.options.window._toProto(serializer), + serializer.encodeValue(this.options.fields)!, + ], + options: this.optionsUtil.getOptionsProto( + serializer, + this.options, + this.options.rawOptions, + ), + }; + } + + _validateUserData(ignoreUndefinedProperties: boolean): void { + this.options.window._validateUserData(ignoreUndefinedProperties); + validateUserDataHelper(this.options.fields, ignoreUndefinedProperties); + } +} + /** * Internal options for Sort stage. */ diff --git a/handwritten/firestore/dev/system-test/pipeline.ts b/handwritten/firestore/dev/system-test/pipeline.ts index d3ae823f929..9e8e476504c 100644 --- a/handwritten/firestore/dev/system-test/pipeline.ts +++ b/handwritten/firestore/dev/system-test/pipeline.ts @@ -183,6 +183,7 @@ import {afterEach, describe, it} from 'mocha'; import '../test/util/mocha_extensions'; import {verifyInstance, isRest} from '../test/util/helpers'; import {getTestDb, getTestRoot} from './firestore'; +import {expectResults} from './pipeline_helpers'; import {Firestore as InternalFirestore} from '../src'; import {ServiceError} from 'google-gax'; @@ -209,26 +210,6 @@ async function testCollectionWithDocs( return targetCol; } -function expectResults(result: PipelineSnapshot, ...docs: string[]): void; -function expectResults(result: PipelineSnapshot, ...data: DocumentData[]): void; -function expectResults( - result: PipelineSnapshot, - ...data: DocumentData[] | string[] -): void { - if (data.length > 0) { - if (typeof data[0] === 'string') { - const actualIds = result.results.map(result => result.id); - expect(actualIds).to.deep.equal(data); - } else { - result.results.forEach(r => { - expect(r.data()).to.deep.equal(data.shift()); - }); - } - } else { - expect(result.results.length).to.equal(data.length); - } -} - describe.skipClassic('Pipeline class', () => { let firestore: Firestore; let randomCol: CollectionReference; diff --git a/handwritten/firestore/dev/system-test/pipeline_helpers.ts b/handwritten/firestore/dev/system-test/pipeline_helpers.ts new file mode 100644 index 00000000000..d979405e74c --- /dev/null +++ b/handwritten/firestore/dev/system-test/pipeline_helpers.ts @@ -0,0 +1,44 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {DocumentData} from '@google-cloud/firestore'; +import {expect} from 'chai'; + +import {PipelineSnapshot} from '../src/pipelines'; + +export function expectResults( + result: PipelineSnapshot, + ...docs: string[] +): void; +export function expectResults( + result: PipelineSnapshot, + ...data: DocumentData[] +): void; +export function expectResults( + result: PipelineSnapshot, + ...data: DocumentData[] | string[] +): void { + if (data.length > 0) { + if (typeof data[0] === 'string') { + const actualIds = result.results.map(r => r.id); + expect(actualIds).to.deep.equal(data); + } else { + result.results.forEach(r => { + expect(r.data()).to.deep.equal(data.shift()); + }); + } + } else { + expect(result.results.length).to.equal(data.length); + } +} diff --git a/handwritten/firestore/dev/system-test/pipeline_window.ts b/handwritten/firestore/dev/system-test/pipeline_window.ts new file mode 100644 index 00000000000..4b8b3e5823e --- /dev/null +++ b/handwritten/firestore/dev/system-test/pipeline_window.ts @@ -0,0 +1,1767 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import {DocumentData} from '@google-cloud/firestore'; +import {expect, use} from 'chai'; +import * as chaiAsPromised from 'chai-as-promised'; +import {afterEach, before, describe, it} from 'mocha'; + +import {CollectionReference, Firestore, Timestamp} from '../src'; +import { + arrayAgg, + arrayAggDistinct, + ascending, + average, + constant, + count, + countAll, + countDistinct, + countIf, + descending, + field, + first, + last, + maximum, + minimum, + multiply, + Pipeline, + PipelineSnapshot, + rank, + sum, + toUpper, +} from '../src/pipelines'; +import {verifyInstance} from '../test/util/helpers'; +import '../test/util/mocha_extensions'; +import {getTestRoot} from './firestore'; +import {expectResults} from './pipeline_helpers'; + +use(chaiAsPromised); + +/** + * Test data for the window function suites below. + * + * The sales are deliberately constructed so that: + * - `product` and `region` give overlapping, non-nested partitions, + * - `salesPrice` has ties (30 and 60 both appear twice), which exercises + * `range` framing and ranking peers, and + * - `date` is strictly increasing and has an 11 day gap before the final sale, + * which exercises time based `range` framing. + */ +const windowTestDocs: {[id: string]: DocumentData} = { + sale1: { + product: 'phone', + region: 'east', + salesPrice: 12, + quantity: 1, + date: Timestamp.fromDate(new Date('2026-07-01T12:00:00Z')), + }, + sale2: { + product: 'phone', + region: 'west', + salesPrice: 30, + quantity: 2, + date: Timestamp.fromDate(new Date('2026-07-02T12:00:00Z')), + }, + sale3: { + product: 'tablet', + region: 'east', + salesPrice: 30, + quantity: 3, + date: Timestamp.fromDate(new Date('2026-07-02T18:00:00Z')), + }, + sale4: { + product: 'tablet', + region: 'west', + salesPrice: 60, + quantity: 4, + date: Timestamp.fromDate(new Date('2026-07-04T12:00:00Z')), + }, + sale5: { + product: 'tablet', + region: 'east', + salesPrice: 60, + quantity: 5, + date: Timestamp.fromDate(new Date('2026-07-15T12:00:00Z')), + }, +}; + +async function populateWindowDocs( + targetCol: CollectionReference, +): Promise { + for (const [id, doc] of Object.entries(windowTestDocs)) { + await targetCol.doc(id).set(doc); + } +} + +function execute(ppl: Pipeline): Promise { + return ppl.execute(); +} + +describe.skipClassic('Pipeline window functions', () => { + let firestore: Firestore; + let productSales: CollectionReference; + + before(async () => { + productSales = getTestRoot(); + await populateWindowDocs(productSales); + firestore = productSales.firestore; + }); + + afterEach(() => verifyInstance(firestore)); + + it('computes running aggregates over a documents frame partitioned by a field and an expression and sorted by multiple keys', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + partition: ['product', toUpper('region')], + sort: [ascending('date'), descending('salesPrice')], + documents: {preceding: 'unbounded', following: constant(0)}, + }, + sum('salesPrice').as('stats.runningTotal'), + arrayAgg('quantity').as('quantities'), + countAll().as('count'), + ) + .sort(ascending('date')) + .select('product', 'stats', 'quantities', 'count'), + ); + + expectResults( + snapshot, + { + product: 'phone', + stats: {runningTotal: 12}, + quantities: [1], + count: 1, + }, + { + product: 'phone', + stats: {runningTotal: 30}, + quantities: [2], + count: 1, + }, + { + product: 'tablet', + stats: {runningTotal: 30}, + quantities: [3], + count: 1, + }, + { + product: 'tablet', + stats: {runningTotal: 60}, + quantities: [4], + count: 1, + }, + { + product: 'tablet', + stats: {runningTotal: 90}, + quantities: [3, 5], + count: 2, + }, + ); + }); + + it('counts documents in a time unit range frame using the options overload', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales.path) + .addWindowFields({ + window: { + sort: ascending('date'), + range: { + preceding: 12, + following: 'current', + unit: 'hour', + }, + }, + fields: [count('quantity').as('windowCount')], + }) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 1}, + {product: 'phone', windowCount: 1}, + {product: 'tablet', windowCount: 2}, + {product: 'tablet', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + ); + }); + + it('computes look ahead and nearby aggregates with accumulator level frames', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {sort: ascending('quantity')}, + average('salesPrice') + .over({documents: {preceding: -1, following: 2}}) + .as('lookAheadAverage'), + sum('salesPrice') + .over({range: {preceding: 1.5, following: 0.5}}) + .as('nearbyTotal'), + ) + .sort(ascending('date')) + .select('product', 'lookAheadAverage', 'nearbyTotal'), + ); + + expectResults( + snapshot, + {product: 'phone', lookAheadAverage: 30, nearbyTotal: 12}, + {product: 'phone', lookAheadAverage: 45, nearbyTotal: 42}, + {product: 'tablet', lookAheadAverage: 60, nearbyTotal: 60}, + {product: 'tablet', lookAheadAverage: 60, nearbyTotal: 90}, + {product: 'tablet', lookAheadAverage: null, nearbyTotal: 120}, + ); + }); + + it('rejects a range frame without a sort', async () => { + await expect( + execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {range: {preceding: 'unbounded', following: 'current'}}, + count('quantity').as('windowCount'), + ), + ), + ).to.be.rejectedWith(Error, /range/i); + }); + + it('rejects duplicate aliases and unsupported field values', () => { + expect(() => + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + sum('salesPrice').as('total'), + average('salesPrice').as('total'), + ), + ).to.throw(Error, /Duplicate alias or field 'total'/); + + expect(() => { + void execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: [constant(undefined as unknown as string)]}, + countAll().as('windowCount'), + ), + ); + }).to.throw(Error, /Cannot use "undefined" as a Firestore value/); + }); +}); + +// Comprehensive window function functional tests. Skipped by default on CI and +// local runs because the backend already tests window function semantics, and +// the active suite above exercises the SDK's window function code paths, server +// errors, and client-side validation. Set FIRESTORE_RUN_COMPREHENSIVE_TESTS=true +// to run the full matrix. +const runComprehensiveTests = + process.env.FIRESTORE_RUN_COMPREHENSIVE_TESTS === 'true' || + process.env.FIRESTORE_RUN_COMPREHENSIVE_TESTS === 'YES'; + +const comprehensiveDescribe = runComprehensiveTests + ? describe.skipClassic + : describe.skip; + +comprehensiveDescribe('Pipeline window functions (count)', () => { + let firestore: Firestore; + let productSales: CollectionReference; + + before(async () => { + productSales = getTestRoot(); + await populateWindowDocs(productSales); + firestore = productSales.firestore; + }); + + afterEach(() => verifyInstance(firestore)); + + describe('partitioning', () => { + it('treats an empty window spec as a single global window', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields({}, count('quantity').as('windowCount')) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 5}, + {product: 'phone', windowCount: 5}, + {product: 'tablet', windowCount: 5}, + {product: 'tablet', windowCount: 5}, + {product: 'tablet', windowCount: 5}, + ); + }); + + it('partitions by a single field name', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 2}, + {product: 'phone', windowCount: 2}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + ); + }); + + it('partitions by multiple fields', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product', 'region']}, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'region', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', region: 'east', windowCount: 1}, + {product: 'phone', region: 'west', windowCount: 1}, + {product: 'tablet', region: 'east', windowCount: 2}, + {product: 'tablet', region: 'west', windowCount: 1}, + {product: 'tablet', region: 'east', windowCount: 2}, + ); + }); + + it('partitions by an expression', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: [toUpper('region')]}, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('region', 'windowCount'), + ); + + expectResults( + snapshot, + {region: 'east', windowCount: 3}, + {region: 'west', windowCount: 2}, + {region: 'east', windowCount: 3}, + {region: 'west', windowCount: 2}, + {region: 'east', windowCount: 3}, + ); + }); + }); + + describe('sorting and default framing', () => { + it('computes a running count with the default (sorted) frame', async () => { + // The implicit default frame when `sort` is specified is a `range` frame + // from `'unbounded'` preceding to `'current'` following (peer-inclusive). + // `quantity` is 1..5 in the same order as `date` and has no ties, so the + // running count is 1..5. + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {sort: ascending('quantity')}, + count('quantity').as('runningCount'), + ) + .sort(ascending('date')) + .select('product', 'runningCount'), + ); + + expectResults( + snapshot, + {product: 'phone', runningCount: 1}, + {product: 'phone', runningCount: 2}, + {product: 'tablet', runningCount: 3}, + {product: 'tablet', runningCount: 4}, + {product: 'tablet', runningCount: 5}, + ); + }); + + it('computes a running count within each partition', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product'], sort: ascending('quantity')}, + count('quantity').as('runningCount'), + ) + .sort(ascending('date')) + .select('product', 'runningCount'), + ); + + expectResults( + snapshot, + {product: 'phone', runningCount: 1}, + {product: 'phone', runningCount: 2}, + {product: 'tablet', runningCount: 1}, + {product: 'tablet', runningCount: 2}, + {product: 'tablet', runningCount: 3}, + ); + }); + + it('honors a descending sort', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: descending('date'), + documents: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity').as('runningCount'), + ) + .sort(ascending('date')) + .select('product', 'runningCount'), + ); + + expectResults( + snapshot, + {product: 'phone', runningCount: 5}, + {product: 'phone', runningCount: 4}, + {product: 'tablet', runningCount: 3}, + {product: 'tablet', runningCount: 2}, + {product: 'tablet', runningCount: 1}, + ); + }); + + it('honors multiple sort orderings', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: [ascending('product'), ascending('date')], + documents: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity').as('runningCount'), + ) + .sort(ascending('date')) + .select('product', 'runningCount'), + ); + + expectResults( + snapshot, + {product: 'phone', runningCount: 1}, + {product: 'phone', runningCount: 2}, + {product: 'tablet', runningCount: 3}, + {product: 'tablet', runningCount: 4}, + {product: 'tablet', runningCount: 5}, + ); + }); + }); + + describe('documents framing', () => { + it("supports 'unbounded' preceding to 'current' following", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 1}, + {product: 'phone', windowCount: 2}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 4}, + {product: 'tablet', windowCount: 5}, + ); + }); + + it("supports 'unbounded' preceding to 'unbounded' following", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + partition: ['product'], + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'unbounded'}, + }, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 2}, + {product: 'phone', windowCount: 2}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + ); + }); + + it("supports 'current' preceding to 'current' following", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 'current', following: 'current'}, + }, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 1}, + {product: 'phone', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + ); + }); + + it("excludes tied peers with 'unbounded' preceding to 'current' following", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('salesPrice'), + documents: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity').as('runningCount'), + ) + .sort(ascending('runningCount')) + .select('salesPrice', 'runningCount'), + ); + + expectResults( + snapshot, + {salesPrice: 12, runningCount: 1}, + {salesPrice: 30, runningCount: 2}, + {salesPrice: 30, runningCount: 3}, + {salesPrice: 60, runningCount: 4}, + {salesPrice: 60, runningCount: 5}, + ); + }); + }); + + describe('range framing', () => { + // In a `range` frame the `'current'` sentinel is peer-inclusive (like SQL + // `CURRENT ROW` in RANGE mode), equivalent to a numeric `0` offset: it + // includes every document whose sort value(s) tie with the current + // document. + it("counts tied peers with 'current' to 'current'", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + partition: ['product'], + sort: ascending('salesPrice'), + range: {preceding: 'current', following: 'current'}, + }, + count('quantity').as('samePriceCount'), + ) + .sort(ascending('date')) + .select('product', 'salesPrice', 'samePriceCount'), + ); + + expectResults( + snapshot, + {product: 'phone', salesPrice: 12, samePriceCount: 1}, + {product: 'phone', salesPrice: 30, samePriceCount: 1}, + {product: 'tablet', salesPrice: 30, samePriceCount: 1}, + {product: 'tablet', salesPrice: 60, samePriceCount: 2}, + {product: 'tablet', salesPrice: 60, samePriceCount: 2}, + ); + }); + + it('counts tied peers with a zero offset on both bounds', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + partition: ['product'], + sort: ascending('salesPrice'), + range: {preceding: 0, following: 0}, + }, + count('quantity').as('samePriceCount'), + ) + .sort(ascending('date')) + .select('product', 'salesPrice', 'samePriceCount'), + ); + + expectResults( + snapshot, + {product: 'phone', salesPrice: 12, samePriceCount: 1}, + {product: 'phone', salesPrice: 30, samePriceCount: 1}, + {product: 'tablet', salesPrice: 30, samePriceCount: 1}, + {product: 'tablet', salesPrice: 60, samePriceCount: 2}, + {product: 'tablet', salesPrice: 60, samePriceCount: 2}, + ); + }); + + it("includes tied peers with 'unbounded' preceding to 'current' following", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('salesPrice'), + range: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity').as('cumulativeCount'), + ) + .sort(ascending('date')) + .select('salesPrice', 'cumulativeCount'), + ); + + expectResults( + snapshot, + {salesPrice: 12, cumulativeCount: 1}, + {salesPrice: 30, cumulativeCount: 3}, + {salesPrice: 30, cumulativeCount: 3}, + {salesPrice: 60, cumulativeCount: 5}, + {salesPrice: 60, cumulativeCount: 5}, + ); + }); + + it("includes tied peers with 'unbounded' preceding to a zero offset", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('salesPrice'), + range: {preceding: 'unbounded', following: 0}, + }, + count('quantity').as('cumulativeCount'), + ) + .sort(ascending('date')) + .select('salesPrice', 'cumulativeCount'), + ); + + expectResults( + snapshot, + {salesPrice: 12, cumulativeCount: 1}, + {salesPrice: 30, cumulativeCount: 3}, + {salesPrice: 30, cumulativeCount: 3}, + {salesPrice: 60, cumulativeCount: 5}, + {salesPrice: 60, cumulativeCount: 5}, + ); + }); + + it("supports a string sort key when bounds are 'unbounded' and 'current'", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('product'), + range: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity').as('cumulativeCount'), + ) + .sort(ascending('date')) + .select('product', 'cumulativeCount'), + ); + + expectResults( + snapshot, + {product: 'phone', cumulativeCount: 2}, + {product: 'phone', cumulativeCount: 2}, + {product: 'tablet', cumulativeCount: 5}, + {product: 'tablet', cumulativeCount: 5}, + {product: 'tablet', cumulativeCount: 5}, + ); + }); + + it("supports multiple sort orderings when bounds are 'unbounded' and 'current'", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: [ascending('product'), ascending('region')], + range: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity').as('cumulativeCount'), + ) + .sort(ascending('date')) + .select('product', 'region', 'cumulativeCount'), + ); + + expectResults( + snapshot, + {product: 'phone', region: 'east', cumulativeCount: 1}, + {product: 'phone', region: 'west', cumulativeCount: 2}, + {product: 'tablet', region: 'east', cumulativeCount: 4}, + {product: 'tablet', region: 'west', cumulativeCount: 5}, + {product: 'tablet', region: 'east', cumulativeCount: 4}, + ); + }); + + it("supports 'unbounded' to 'unbounded'", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + partition: ['product'], + sort: ascending('salesPrice'), + range: {preceding: 'unbounded', following: 'unbounded'}, + }, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 2}, + {product: 'phone', windowCount: 2}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + ); + }); + + it('applies the time unit to a range frame offset', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + range: { + preceding: 12, + following: 'current', + unit: 'hour', + }, + }, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 1}, + {product: 'phone', windowCount: 1}, + {product: 'tablet', windowCount: 2}, + {product: 'tablet', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + ); + }); + }); + + describe('accumulator level framing', () => { + it('evaluates each accumulator over its own frame', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product'], sort: ascending('date')}, + count('quantity') + .over({ + documents: {preceding: 'unbounded', following: 'current'}, + }) + .as('runningCount'), + count('quantity') + .over({ + documents: {preceding: 'unbounded', following: 'unbounded'}, + }) + .as('partitionCount'), + ) + .sort(ascending('date')) + .select('product', 'runningCount', 'partitionCount'), + ); + + expectResults( + snapshot, + {product: 'phone', runningCount: 1, partitionCount: 2}, + {product: 'phone', runningCount: 2, partitionCount: 2}, + {product: 'tablet', runningCount: 1, partitionCount: 3}, + {product: 'tablet', runningCount: 2, partitionCount: 3}, + {product: 'tablet', runningCount: 3, partitionCount: 3}, + ); + }); + + it('supports a range frame in over()', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product'], sort: ascending('salesPrice')}, + count('quantity') + .over({range: {preceding: 0, following: 0}}) + .as('samePriceCount'), + ) + .sort(ascending('date')) + .select('product', 'salesPrice', 'samePriceCount'), + ); + + expectResults( + snapshot, + {product: 'phone', salesPrice: 12, samePriceCount: 1}, + {product: 'phone', salesPrice: 30, samePriceCount: 1}, + {product: 'tablet', salesPrice: 30, samePriceCount: 1}, + {product: 'tablet', salesPrice: 60, samePriceCount: 2}, + {product: 'tablet', salesPrice: 60, samePriceCount: 2}, + ); + }); + }); + + describe('output fields and composition', () => { + it('supports multiple output fields in a single stage', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + count('quantity').as('productCount'), + count('quantity').as('productCountCopy'), + ) + .sort(ascending('date')) + .select('product', 'productCount', 'productCountCopy'), + ); + + expectResults( + snapshot, + {product: 'phone', productCount: 2, productCountCopy: 2}, + {product: 'phone', productCount: 2, productCountCopy: 2}, + {product: 'tablet', productCount: 3, productCountCopy: 3}, + {product: 'tablet', productCount: 3, productCountCopy: 3}, + {product: 'tablet', productCount: 3, productCountCopy: 3}, + ); + }); + + it('supports nested output field paths', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + count('quantity').as('stats.productCount'), + ) + .sort(ascending('date')) + .select('product', 'stats'), + ); + + expectResults( + snapshot, + {product: 'phone', stats: {productCount: 2}}, + {product: 'phone', stats: {productCount: 2}}, + {product: 'tablet', stats: {productCount: 3}}, + {product: 'tablet', stats: {productCount: 3}}, + {product: 'tablet', stats: {productCount: 3}}, + ); + }); + + it('supports chaining multiple addWindowFields stages', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + count('quantity').as('productCount'), + ) + .addWindowFields( + {partition: ['region']}, + count('quantity').as('regionCount'), + ) + .sort(ascending('date')) + .select('product', 'region', 'productCount', 'regionCount'), + ); + + expectResults( + snapshot, + { + product: 'phone', + region: 'east', + productCount: 2, + regionCount: 3, + }, + { + product: 'phone', + region: 'west', + productCount: 2, + regionCount: 2, + }, + { + product: 'tablet', + region: 'east', + productCount: 3, + regionCount: 3, + }, + { + product: 'tablet', + region: 'west', + productCount: 3, + regionCount: 2, + }, + { + product: 'tablet', + region: 'east', + productCount: 3, + regionCount: 3, + }, + ); + }); + + it('supports filtering on a window field in a later stage', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + count('quantity').as('productCount'), + ) + .where(field('productCount').greaterThan(2)) + .sort(ascending('date')) + .select('product', 'productCount'), + ); + + expectResults( + snapshot, + {product: 'tablet', productCount: 3}, + {product: 'tablet', productCount: 3}, + {product: 'tablet', productCount: 3}, + ); + }); + + it('supports sorting on a window field in a later stage', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + count('quantity').as('productCount'), + ) + .sort(ascending('productCount'), ascending('date')) + .select('product', 'productCount'), + ); + + expectResults( + snapshot, + {product: 'phone', productCount: 2}, + {product: 'phone', productCount: 2}, + {product: 'tablet', productCount: 3}, + {product: 'tablet', productCount: 3}, + {product: 'tablet', productCount: 3}, + ); + }); + }); + + describe('options overload', () => { + it('supports the options object signature', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales.path) + .addWindowFields({ + window: { + partition: ['product'], + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'current'}, + }, + fields: [ + count('quantity').as('runningCount'), + count('salesPrice').as('runningPriceCount'), + ], + }) + .sort(ascending('date')) + .select('product', 'runningCount', 'runningPriceCount'), + ); + + expectResults( + snapshot, + {product: 'phone', runningCount: 1, runningPriceCount: 1}, + {product: 'phone', runningCount: 2, runningPriceCount: 2}, + {product: 'tablet', runningCount: 1, runningPriceCount: 1}, + {product: 'tablet', runningCount: 2, runningPriceCount: 2}, + {product: 'tablet', runningCount: 3, runningPriceCount: 3}, + ); + }); + }); + + describe('error handling', () => { + it('rejects a range frame without a sort', async () => { + await expect( + execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {range: {preceding: 'unbounded', following: 'current'}}, + count('quantity').as('windowCount'), + ), + ), + ).to.be.rejectedWith(Error, /range/i); + }); + + it('rejects mixing stage level and accumulator level framing', async () => { + await expect( + execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'current'}, + }, + count('quantity') + .over({ + documents: {preceding: 'unbounded', following: 'unbounded'}, + }) + .as('windowCount'), + ), + ), + ).to.be.rejectedWith(Error); + }); + + it('evaluates accumulators without over() using the default frame when others use over()', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {sort: ascending('date')}, + count('quantity') + .over({documents: {preceding: 0, following: 0}}) + .as('framed'), + count('quantity').as('unframed'), + ) + .sort(ascending('date')) + .select('product', 'framed', 'unframed'), + ); + + expectResults( + snapshot, + {product: 'phone', framed: 1, unframed: 1}, + {product: 'phone', framed: 1, unframed: 2}, + {product: 'tablet', framed: 1, unframed: 3}, + {product: 'tablet', framed: 1, unframed: 4}, + {product: 'tablet', framed: 1, unframed: 5}, + ); + }); + }); +}); + +comprehensiveDescribe('Pipeline window functions (aggregators)', () => { + let firestore: Firestore; + let productSales: CollectionReference; + + before(async () => { + productSales = getTestRoot(); + await populateWindowDocs(productSales); + firestore = productSales.firestore; + }); + + afterEach(() => verifyInstance(firestore)); + + describe('numeric aggregators over a partition', () => { + it('computes sum, average, minimum and maximum', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + sum('salesPrice').as('total'), + average('salesPrice').as('averagePrice'), + minimum('salesPrice').as('minimumPrice'), + maximum('salesPrice').as('maximumPrice'), + ) + .sort(ascending('date')) + .select( + 'product', + 'total', + 'averagePrice', + 'minimumPrice', + 'maximumPrice', + ), + ); + + expectResults( + snapshot, + { + product: 'phone', + total: 42, + averagePrice: 21, + minimumPrice: 12, + maximumPrice: 30, + }, + { + product: 'phone', + total: 42, + averagePrice: 21, + minimumPrice: 12, + maximumPrice: 30, + }, + { + product: 'tablet', + total: 150, + averagePrice: 50, + minimumPrice: 30, + maximumPrice: 60, + }, + { + product: 'tablet', + total: 150, + averagePrice: 50, + minimumPrice: 30, + maximumPrice: 60, + }, + { + product: 'tablet', + total: 150, + averagePrice: 50, + minimumPrice: 30, + maximumPrice: 60, + }, + ); + }); + + // Un-skip once the backend supports countIf and countDistinct in addWindowFields. + it.skip('computes countIf', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + countIf(field('salesPrice').greaterThan(20)).as('expensiveCount'), + ) + .sort(ascending('date')) + .select('product', 'expensiveCount'), + ); + + expectResults( + snapshot, + {product: 'phone', expensiveCount: 1}, + {product: 'phone', expensiveCount: 1}, + {product: 'tablet', expensiveCount: 3}, + {product: 'tablet', expensiveCount: 3}, + {product: 'tablet', expensiveCount: 3}, + ); + }); + + // Un-skip once the backend supports countIf and countDistinct in addWindowFields. + it.skip('computes countDistinct', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + countDistinct('salesPrice').as('distinctPriceCount'), + ) + .sort(ascending('date')) + .select('product', 'distinctPriceCount'), + ); + + expectResults( + snapshot, + {product: 'phone', distinctPriceCount: 2}, + {product: 'phone', distinctPriceCount: 2}, + {product: 'tablet', distinctPriceCount: 2}, + {product: 'tablet', distinctPriceCount: 2}, + {product: 'tablet', distinctPriceCount: 2}, + ); + }); + + it('aggregates over a computed expression', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + sum(multiply(field('salesPrice'), field('quantity'))).as( + 'totalRevenue', + ), + ) + .sort(ascending('date')) + .select('product', 'totalRevenue'), + ); + + expectResults( + snapshot, + {product: 'phone', totalRevenue: 72}, + {product: 'phone', totalRevenue: 72}, + {product: 'tablet', totalRevenue: 630}, + {product: 'tablet', totalRevenue: 630}, + {product: 'tablet', totalRevenue: 630}, + ); + }); + }); + + describe('ordered aggregators', () => { + it('computes first and last over the whole partition', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + partition: ['product'], + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'unbounded'}, + }, + first('salesPrice').as('firstPrice'), + last('salesPrice').as('lastPrice'), + ) + .sort(ascending('date')) + .select('product', 'firstPrice', 'lastPrice'), + ); + + expectResults( + snapshot, + {product: 'phone', firstPrice: 12, lastPrice: 30}, + {product: 'phone', firstPrice: 12, lastPrice: 30}, + {product: 'tablet', firstPrice: 30, lastPrice: 60}, + {product: 'tablet', firstPrice: 30, lastPrice: 60}, + {product: 'tablet', firstPrice: 30, lastPrice: 60}, + ); + }); + + it('computes arrayAgg and arrayAggDistinct', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + partition: ['product'], + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'unbounded'}, + }, + arrayAgg('salesPrice').as('allPrices'), + arrayAggDistinct('salesPrice').as('distinctPrices'), + ) + .sort(ascending('date')) + .select('product', 'allPrices', 'distinctPrices'), + ); + + const results = snapshot.results.map(r => { + const data = r.data(); + data['distinctPrices'] = [...data['distinctPrices']].sort( + (a: number, b: number) => a - b, + ); + return data; + }); + + expect(results).to.deep.equal([ + { + product: 'phone', + allPrices: [12, 30], + distinctPrices: [12, 30], + }, + { + product: 'phone', + allPrices: [12, 30], + distinctPrices: [12, 30], + }, + { + product: 'tablet', + allPrices: [30, 60, 60], + distinctPrices: [30, 60], + }, + { + product: 'tablet', + allPrices: [30, 60, 60], + distinctPrices: [30, 60], + }, + { + product: 'tablet', + allPrices: [30, 60, 60], + distinctPrices: [30, 60], + }, + ]); + }); + + it('computes a running total', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'current'}, + }, + sum('salesPrice').as('runningTotal'), + ) + .sort(ascending('date')) + .select('product', 'runningTotal'), + ); + + expectResults( + snapshot, + {product: 'phone', runningTotal: 12}, + {product: 'phone', runningTotal: 42}, + {product: 'tablet', runningTotal: 72}, + {product: 'tablet', runningTotal: 132}, + {product: 'tablet', runningTotal: 192}, + ); + }); + }); + + describe('numeric documents offsets', () => { + it('computes a centered moving average', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 1, following: 1}, + }, + average('salesPrice').as('movingAverage'), + countAll().as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'movingAverage', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', movingAverage: 21, windowCount: 2}, + {product: 'phone', movingAverage: 24, windowCount: 3}, + {product: 'tablet', movingAverage: 40, windowCount: 3}, + {product: 'tablet', movingAverage: 50, windowCount: 3}, + {product: 'tablet', movingAverage: 60, windowCount: 2}, + ); + }); + + it('computes a trailing sum', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 2, following: 0}, + }, + sum('salesPrice').as('trailingTotal'), + ) + .sort(ascending('date')) + .select('product', 'trailingTotal'), + ); + + expectResults( + snapshot, + {product: 'phone', trailingTotal: 12}, + {product: 'phone', trailingTotal: 42}, + {product: 'tablet', trailingTotal: 72}, + {product: 'tablet', trailingTotal: 120}, + {product: 'tablet', trailingTotal: 150}, + ); + }); + + it('computes a look ahead average with a negative preceding bound', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + // Equivalent to "documents between 1 following and 2 + // following": the window is [index - preceding, index + + // following]. + sort: ascending('date'), + documents: {preceding: -1, following: 2}, + }, + average('salesPrice').as('lookAheadAverage'), + countAll().as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'lookAheadAverage', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', lookAheadAverage: 30, windowCount: 2}, + {product: 'phone', lookAheadAverage: 45, windowCount: 2}, + {product: 'tablet', lookAheadAverage: 60, windowCount: 2}, + {product: 'tablet', lookAheadAverage: 60, windowCount: 1}, + {product: 'tablet', lookAheadAverage: null, windowCount: 0}, + ); + }); + + it('computes a look behind average with a negative following bound', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + // Equivalent to "documents between 2 preceding and 1 + // preceding": the window is [index - preceding, index + + // following]. + sort: ascending('date'), + documents: {preceding: 2, following: -1}, + }, + average('salesPrice').as('lookBehindAverage'), + countAll().as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'lookBehindAverage', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', lookBehindAverage: null, windowCount: 0}, + {product: 'phone', lookBehindAverage: 12, windowCount: 1}, + {product: 'tablet', lookBehindAverage: 21, windowCount: 2}, + {product: 'tablet', lookBehindAverage: 30, windowCount: 2}, + {product: 'tablet', lookBehindAverage: 45, windowCount: 2}, + ); + }); + + it('computes a range look behind average with a negative following bound', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + // Equivalent to a range frame of [value - Infinity, value - 1], + // including only documents with a strictly lower salesPrice. + sort: ascending('salesPrice'), + range: {preceding: 'unbounded', following: -1}, + }, + average('salesPrice').as('lookBehindAverage'), + countAll().as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'lookBehindAverage', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', lookBehindAverage: null, windowCount: 0}, + {product: 'phone', lookBehindAverage: 12, windowCount: 1}, + {product: 'tablet', lookBehindAverage: 12, windowCount: 1}, + {product: 'tablet', lookBehindAverage: 24, windowCount: 3}, + {product: 'tablet', lookBehindAverage: 24, windowCount: 3}, + ); + }); + + it('computes a range look ahead average with a negative preceding bound', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + // Equivalent to a range frame of [value + 1, value + Infinity], + // including only documents with a strictly higher salesPrice. + sort: ascending('salesPrice'), + range: {preceding: -1, following: 'unbounded'}, + }, + average('salesPrice').as('lookAheadAverage'), + countAll().as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'lookAheadAverage', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', lookAheadAverage: 45, windowCount: 4}, + {product: 'phone', lookAheadAverage: 60, windowCount: 2}, + {product: 'tablet', lookAheadAverage: 60, windowCount: 2}, + {product: 'tablet', lookAheadAverage: null, windowCount: 0}, + {product: 'tablet', lookAheadAverage: null, windowCount: 0}, + ); + }); + }); + + describe('numeric and time range offsets', () => { + it('computes a value based range window', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('salesPrice'), + range: {preceding: 20, following: 0}, + }, + sum('salesPrice').as('nearbyTotal'), + ) + .sort(ascending('date')) + .select('salesPrice', 'nearbyTotal'), + ); + + expectResults( + snapshot, + {salesPrice: 12, nearbyTotal: 12}, + {salesPrice: 30, nearbyTotal: 72}, + {salesPrice: 30, nearbyTotal: 72}, + {salesPrice: 60, nearbyTotal: 120}, + {salesPrice: 60, nearbyTotal: 120}, + ); + }); + + it('computes a value based range window with fractional offsets', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('salesPrice'), + range: {preceding: 17.5, following: 0.5}, + }, + sum('salesPrice').as('nearbyTotal'), + ) + .sort(ascending('date')) + .select('salesPrice', 'nearbyTotal'), + ); + + expectResults( + snapshot, + {salesPrice: 12, nearbyTotal: 12}, + {salesPrice: 30, nearbyTotal: 60}, + {salesPrice: 30, nearbyTotal: 60}, + {salesPrice: 60, nearbyTotal: 120}, + {salesPrice: 60, nearbyTotal: 120}, + ); + }); + + it('computes a trailing 3 day total', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + range: {preceding: 3, following: 'current', unit: 'day'}, + }, + sum('salesPrice').as('threeDayTotal'), + ) + .sort(ascending('date')) + .select('product', 'threeDayTotal'), + ); + + expectResults( + snapshot, + {product: 'phone', threeDayTotal: 12}, + {product: 'phone', threeDayTotal: 42}, + {product: 'tablet', threeDayTotal: 72}, + {product: 'tablet', threeDayTotal: 132}, + {product: 'tablet', threeDayTotal: 60}, + ); + }); + }); + + describe('countAll', () => { + it('counts every document in the window', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {partition: ['product']}, + countAll().as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 2}, + {product: 'phone', windowCount: 2}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 3}, + ); + }); + }); + + describe('documents framing with numeric offsets', () => { + it('supports zero preceding and zero following', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 0, following: 0}, + }, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 1}, + {product: 'phone', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + {product: 'tablet', windowCount: 1}, + ); + }); + + it("supports zero preceding to 'unbounded' following", async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 0, following: 'unbounded'}, + }, + count('quantity').as('windowCount'), + ) + .sort(ascending('date')) + .select('product', 'windowCount'), + ); + + expectResults( + snapshot, + {product: 'phone', windowCount: 5}, + {product: 'phone', windowCount: 4}, + {product: 'tablet', windowCount: 3}, + {product: 'tablet', windowCount: 2}, + {product: 'tablet', windowCount: 1}, + ); + }); + }); +}); + +// Un-skip once the backend supports `rank()`. +describe.skip('Pipeline window functions (ranking)', () => { + let firestore: Firestore; + let productSales: CollectionReference; + + before(async () => { + productSales = getTestRoot(); + await populateWindowDocs(productSales); + firestore = productSales.firestore; + }); + + afterEach(() => verifyInstance(firestore)); + + it('computes rank', async () => { + const snapshot = await execute( + firestore + .pipeline() + .collection(productSales) + .addWindowFields( + {sort: [ascending('salesPrice'), ascending('date')]}, + rank().as('priceRank'), + ) + .sort(ascending('date')) + .select('salesPrice', 'priceRank'), + ); + + expectResults( + snapshot, + { + salesPrice: 12, + priceRank: 1, + }, + { + salesPrice: 30, + priceRank: 2, + }, + { + salesPrice: 30, + priceRank: 2, + }, + { + salesPrice: 60, + priceRank: 4, + }, + { + salesPrice: 60, + priceRank: 4, + }, + ); + }); +}); diff --git a/handwritten/firestore/dev/test/pipelines/add-window-fields.ts b/handwritten/firestore/dev/test/pipelines/add-window-fields.ts new file mode 100644 index 00000000000..0457a8a2e5d --- /dev/null +++ b/handwritten/firestore/dev/test/pipelines/add-window-fields.ts @@ -0,0 +1,857 @@ +/** + * @license + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import {expect} from 'chai'; + +import {google} from '../../protos/firestore_v1_proto_api'; +import {Firestore, Pipelines} from '../../src'; +import {createInstance} from '../util/helpers'; + +import Pipeline = Pipelines.Pipeline; +import ProtoStage = google.firestore.v1.Pipeline.IStage; +import ProtoValue = google.firestore.v1.IValue; + +const { + arrayAgg, + arrayAggDistinct, + ascending, + average, + constant, + count, + countAll, + countDistinct, + countIf, + descending, + field, + first, + last, + maximum, + minimum, + rank, + sum, + toLower, +} = Pipelines; + +let db: Firestore; + +/** + * Serializes `pipeline` the same way `execute()` does, and returns the proto + * for its (single) `add_window_fields` stage. + */ +function windowStage(pipeline: Pipeline): ProtoStage { + pipeline._validateUserData(db._settings.ignoreUndefinedProperties ?? false); + const proto = pipeline._toProto(); + const stages = (proto.stages ?? []).filter( + s => s.name === 'add_window_fields', + ); + expect(stages).to.have.lengthOf( + 1, + 'expected exactly one add_window_fields stage', + ); + return stages[0]; +} + +/** The `window_spec` argument (args[0]) of the `add_window_fields` stage. */ +function windowSpecArg(pipeline: Pipeline): ProtoValue { + const stage = windowStage(pipeline); + expect(stage.args).to.have.lengthOf( + 2, + 'add_window_fields must have exactly 2 args', + ); + return stage.args![0]; +} + +/** The `fields` argument (args[1]) of the `add_window_fields` stage. */ +function fieldsArg(pipeline: Pipeline): ProtoValue { + const stage = windowStage(pipeline); + expect(stage.args).to.have.lengthOf( + 2, + 'add_window_fields must have exactly 2 args', + ); + return stage.args![1]; +} + +function fieldRef(name: string): ProtoValue { + return {fieldReferenceValue: name}; +} + +function int(value: number): ProtoValue { + return {integerValue: value}; +} + +function str(value: string): ProtoValue { + return {stringValue: value}; +} + +function map(fields: Record): ProtoValue { + return {mapValue: {fields}}; +} + +function array(...values: ProtoValue[]): ProtoValue { + return {arrayValue: {values}}; +} + +function fn(name: string, ...args: ProtoValue[]): ProtoValue { + return {functionValue: {name, args}}; +} + +function ordering(expression: ProtoValue, direction: string): ProtoValue { + return map({direction: str(direction), expression}); +} + +function basePipeline(): Pipeline { + return db.pipeline().collection('sales'); +} + +describe('addWindowFields() serialization', () => { + beforeEach(async () => { + db = await createInstance(); + }); + + describe('stage shape', () => { + it('uses the add_window_fields stage name and 2 args', () => { + const stage = windowStage( + basePipeline().addWindowFields( + {partition: ['product']}, + countAll().as('c'), + ), + ); + + expect(stage.name).to.equal('add_window_fields'); + expect(stage.args).to.have.lengthOf(2); + }); + + it('does not send any stage options', () => { + const stage = windowStage( + basePipeline().addWindowFields( + {partition: ['product']}, + countAll().as('c'), + ), + ); + + expect(stage.options).to.deep.equal({}); + }); + }); + + describe('window_spec', () => { + it('serializes an empty window spec as an empty map', () => { + expect( + windowSpecArg(basePipeline().addWindowFields({}, countAll().as('c'))), + ).to.deep.equal(map({})); + }); + + it('serializes a partition of field name strings', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {partition: ['product', 'region']}, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({partition: array(fieldRef('product'), fieldRef('region'))}), + ); + }); + + it('serializes a partition of expressions', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {partition: [field('product'), toLower('region')]}, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + partition: array( + fieldRef('product'), + fn('to_lower', fieldRef('region')), + ), + }), + ); + }); + + it('serializes a nested field path partition', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {partition: ['metadata.region']}, + countAll().as('c'), + ), + ), + ).to.deep.equal(map({partition: array(fieldRef('metadata.region'))})); + }); + + it('serializes a single (non array) sort ordering', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {sort: ascending('date')}, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({sort: array(ordering(fieldRef('date'), 'ascending'))}), + ); + }); + + it('serializes multiple sort orderings with mixed directions', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {sort: [ascending('date'), descending('salesPrice')]}, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array( + ordering(fieldRef('date'), 'ascending'), + ordering(fieldRef('salesPrice'), 'descending'), + ), + }), + ); + }); + + it('serializes partition and sort together', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {partition: ['product'], sort: ascending('date')}, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + partition: array(fieldRef('product')), + sort: array(ordering(fieldRef('date'), 'ascending')), + }), + ); + }); + + it('omits an empty partition array', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {partition: [], sort: ascending('date')}, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({sort: array(ordering(fieldRef('date'), 'ascending'))}), + ); + }); + + it('omits an empty sort array', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {partition: ['product'], sort: []}, + countAll().as('c'), + ), + ), + ).to.deep.equal(map({partition: array(fieldRef('product'))})); + }); + + it('omits the frame when using default framing', () => { + const spec = windowSpecArg( + basePipeline().addWindowFields( + {sort: ascending('date')}, + countAll().as('c'), + ), + ); + + expect(spec.mapValue!.fields).to.not.have.property('documents'); + expect(spec.mapValue!.fields).to.not.have.property('range'); + }); + }); + + describe('stage level documents framing', () => { + it('serializes numeric offsets as integers', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 2, following: 1}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('date'), 'ascending')), + documents: map({preceding: int(2), following: int(1)}), + }), + ); + }); + + it('serializes zero offsets', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 0, following: 0}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('date'), 'ascending')), + documents: map({preceding: int(0), following: int(0)}), + }), + ); + }); + + it('serializes negative offsets (look ahead windows)', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('date'), + documents: {preceding: -1, following: 2}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('date'), 'ascending')), + documents: map({preceding: int(-1), following: int(2)}), + }), + ); + }); + + it("serializes the 'unbounded' and 'current' sentinels", () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('date'), + documents: {preceding: 'unbounded', following: 'current'}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('date'), 'ascending')), + documents: map({ + preceding: str('unbounded'), + following: str('current'), + }), + }), + ); + }); + + it('serializes an unbounded to unbounded frame', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + {documents: {preceding: 'unbounded', following: 'unbounded'}}, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + documents: map({ + preceding: str('unbounded'), + following: str('unbounded'), + }), + }), + ); + }); + + it('serializes constant expression bounds', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('date'), + documents: {preceding: constant(3), following: constant(0)}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('date'), 'ascending')), + documents: map({preceding: int(3), following: int(0)}), + }), + ); + }); + }); + + describe('stage level range framing', () => { + it('serializes numeric range offsets', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('salesPrice'), + range: {preceding: 10, following: 10}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('salesPrice'), 'ascending')), + range: map({preceding: int(10), following: int(10)}), + }), + ); + }); + + it('serializes fractional range offsets as doubles', () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('salesPrice'), + range: {preceding: 2.5, following: 0}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('salesPrice'), 'ascending')), + range: map({preceding: {doubleValue: 2.5}, following: int(0)}), + }), + ); + }); + + it("serializes a range frame with a 'day' time unit", () => { + expect( + windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('date'), + range: {preceding: 30, following: 'current', unit: 'day'}, + }, + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + sort: array(ordering(fieldRef('date'), 'ascending')), + range: map({ + preceding: int(30), + following: str('current'), + unit: str('day'), + }), + }), + ); + }); + + for (const unit of [ + 'microsecond', + 'millisecond', + 'second', + 'minute', + 'hour', + 'day', + 'week', + 'month', + 'quarter', + 'year', + ] as const) { + it(`serializes the '${unit}' time unit`, () => { + const spec = windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('date'), + range: {preceding: 1, following: 'current', unit}, + }, + countAll().as('c'), + ), + ); + + expect(spec.mapValue!.fields!.range).to.deep.equal( + map({preceding: int(1), following: str('current'), unit: str(unit)}), + ); + }); + } + + it("omits 'unit' when it is not specified", () => { + const spec = windowSpecArg( + basePipeline().addWindowFields( + { + sort: ascending('salesPrice'), + range: {preceding: 'unbounded', following: 'current'}, + }, + countAll().as('c'), + ), + ); + + expect(spec.mapValue!.fields!.range).to.deep.equal( + map({preceding: str('unbounded'), following: str('current')}), + ); + }); + }); + + describe('fields', () => { + it('maps each alias to its aggregate function', () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {partition: ['product']}, + sum('salesPrice').as('total'), + countAll().as('c'), + ), + ), + ).to.deep.equal( + map({ + total: fn('sum', fieldRef('salesPrice')), + c: fn('count'), + }), + ); + }); + + it('supports nested output field paths', () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {partition: ['product']}, + sum('salesPrice').as('stats.total'), + ), + ), + ).to.deep.equal(map({'stats.total': fn('sum', fieldRef('salesPrice'))})); + }); + + it('serializes all supported aggregate function names', () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {partition: ['product']}, + countAll().as('countAll'), + count('salesPrice').as('count'), + countIf(field('salesPrice').greaterThan(10)).as('countIf'), + countDistinct('salesPrice').as('countDistinct'), + sum('salesPrice').as('sum'), + average('salesPrice').as('average'), + minimum('salesPrice').as('minimum'), + maximum('salesPrice').as('maximum'), + first('salesPrice').as('first'), + last('salesPrice').as('last'), + arrayAgg('salesPrice').as('arrayAgg'), + arrayAggDistinct('salesPrice').as('arrayAggDistinct'), + ), + ), + ).to.deep.equal( + map({ + countAll: fn('count'), + count: fn('count', fieldRef('salesPrice')), + countIf: fn( + 'count_if', + fn('greater_than', fieldRef('salesPrice'), int(10)), + ), + countDistinct: fn('count_distinct', fieldRef('salesPrice')), + sum: fn('sum', fieldRef('salesPrice')), + average: fn('average', fieldRef('salesPrice')), + minimum: fn('minimum', fieldRef('salesPrice')), + maximum: fn('maximum', fieldRef('salesPrice')), + first: fn('first', fieldRef('salesPrice')), + last: fn('last', fieldRef('salesPrice')), + arrayAgg: fn('array_agg', fieldRef('salesPrice')), + arrayAggDistinct: fn('array_agg_distinct', fieldRef('salesPrice')), + }), + ); + }); + + it('serializes aggregates over computed expressions', () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {partition: ['product']}, + sum(field('salesPrice').multiply(2)).as('doubled'), + ), + ), + ).to.deep.equal( + map({ + doubled: fn('sum', fn('multiply', fieldRef('salesPrice'), int(2))), + }), + ); + }); + + it('serializes the rank window function', () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {sort: descending('salesPrice')}, + rank().as('rank'), + ), + ), + ).to.deep.equal( + map({ + rank: fn('rank'), + }), + ); + }); + + it('rejects duplicate aliases', () => { + expect(() => + basePipeline().addWindowFields( + {partition: ['product']}, + sum('salesPrice').as('total'), + average('salesPrice').as('total'), + ), + ).to.throw(/Duplicate alias or field 'total'/); + }); + }); + + describe('accumulator level framing (over)', () => { + it("wraps an aggregate in over() with a 'documents' frame", () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {sort: ascending('date')}, + average('salesPrice') + .over({documents: {preceding: 1, following: 1}}) + .as('movingAverage'), + ), + ), + ).to.deep.equal( + map({ + movingAverage: fn( + 'over', + fn('average', fieldRef('salesPrice')), + map({ + documents: map({preceding: int(1), following: int(1)}), + }), + ), + }), + ); + }); + + it("wraps an aggregate in over() with a 'range' frame and time unit", () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {sort: ascending('date')}, + sum('salesPrice') + .over({ + range: {preceding: 10, following: 0, unit: 'day'}, + }) + .as('tenDayTotal'), + ), + ), + ).to.deep.equal( + map({ + tenDayTotal: fn( + 'over', + fn('sum', fieldRef('salesPrice')), + map({ + range: map({ + preceding: int(10), + following: int(0), + unit: str('day'), + }), + }), + ), + }), + ); + }); + + it('supports different frames for different accumulators', () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {partition: ['product'], sort: ascending('date')}, + sum('salesPrice') + .over({ + documents: {preceding: 'unbounded', following: 'current'}, + }) + .as('runningTotal'), + average('salesPrice') + .over({documents: {preceding: 1, following: 1}}) + .as('movingAverage'), + ), + ), + ).to.deep.equal( + map({ + runningTotal: fn( + 'over', + fn('sum', fieldRef('salesPrice')), + map({ + documents: map({ + preceding: str('unbounded'), + following: str('current'), + }), + }), + ), + movingAverage: fn( + 'over', + fn('average', fieldRef('salesPrice')), + map({documents: map({preceding: int(1), following: int(1)})}), + ), + }), + ); + }); + + it('supports over() on a ranking window function', () => { + expect( + fieldsArg( + basePipeline().addWindowFields( + {sort: ascending('date')}, + rank() + .over({ + documents: {preceding: 'unbounded', following: 'current'}, + }) + .as('r'), + ), + ), + ).to.deep.equal( + map({ + r: fn( + 'over', + fn('rank'), + map({ + documents: map({ + preceding: str('unbounded'), + following: str('current'), + }), + }), + ), + }), + ); + }); + }); + + describe('options overload', () => { + it('produces the same proto as the varargs overload', () => { + const varargs = windowStage( + basePipeline().addWindowFields( + { + partition: ['product'], + sort: ascending('date'), + documents: {preceding: 1, following: 1}, + }, + average('salesPrice').as('movingAverage'), + countAll().as('windowCount'), + ), + ); + + const options = windowStage( + basePipeline().addWindowFields({ + window: { + partition: ['product'], + sort: ascending('date'), + documents: {preceding: 1, following: 1}, + }, + fields: [ + average('salesPrice').as('movingAverage'), + countAll().as('windowCount'), + ], + }), + ); + + expect(options).to.deep.equal(varargs); + }); + + it('does not leak window/fields into the stage options', () => { + const stage = windowStage( + basePipeline().addWindowFields({ + window: {partition: ['product']}, + fields: [countAll().as('c')], + }), + ); + + expect(stage.options).to.deep.equal({}); + }); + }); + + describe('user data validation', () => { + const invalidExpr = constant(undefined as unknown as number); + const expectedError = /Cannot use "undefined" as a Firestore value/; + + it('validates expressions in partition', () => { + expect(() => + windowStage( + basePipeline().addWindowFields( + {partition: [invalidExpr]}, + countAll().as('c'), + ), + ), + ).to.throw(expectedError); + }); + + it('validates expressions in sort', () => { + expect(() => + windowStage( + basePipeline().addWindowFields( + {sort: ascending(invalidExpr)}, + countAll().as('c'), + ), + ), + ).to.throw(expectedError); + }); + + it('validates expressions in documents and range frame bounds and unit', () => { + expect(() => + windowStage( + basePipeline().addWindowFields( + { + sort: ascending('date'), + documents: {preceding: invalidExpr, following: 'current'}, + }, + countAll().as('c'), + ), + ), + ).to.throw(expectedError); + + expect(() => + windowStage( + basePipeline().addWindowFields( + { + sort: ascending('date'), + range: { + preceding: 1, + following: 'current', + unit: invalidExpr, + }, + }, + countAll().as('c'), + ), + ), + ).to.throw(expectedError); + }); + + it('validates expressions in fields and over() frames', () => { + expect(() => + windowStage( + basePipeline().addWindowFields({}, sum(invalidExpr).as('total')), + ), + ).to.throw(expectedError); + + expect(() => + windowStage( + basePipeline().addWindowFields( + {sort: ascending('date')}, + countAll() + .over({ + documents: {preceding: 'unbounded', following: invalidExpr}, + }) + .as('c'), + ), + ), + ).to.throw(expectedError); + }); + }); +}); diff --git a/handwritten/firestore/dev/test/pipelines/stage.ts b/handwritten/firestore/dev/test/pipelines/stage.ts index 6cb7ccb7e5f..bd8a3a4c070 100644 --- a/handwritten/firestore/dev/test/pipelines/stage.ts +++ b/handwritten/firestore/dev/test/pipelines/stage.ts @@ -156,6 +156,19 @@ describe('stage option serialization', () => { }), stageIndex: 1, }, + { + name: 'addWindowFields stage', + pipeline: firestore => + firestore + .pipeline() + .database() + .addWindowFields({ + window: {partition: ['foo']}, + fields: [sum('foo').as('fooSum')], + rawOptions, + }), + stageIndex: 1, + }, { name: 'aggregate stage', pipeline: firestore => @@ -445,6 +458,48 @@ describe('stage _validateUserData', () => { .execute(); }); + it('addWindowFields stage validation', async () => { + const firestore = await createInstance(); + + // Should throw when ignoreUndefinedProperties is false (default) + expect(() => { + void firestore + .pipeline() + .collection('foo') + .addWindowFields( + { + partition: [constant({foo: undefined})], + }, + sum('amount').as('total'), + ) + .execute(); + }).to.throw( + 'Value for argument "value" is not a valid constant value. Cannot use "undefined" as a Firestore value', + ); + + // Should not throw when ignoreUndefinedProperties is true + const spy = sinon.fake.returns(stream()); + const firestoreWithIgnore = await createInstance( + { + executePipeline: spy, + }, + { + ignoreUndefinedProperties: true, + }, + ); + + await firestoreWithIgnore + .pipeline() + .collection('foo') + .addWindowFields( + { + partition: [constant({foo: undefined})], + }, + sum('amount').as('total'), + ) + .execute(); + }); + describe('literals stage rejects FieldValue sentinels', () => { const cases: Array<{ name: string; diff --git a/handwritten/firestore/types/firestore.d.ts b/handwritten/firestore/types/firestore.d.ts index 9d2ae5ec441..45babc3881e 100644 --- a/handwritten/firestore/types/firestore.d.ts +++ b/handwritten/firestore/types/firestore.d.ts @@ -3421,6 +3421,7 @@ declare namespace FirebaseFirestore { | 'Constant' | 'Function' | 'AggregateFunction' + | 'WindowFunction' | 'ListOfExprs' | 'AliasedExpression' | 'Variable' @@ -6094,6 +6095,35 @@ declare namespace FirebaseFirestore { * AggregateFunction and associates it with the provided alias. */ as(name: string): AliasedAggregate; + + /** + * Evaluates this aggregate function over a specific window frame, rather than + * over the window frame defined by the enclosing + * {@link Pipeline.addWindowFields} stage. + * + * @example + * ```typescript + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { sort: ascending('date') }, + * // Uses the stage's default frame. + * sum('amount').as('runningTotal'), + * // Overrides the stage frame with a 3 document moving window. + * average('amount') + * .over({ documents: { preceding: 1, following: 1 } }) + * .as('movingAverage') + * ); + * ``` + * + * @param frame - The `documents` or `range` frame to evaluate this aggregate over. + * @returns A new {@link WindowFunction}. + */ + over( + frame: OneOf<{ + documents: DocumentWindowFrame; + range: RangeWindowFrame; + }>, + ): WindowFunction; } /** @@ -6114,6 +6144,61 @@ declare namespace FirebaseFirestore { readonly _alias: string; } + /** + * A function that is evaluated over a window frame of documents, as part of an + * {@link Pipeline.addWindowFields} stage. + * + * A `WindowFunction` is created either by one of the dedicated window function + * factories (such as {@link rank}), or by calling + * `over()` on an {@link AggregateFunction}. + */ + export class WindowFunction { + expressionType: ExpressionType; + + constructor(name: string, params?: Expression[]); + + /** + * Evaluates this window function over a specific window frame, rather than + * over the window frame defined by the enclosing + * {@link Pipeline.addWindowFields} stage. + * + * @param frame - The `documents` or `range` frame to evaluate this function over. + * @returns A new {@link WindowFunction}. + */ + over( + frame: OneOf<{ + documents: DocumentWindowFrame; + range: RangeWindowFrame; + }>, + ): WindowFunction; + + /** + * Assigns an alias to this `WindowFunction`. The alias specifies the name that + * the computed value will have in the output document. + * + * @param name - The alias to assign to this `WindowFunction`. + * @returns A new {@link AliasedWindowFunction}. + */ + as(name: string): AliasedWindowFunction; + } + + /** + * A {@link WindowFunction} with an alias. + */ + export class AliasedWindowFunction { + /** + * The underlying `WindowFunction` that this aliased window function wraps. + * @internal + */ + readonly _windowFunction: WindowFunction; + + /** + * Specifies the name of the property that will contain the window function result in the output document. + * @internal + */ + readonly _alias: string; + } + /** * Represents an expression that has been assigned an alias using the `.as()` method. * @@ -12190,6 +12275,25 @@ declare namespace FirebaseFirestore { location: GeoPoint | Expression, ): Expression; + /** + * Creates a window function that computes the rank of the current document + * within its window frame. Documents that compare equal in the window sort + * order receive the same rank, and the next rank is offset by the number of + * tied documents. + * + * @example + * ```typescript + * firestore.pipeline().collection("employees") + * .addWindowFields( + * { partition: ['department'], sort: descending('salary') }, + * rank().as('salaryRank') + * ); + * ``` + * + * @returns A new {@link WindowFunction}. + */ + export function rank(): WindowFunction; + // TODO(search) enable when supported by the backend // /** // * Evaluates if the value in the field specified by `fieldName` is between @@ -12577,6 +12681,110 @@ declare namespace FirebaseFirestore { * @returns A new `Pipeline` object with this stage appended to the stage list. */ addFields(options: AddFieldsStageOptions): Pipeline; + + /** + * Adds window function results to the output documents of the pipeline. + * + * Window functions evaluate expressions over a subset of documents (a "window frame") relative to the + * current document. + * + * @example + * ```typescript + * // 1. Unsorted partition/group aggregation (evaluates over the entire group) + * firestore.pipeline().collection("employees") + * .addWindowFields( + * { + * partition: ['department'] + * }, + * average(field('salary')).as('departmentAverageSalary') + * ); + * + * // 2. Document-based moving average with explicit boundaries + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { + * sort: ascending('date'), + * documents: { preceding: 1, following: 1 } + * }, + * average(field('amount')).as('movingAverageAmount') + * ); + * + * // 3. Document-based running total ('unbounded' preceding to 'current' document position, excluding later ties). + * // Note: Offsets are physical document counts, so no time unit is required or used even when sorting on 'date'. + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { + * sort: ascending('date'), + * documents: { preceding: 'unbounded', following: 'current' } + * }, + * sum(field('amount')).as('runningTotal') + * ); + * + * // 4. Range-based running average using default boundaries ('unbounded' preceding to 'current', including all peers with tied sort values) + * firestore.pipeline().collection("products") + * .addWindowFields( + * { + * sort: ascending('price'), + * range: { preceding: 'unbounded', following: 'current' } + * }, + * average(field('rating')).as('cumulativeAvgRating') + * ); + * + * // 5. Range-based date/time window with a time unit (cumulative sales over the last 30 days) + * firestore.pipeline().collection("sales") + * .addWindowFields( + * { + * sort: ascending('date'), + * range: { preceding: 30, following: 'current', unit: 'day' } + * }, + * sum(field('amount')).as('thirtyDayCumulativeSales') + * ); + * ``` + * + * @param window - The specification defining how documents are partitioned, ordered, and bounded in the window frame. + * @param field - The first window field to add, specified as an {@link AliasedAggregate} or {@link AliasedWindowFunction}. + * @param additionalFields - Optional additional window fields to add to the documents. + * @returns A new Pipeline object with this stage appended to the stage list. + */ + addWindowFields( + window: WindowSpec, + field: AliasedAggregate | AliasedWindowFunction, + ...additionalFields: Array + ): Pipeline; + + /** + * Adds window function results to the output documents of the pipeline using options. + * + * @example + * ```typescript + * // 1. Unsorted partition/group aggregation using options + * firestore.pipeline().collection("employees") + * .addWindowFields({ + * window: { + * partition: ['department'] + * }, + * fields: [ + * average(field('salary')).as('departmentAverageSalary') + * ] + * }); + * + * // 2. Document-based moving average using options + * firestore.pipeline().collection("sales") + * .addWindowFields({ + * window: { + * sort: ascending('date'), + * documents: { preceding: 1, following: 1 } + * }, + * fields: [ + * average(field('amount')).as('movingAverageAmount') + * ] + * }); + * ``` + * + * @param options - An object specifying the window frame configuration and the fields to add. + * @returns A new Pipeline object with this stage appended to the stage list. + */ + addWindowFields(options: AddWindowFieldsStageOptions): Pipeline; /** * Remove fields from outputs of previous stages. * @@ -13912,6 +14120,141 @@ declare namespace FirebaseFirestore { */ fields: Selectable[]; }; + + /** + * Defines a document-count based window frame relative to the position of the current document in the group. + * + * Specifying `sort` on the enclosing {@link WindowSpec} is optional for `documents` window frames; if no `sort` expressions are specified, documents are processed in incoming stream (fetch) order. + */ + export interface DocumentWindowFrame { + /** + * The lower bound (inclusive) of the window frame, relative to the current document's position. + * + * Can be: + * - A number specifying a document offset relative to the current document. A positive integer (e.g. `2`) includes up to that many documents before the current document. A negative integer (e.g. `-1`) indicates a boundary following the current document, enabling frames that start after the current document. + * - `'current'` to represent the current document's position (documents with tied sort values are not automatically included). + * - `'unbounded'` to include all documents from the first document in the group. + */ + preceding: number | 'current' | 'unbounded' | Expression; + + /** + * The upper bound (inclusive) of the window frame, relative to the current document's position. + * + * Can be: + * - A number specifying a document offset relative to the current document. A positive integer (e.g. `2`) includes up to that many documents after the current document. A negative integer (e.g. `-1`) indicates a boundary preceding the current document, enabling frames that end before the current document (e.g. excluding the current document). + * - `'current'` to represent the current document's position (documents with tied sort values are not automatically included). + * - `'unbounded'` to include all documents to the last document in the group. + */ + following: number | 'current' | 'unbounded' | Expression; + } + + /** + * Defines a range-value based window frame relative to the sort value(s) of the current document. + * + * Sort requirements on the enclosing {@link WindowSpec}: + * - One or more `sort` expressions are required when specifying a `range` window frame. + * - If `preceding` and `following` use only `'current'` or `'unbounded'` (no numeric or time offsets), multiple `sort` expressions and non-numeric sort values (such as strings or booleans) are supported. + * - If `preceding` or `following` specifies a numeric offset (or a `unit` is specified for a time-based offset), exactly one `sort` expression is required and it must evaluate to a numeric or timestamp value. + */ + export interface RangeWindowFrame { + /** + * The lower bound (inclusive) of the window frame, relative to the sort value(s) of the current document. + * + * Can be: + * - A number specifying the value-based offset from the current document's sort value (requires a single numeric or timestamp `sort` expression). A positive offset subtracts from the current document's sort value (looking into the past). A negative offset adds to the current document's sort value (shifting the lower boundary past the current document). + * - `'current'` to include all documents with the same sort value(s) as the current document (peers/ties, equivalent to an offset of `0`). + * - `'unbounded'` to include all documents from the start of the group. + */ + preceding: number | 'current' | 'unbounded' | Expression; + + /** + * The upper bound (inclusive) of the window frame, relative to the sort value(s) of the current document. + * + * Can be: + * - A number specifying the value-based offset from the current document's sort value (requires a single numeric or timestamp `sort` expression). A positive offset adds to the current document's sort value (looking into the future). A negative offset subtracts from the current document's sort value (shifting the upper boundary before the current document). + * - `'current'` to include all documents with the same sort value(s) as the current document (peers/ties, equivalent to an offset of `0`). + * - `'unbounded'` to include all documents to the end of the group. + */ + following: number | 'current' | 'unbounded' | Expression; + + /** + * The unit used to calculate range boundaries when the `sort` field contains date or time values. + */ + unit?: + | 'microsecond' + | 'millisecond' + | 'second' + | 'minute' + | 'hour' + | 'day' + | 'week' + | 'month' + | 'quarter' + | 'year' + | Expression; + } + + /** + * A type that specifies a window frame, over which a window function will be evaluated. + * + * Default frame behavior: + * - If `sort` is not specified, the default frame is `documents` from `'unbounded'` preceding to `'unbounded'` following (the entire partition/group). + * - If `sort` is specified, the default frame is `range` from `'unbounded'` preceding to `'current'` (which includes all documents with the same sort value(s) as the current document). + */ + export type WindowSpec = { + /** + * Evaluate the window function over documents in the same group as the current document. Documents are grouped by having the same value for all provided {@link Expression}s and fields as specified in this `partition` array. If a string value is provided, it is treated as the name of a field in the document. If this value is left unspecified, then a single group is used for all input documents to the stage. + */ + partition?: Array; + + /** + * The sort order of the documents in each group. + * + * - For `documents` window frames, `sort` is optional. If no `sort` expressions are specified, documents are processed in incoming stream (fetch) order. + * - For `range` window frames whose bounds use only `'current'` or `'unbounded'`, one or more `sort` expressions are required (and non-numeric sort values such as strings or booleans are supported). + * - For `range` window frames with numeric or time-based offsets, exactly one `sort` expression (numeric or timestamp) is required. + * + * Setting a value for `sort` also changes the default window frame behavior when neither `documents` nor `range` is specified. + * See {@link WindowSpec} for default frame specifications. + * + * See {@link Ordering}. + */ + sort?: Ordering | Ordering[]; + } & OneOf<{ + /** + * Defines a document-count based window frame relative to the position of the current document in the group. + * + * Specifying `sort` is optional for `documents` window frames; if no `sort` expressions are specified, documents are processed in incoming stream (fetch) order. + * + * See {@link WindowSpec} for default frame specifications if `documents` or `range` is not set. + */ + documents?: DocumentWindowFrame; + + /** + * Defines a range-value based window frame relative to the sort value(s) of the current document. + * + * One or more `sort` expressions are required when specifying a `range` window frame (and numeric or time-based offsets require a single numeric or timestamp `sort` expression). + * + * See {@link WindowSpec} for default frame specifications if `documents` or `range` is not set. + */ + range?: RangeWindowFrame; + }>; + + /** + * Options defining how an AddWindowFieldsStage is evaluated. See {@link Pipeline.addWindowFields}. + */ + export type AddWindowFieldsStageOptions = StageOptions & { + /** + * The window spec to evaluate the window function over. + */ + window: WindowSpec; + + /** + * The fields to add to each document, specified as an {@link AliasedAggregate} or {@link AliasedWindowFunction}. + * At least one field is required. + */ + fields: Array; + }; /** * Options defining how a RemoveFieldsStage is evaluated. See {@link Pipeline.removeFields}. */