Skip to content

feat(firestore): add DML stages, literals source, and atomic execution option to Node SDK pipelines #9118

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wu-hui wants to merge 2 commits into
base: main
Choose a base branch
from
+624 −11
Open

feat(firestore): add DML stages, literals source, and atomic execution option to Node SDK pipelines #9118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ad72802
feat(firestore): add DML stages (insert, upsert, delete, update), lit…
wu-hui Aug 10, 2026
b8d4a66
test(firestore): refine Node SDK DML integration tests for backend ex…
wu-hui Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions handwritten/firestore/dev/protos/firestore_v1_proto_api.d.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2690,6 +2690,9 @@ export namespace google {

/** ExecutePipelineRequest readTime */
readTime?: (google.protobuf.ITimestamp|null);

/** ExecutePipelineRequest autoCommitTransaction */
autoCommitTransaction?: (boolean|null);
}

/** Represents an ExecutePipelineRequest. */
Expand All @@ -2716,6 +2719,9 @@ export namespace google {
/** ExecutePipelineRequest readTime. */
public readTime?: (google.protobuf.ITimestamp|null);

/** ExecutePipelineRequest autoCommitTransaction. */
public autoCommitTransaction?: (boolean|null);

/** ExecutePipelineRequest pipelineType. */
public pipelineType?: "structuredPipeline";

Expand Down
5 changes: 4 additions & 1 deletion handwritten/firestore/dev/src/pipelines/pipeline-util.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,7 +264,10 @@ export class ExecutionUtil {
structuredPipeline: structuredPipeline._toProto(this._serializer),
};

if (transactionOrReadTime instanceof Uint8Array) {
if (structuredPipeline.options?.atomic) {
request.newTransaction = {readWrite: {}};
request.autoCommitTransaction = true;
} else if (transactionOrReadTime instanceof Uint8Array) {
request.transaction = transactionOrReadTime;
} else if (transactionOrReadTime instanceof Timestamp) {
request.readTime = transactionOrReadTime.toProto().timestampValue;
Comment on lines +267 to 273

@gemini-code-assist gemini-code-assist Bot Aug 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

When executing a pipeline within an active transaction runner, the transaction ID (Uint8Array) must take precedence over the atomic option. If atomic is checked first, any pipeline with atomic: true will ignore the active transaction and execute as a separate, independent transaction, violating transaction isolation and atomicity. Reordering the checks ensures the active transaction is always respected.

        if (transactionOrReadTime instanceof Uint8Array) {
          request.transaction = transactionOrReadTime;
        } else if (structuredPipeline.options?.atomic) {
          request.newTransaction = {readWrite: {}};
          request.autoCommitTransaction = true;
        } else if (transactionOrReadTime instanceof Timestamp) {
          request.readTime = transactionOrReadTime.toProto().timestampValue;
        }

Expand Down
165 changes: 163 additions & 2 deletions handwritten/firestore/dev/src/pipelines/pipelines.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,12 @@ import {
UpdateStage,
Search,
InternalSearchStageOptions,
InsertStage,
InternalInsertStageOptions,
UpsertStage,
InternalUpsertStageOptions,
LiteralsSource,
InternalLiteralsStageOptions,
} from './stage';
import {StructuredPipeline} from './structured-pipeline';
import Selectable = FirebaseFirestore.Pipelines.Selectable;
Expand All @@ -125,6 +131,46 @@ import {
export class PipelineSource implements firestore.Pipelines.PipelineSource {
constructor(private db: Firestore) {}

/**
* Set the pipeline's source to the in-memory documents specified by the given records.
*
* @param documents An array of objects/records specifying the in-memory documents.
* @param options Options defining how this LiteralsSource stage is evaluated.
*/
literals(
documents: Array<Record<string, unknown>>,
options?: firestore.Pipelines.LiteralsStageOptions,
): Pipeline;
literals(
options: firestore.Pipelines.LiteralsStageOptions,
): Pipeline;
literals(
docsOrOptions:
| Array<Record<string, unknown>>
| firestore.Pipelines.LiteralsStageOptions,
options?: firestore.Pipelines.LiteralsStageOptions,
): Pipeline {
let documents: Array<Record<string, unknown>> = [];
let opts: InternalLiteralsStageOptions = {};

if (Array.isArray(docsOrOptions)) {
documents = docsOrOptions;
opts = options ?? {};
} else if (
docsOrOptions &&
Array.isArray(
(docsOrOptions as firestore.Pipelines.LiteralsStageOptions).documents,
)
) {
const {documents: docs, ...rest} =
docsOrOptions as firestore.Pipelines.LiteralsStageOptions;
documents = docs ?? [];
opts = rest;
}
Comment on lines +156 to +169

@gemini-code-assist gemini-code-assist Bot Aug 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

If docsOrOptions is passed as an options object without a documents array property (e.g., pipeline.literals({ rawOptions: ... })), the else if condition fails because Array.isArray(docsOrOptions.documents) is false. This causes the options to be silently ignored and opts to remain empty. Simplifying the condition to check if docsOrOptions is truthy and not an array ensures all options are correctly captured.

    if (Array.isArray(docsOrOptions)) {
      documents = docsOrOptions;
      opts = options ?? {};
    } else if (docsOrOptions) {
      const {documents: docs, ...rest} = 
        docsOrOptions as firestore.Pipelines.LiteralsStageOptions;
      documents = docs ?? [];
      opts = rest;
    }


return new Pipeline(this.db, [new LiteralsSource(documents, opts)]);
}

/**
* Returns all documents from the entire collection. The collection can be nested.
* @param collection - Name or reference to the collection that will be used as the Pipeline source.
Expand Down Expand Up @@ -1757,8 +1803,123 @@ export class Pipeline implements firestore.Pipelines.Pipeline {
* @return A new {@code Pipeline} object with this stage appended to the stage list.
*/
update(transformedFields: AliasedExpression[]): Pipeline;
update(transformedFields?: AliasedExpression[]): Pipeline {
return this._addStage(new UpdateStage(transformedFields));
/**
* @beta
* Performs an update operation using documents from previous stages.
*
* @param fieldsMap - Map of field transformations to apply.
* @return A new {@code Pipeline} object with this stage appended to the stage list.
*/
update(
fieldsMap: Record<string, Expression> | Map<string, Expression>,
): Pipeline;
update(
transformedFieldsOrMap?:
| AliasedExpression[]
| Map<string, Expression>
| Record<string, Expression>,
): Pipeline {
if (
transformedFieldsOrMap instanceof Map ||
Array.isArray(transformedFieldsOrMap)
) {
return this._addStage(new UpdateStage(transformedFieldsOrMap));
} else if (transformedFieldsOrMap && isPlainObject(transformedFieldsOrMap)) {
const map = new Map<string, Expression>(
Object.entries(transformedFieldsOrMap as Record<string, Expression>),
);
return this._addStage(new UpdateStage(map));
}
return this._addStage(new UpdateStage());
}

/**
* @beta
* Performs an insert operation on documents from previous stages.
*
* @param options - Options defining how this Insert stage is evaluated.
* @return A new {@code Pipeline} object with this stage appended to the stage list.
*/
insert(options?: firestore.Pipelines.InsertStageOptions): Pipeline {
return this._addStage(
new InsertStage((options ?? {}) as InternalInsertStageOptions),
);
}

/**
* @beta
* Performs an upsert operation on documents from previous stages.
*
* @param transforms - Transformations to apply on upsert.
* @param options - Options defining how this Upsert stage is evaluated.
* @return A new {@code Pipeline} object with this stage appended to the stage list.
*/
upsert(
transforms?: AliasedExpression[],
options?: Omit<firestore.Pipelines.UpsertStageOptions, 'transforms'>,
): Pipeline;
upsert(options?: firestore.Pipelines.UpsertStageOptions): Pipeline;
upsert(
transformsOrOptions?:
| AliasedExpression[]
| firestore.Pipelines.UpsertStageOptions,
options?: Omit<firestore.Pipelines.UpsertStageOptions, 'transforms'>,
): Pipeline {
let transforms: AliasedExpression[] = [];
let opts: InternalUpsertStageOptions = {};

if (Array.isArray(transformsOrOptions)) {
transforms = transformsOrOptions;
opts = (options ?? {}) as InternalUpsertStageOptions;
} else if (transformsOrOptions) {
const {transforms: t, ...rest} =
transformsOrOptions as firestore.Pipelines.UpsertStageOptions;
transforms = (t ?? []) as AliasedExpression[];
opts = rest as InternalUpsertStageOptions;
}

return this._addStage(new UpsertStage(transforms, opts));
}

/**
* Appends a literals stage to the pipeline.
*
* @param documents An array of objects/records specifying in-memory documents.
* @param options Options defining how this LiteralsSource stage is evaluated.
* @return A new {@code Pipeline} object with this stage appended to the stage list.
*/
literals(
documents: Array<Record<string, unknown>>,
options?: firestore.Pipelines.LiteralsStageOptions,
): Pipeline;
literals(
options: firestore.Pipelines.LiteralsStageOptions,
): Pipeline;
literals(
docsOrOptions:
| Array<Record<string, unknown>>
| firestore.Pipelines.LiteralsStageOptions,
options?: firestore.Pipelines.LiteralsStageOptions,
): Pipeline {
let documents: Array<Record<string, unknown>> = [];
let opts: InternalLiteralsStageOptions = {};

if (Array.isArray(docsOrOptions)) {
documents = docsOrOptions;
opts = options ?? {};
} else if (
docsOrOptions &&
Array.isArray(
(docsOrOptions as firestore.Pipelines.LiteralsStageOptions).documents,
)
) {
const {documents: docs, ...rest} =
docsOrOptions as firestore.Pipelines.LiteralsStageOptions;
documents = docs ?? [];
opts = rest;
}
Comment on lines +1907 to +1920

@gemini-code-assist gemini-code-assist Bot Aug 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

If docsOrOptions is passed as an options object without a documents array property (e.g., pipeline.literals({ rawOptions: ... })), the else if condition fails because Array.isArray(docsOrOptions.documents) is false. This causes the options to be silently ignored and opts to remain empty. Simplifying the condition to check if docsOrOptions is truthy and not an array ensures all options are correctly captured.

    if (Array.isArray(docsOrOptions)) {
      documents = docsOrOptions;
      opts = options ?? {};
    } else if (docsOrOptions) {
      const {documents: docs, ...rest} = 
        docsOrOptions as firestore.Pipelines.LiteralsStageOptions;
      documents = docs ?? [];
      opts = rest;
    }


return this._addStage(new LiteralsSource(documents, opts));
}

/**
Expand Down
Loading
Loading