From 39abeea81096f17248e9e10a5f2ee96646c3b873 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 26 Mar 2026 19:57:58 -0700
Subject: [PATCH 1/6] initial commit

---
 .../cloud/firestore_v1/base_pipeline.py       | 72 +++++++++++++++++++
 .../firestore_v1/pipeline_expressions.py      | 50 +++++++++++++
 .../cloud/firestore_v1/pipeline_source.py     | 22 ++++++
 .../cloud/firestore_v1/pipeline_stages.py     | 23 ++++++
 .../tests/system/pipeline_e2e/subqueries.yaml | 57 +++++++++++++++
 .../tests/system/test_pipeline_acceptance.py  |  6 ++
 6 files changed, 230 insertions(+)
 create mode 100644 packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml

diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py
index fac7f8bc4bce..615cf03e9b59 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py
@@ -25,6 +25,8 @@
     Expression,
     Field,
     Selectable,
+    FunctionExpression,
+    _PipelineValueExpression,
 )
 from google.cloud.firestore_v1.types.pipeline import (
     StructuredPipeline as StructuredPipeline_pb,
@@ -90,6 +92,51 @@ def _to_pb(self, **options) -> StructuredPipeline_pb:
             options=options,
         )
 
+    def to_array_expression(self) -> Expression:
+        """
+        Converts this Pipeline into an expression that evaluates to an array of results.
+        Used for embedding 1:N subqueries into stages like `addFields`.
+
+        Example:
+            >>> # Get a list of all reviewer names for each book
+            >>> db.pipeline().collection("books").define(Field.of("id").as_("book_id")).add_fields(
+            ...     db.pipeline().collection("reviews")
+            ...         .where(Field.of("book_id").equal(Variable("book_id")))
+            ...         .select(Field.of("reviewer").as_("name"))
+            ...         .to_array_expression().as_("reviewers")
+            ... )
+
+        Returns:
+            An :class:`Expression` representing the execution of this pipeline.
+        """
+        return FunctionExpression("array", [_PipelineValueExpression(self)])
+
+    def to_scalar_expression(self) -> Expression:
+        """
+        Converts this Pipeline into an expression that evaluates to a single scalar result.
+        Used for 1:1 lookups or Aggregations when the subquery is expected to return a single value or object.
+
+        **Result Unwrapping:**
+        For simpler access, scalar subqueries producing a single field automatically unwrap that value to the
+        top level, ignoring the inner alias. If the subquery returns multiple fields, they are preserved as a map.
+
+        Example:
+            >>> # Calculate average rating for each restaurant using a subquery
+            >>> db.pipeline().collection("restaurants").define(Field.of("id").as_("rid")).add_fields(
+            ...     db.pipeline().collection("reviews")
+            ...         .where(Field.of("restaurant_id").equal(Variable("rid")))
+            ...         .aggregate(AggregateFunction.average("rating").as_("value"))
+            ...         .to_scalar_expression().as_("average_rating")
+            ... )
+
+        Raises:
+            RuntimeError: If the result set contains more than one item. If the pipeline has zero results, it evaluates to `null` instead of raising an error.
+
+        Returns:
+            An :class:`Expression` representing the execution of this pipeline.
+        """
+        return FunctionExpression("scalar", [_PipelineValueExpression(self)])
+
     def _append(self, new_stage):
         """
         Create a new Pipeline object with a new stage appended
@@ -610,3 +657,28 @@ def distinct(self, *fields: str | Selectable) -> "_BasePipeline":
             A new Pipeline object with this stage appended to the stage list
         """
         return self._append(stages.Distinct(*fields))
+
+    def define(self, *aliased_expressions: AliasedExpression) -> "_BasePipeline":
+        """
+        Binds one or more expressions to Variables that can be accessed in subsequent stages
+        or inner subqueries using `Variable`.
+
+        Each Variable is defined using an :class:`AliasedExpression`, which pairs an expression with
+        a name (alias).
+
+        Example:
+            >>> db.pipeline().collection("products").define(
+            ...     Field.of("price").multiply(0.9).as_("discountedPrice"),
+            ...     Field.of("stock").add(10).as_("newStock")
+            ... ).where(
+            ...     Variable("discountedPrice").less_than(100)
+            ... ).select(Field.of("name"), Variable("newStock"))
+
+        Args:
+            *aliased_expressions: One or more :class:`AliasedExpression` defining the Variable names and values.
+
+        Returns:
+            A new Pipeline object with this stage appended to the stage list.
+        """
+        return self._append(stages.Define(*aliased_expressions))
+
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
index 1833d2ffbac0..b58ae6834b41 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
@@ -2634,3 +2634,53 @@ class Rand(FunctionExpression):
 
     def __init__(self):
         super().__init__("rand", [], use_infix_repr=False)
+
+
+class Variable(Expression):
+    """
+    Creates an expression that retrieves the value of a variable bound via `Pipeline.define`.
+
+    Example:
+        >>> # Define a variable "discountedPrice" and use it in a filter
+        >>> db.pipeline().collection("products").define(
+        ...     Field.of("price").multiply(0.9).as_("discountedPrice")
+        ... ).where(Variable("discountedPrice").less_than(100))
+
+    Args:
+        name: The name of the variable to retrieve.
+    """
+
+    def __init__(self, name: str):
+        self.name = name
+
+    def _to_pb(self) -> Value:
+        return Value(variable_reference_value=self.name)
+
+
+class _PipelineValueExpression(Expression):
+    """Internal wrapper to represent a pipeline as an expression."""
+
+    def __init__(self, pipeline):
+        self.pipeline = pipeline
+
+    def _to_pb(self) -> Value:
+        return Value(pipeline_value=self.pipeline._to_pb())
+
+
+class CurrentDocument(FunctionExpression):
+    """
+    Creates an expression that represents the current document being processed.
+
+    This acts as a handle, allowing you to bind the entire document to a variable or pass the
+    document itself to a function or subquery.
+
+    Example:
+        >>> # Define the current document as a variable "doc"
+        >>> db.pipeline().collection("books").define(
+        ...     CurrentDocument().as_("doc")
+        ... ).select(Variable("doc").get_field("title"))
+    """
+
+    def __init__(self):
+        super().__init__("current_document", [])
+
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py
index 7075797b3d57..a8ee5230dfcb 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py
@@ -171,3 +171,25 @@ def literals(
             A new Pipeline object with this stage appended to the stage list.
         """
         return self._create_pipeline(stages.Literals(*documents))
+
+    def subcollection(self, path: str) -> PipelineType:
+        """
+        Creates a new Pipeline targeted at a subcollection relative to the current document context.
+
+        This is used inside stages like `addFields` to query physically nested subcollections
+        without manually joining on IDs.
+
+        Example:
+            >>> db.pipeline().collection("books").add_fields(
+            ...     db.pipeline().subcollection("reviews")
+            ...         .aggregate(AggregateFunction.average("rating").as_("avg_rating"))
+            ...         .to_scalar_expression().as_("average_rating")
+            ... )
+
+        Args:
+            path: The path of the subcollection.
+
+        Returns:
+            A new :class:`Pipeline` instance scoped to the subcollection.
+        """
+        return self._create_pipeline(stages.Subcollection(path))
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py
index cac9c70d4b99..d1f5a459833d 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py
@@ -494,3 +494,26 @@ def __init__(self, condition: BooleanExpression):
 
     def _pb_args(self):
         return [self.condition._to_pb()]
+
+
+class Define(Stage):
+    """Binds one or more expressions to variables."""
+
+    def __init__(self, *expressions: AliasedExpression):
+        super().__init__("let")
+        self.expressions = list(expressions)
+
+    def _pb_args(self) -> list[Value]:
+        return [Selectable._to_value(self.expressions)]
+
+
+class Subcollection(Stage):
+    """Targets a subcollection relative to the current document."""
+
+    def __init__(self, path: str):
+        super().__init__("subcollection")
+        self.path = path
+
+    def _pb_args(self) -> list[Value]:
+        return [encode_value(self.path)]
+
diff --git a/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml b/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
new file mode 100644
index 000000000000..3228bbf998e5
--- /dev/null
+++ b/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
@@ -0,0 +1,57 @@
+tests:
+  - description: array_subquery_with_variable
+    pipeline:
+      - Collection: publishers
+      - Where:
+          - Field.of.equal:
+              - publisherId
+              - pub1
+      - Define:
+          - Field.of.as_:
+              - publisherId
+              - pub_id
+      - AddFields:
+          - Pipeline.to_array_expression:
+              - Collection: books
+              - Where:
+                  - Field.of.equal:
+                      - publisherId
+                      - Variable: pub_id
+            as_: books
+      - Select:
+          - Field: name
+          - Field: books
+    assert_results:
+      - name: Publisher 1
+        books:
+          - title: The Hitchhiker's Guide to the Galaxy
+            author: Douglas Adams
+          - title: Pride and Prejudice
+            author: Jane Austen
+
+  - description: scalar_subquery_with_current_document
+    pipeline:
+      - Collection: books
+      - Where:
+          - Field.of.equal:
+              - title
+              - 1984
+      - Define:
+          - CurrentDocument.as_: doc
+      - AddFields:
+          - Pipeline.to_scalar_expression:
+              - Collection: reviews
+              - Where:
+                  - Field.of.equal:
+                      - bookId
+                      - Variable.get_field:
+                          - doc
+                          - __name__
+              - Aggregate:
+                  - AggregateFunction.average.as_:
+                      - rating
+                      - avg_rating
+            as_: average_rating
+    assert_results:
+      - title: 1984
+        average_rating: 4.5
diff --git a/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py b/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py
index f1fd1326c765..9023ea389aab 100644
--- a/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py
+++ b/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py
@@ -258,6 +258,12 @@ def _parse_expressions(client, yaml_element: Any):
             # find Pipeline objects for Union expressions
             other_ppl = yaml_element["Pipeline"]
             return parse_pipeline(client, other_ppl)
+        elif len(yaml_element) == 1 and list(yaml_element)[0] == "Pipeline.to_array_expression":
+            other_ppl = yaml_element["Pipeline.to_array_expression"]
+            return parse_pipeline(client, other_ppl).to_array_expression()
+        elif len(yaml_element) == 1 and list(yaml_element)[0] == "Pipeline.to_scalar_expression":
+            other_ppl = yaml_element["Pipeline.to_scalar_expression"]
+            return parse_pipeline(client, other_ppl).to_scalar_expression()
         else:
             # otherwise, return dict
             return {

From 479d1cd7401afd986d399662d924530682699a14 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 26 Mar 2026 20:06:19 -0700
Subject: [PATCH 2/6] update docstring

---
 .../google/cloud/firestore_v1/pipeline_source.py         | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py
index a8ee5230dfcb..5b16f7b67926 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_source.py
@@ -174,10 +174,13 @@ def literals(
 
     def subcollection(self, path: str) -> PipelineType:
         """
-        Creates a new Pipeline targeted at a subcollection relative to the current document context.
+        Initializes a pipeline scoped to a subcollection.
 
-        This is used inside stages like `addFields` to query physically nested subcollections
-        without manually joining on IDs.
+        This method allows you to start a new pipeline that operates on a subcollection of the
+        current document. It is intended to be used as a subquery.
+
+        **Note:** A pipeline created with `subcollection` cannot be executed directly using
+        `execute()`. It must be used within a parent pipeline.
 
         Example:
             >>> db.pipeline().collection("books").add_fields(

From 351bd022c65a1f2083574a7e08ef359b36e5d95e Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 26 Mar 2026 21:22:36 -0700
Subject: [PATCH 3/6] fixed formatting

---
 .../cloud/firestore_v1/base_pipeline.py       |  1 -
 .../firestore_v1/pipeline_expressions.py      |  6 +-
 .../cloud/firestore_v1/pipeline_stages.py     |  1 -
 .../tests/system/pipeline_e2e/subqueries.yaml | 90 ++++++++++---------
 .../tests/system/test_pipeline_acceptance.py  | 12 ++-
 5 files changed, 62 insertions(+), 48 deletions(-)

diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py
index 615cf03e9b59..328e64155155 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/base_pipeline.py
@@ -681,4 +681,3 @@ def define(self, *aliased_expressions: AliasedExpression) -> "_BasePipeline":
             A new Pipeline object with this stage appended to the stage list.
         """
         return self._append(stages.Define(*aliased_expressions))
-
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
index b58ae6834b41..a28ddeffc390 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
@@ -30,7 +30,7 @@
 )
 
 from google.cloud.firestore_v1._helpers import GeoPoint, decode_value, encode_value
-from google.cloud.firestore_v1.types.document import Value
+from google.cloud.firestore_v1.types.document import Value, Pipeline
 from google.cloud.firestore_v1.types.query import StructuredQuery as Query_pb
 from google.cloud.firestore_v1.vector import Vector
 
@@ -2664,7 +2664,8 @@ def __init__(self, pipeline):
         self.pipeline = pipeline
 
     def _to_pb(self) -> Value:
-        return Value(pipeline_value=self.pipeline._to_pb())
+        pipeline_pb = Pipeline(stages=[s._to_pb() for s in self.pipeline.stages])
+        return Value(pipeline_value=pipeline_pb)
 
 
 class CurrentDocument(FunctionExpression):
@@ -2683,4 +2684,3 @@ class CurrentDocument(FunctionExpression):
 
     def __init__(self):
         super().__init__("current_document", [])
-
diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py
index d1f5a459833d..7a2d8d3b08af 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_stages.py
@@ -516,4 +516,3 @@ def __init__(self, path: str):
 
     def _pb_args(self) -> list[Value]:
         return [encode_value(self.path)]
-
diff --git a/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml b/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
index 3228bbf998e5..ac2c854b84e8 100644
--- a/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
+++ b/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
@@ -1,57 +1,65 @@
 tests:
   - description: array_subquery_with_variable
     pipeline:
-      - Collection: publishers
+      - Collection: books
       - Where:
-          - Field.of.equal:
-              - publisherId
-              - pub1
+          - FunctionExpression.equal:
+              - Field: title
+              - Constant: "1984"
       - Define:
-          - Field.of.as_:
-              - publisherId
-              - pub_id
+          - AliasedExpression:
+              - Field: genre
+              - target_genre
       - AddFields:
-          - Pipeline.to_array_expression:
-              - Collection: books
-              - Where:
-                  - Field.of.equal:
-                      - publisherId
-                      - Variable: pub_id
-            as_: books
+          - AliasedExpression:
+              - Pipeline.to_array_expression:
+                  - Collection: books
+                  - Where:
+                      - FunctionExpression.equal:
+                          - Field: genre
+                          - Variable: target_genre
+                  - Select:
+                      - Field: title
+              - same_genre_books
       - Select:
-          - Field: name
-          - Field: books
+          - Field: title
+          - Field: same_genre_books
     assert_results:
-      - name: Publisher 1
-        books:
-          - title: The Hitchhiker's Guide to the Galaxy
-            author: Douglas Adams
-          - title: Pride and Prejudice
-            author: Jane Austen
+      - title: "1984"
+        same_genre_books:
+          - "1984"
+          - "The Handmaid's Tale"
 
   - description: scalar_subquery_with_current_document
     pipeline:
       - Collection: books
       - Where:
-          - Field.of.equal:
-              - title
-              - 1984
+          - FunctionExpression.equal:
+              - Field: title
+              - Constant: "1984"
       - Define:
-          - CurrentDocument.as_: doc
+          - AliasedExpression:
+              - CurrentDocument:
+              - doc
       - AddFields:
-          - Pipeline.to_scalar_expression:
-              - Collection: reviews
-              - Where:
-                  - Field.of.equal:
-                      - bookId
-                      - Variable.get_field:
-                          - doc
-                          - __name__
-              - Aggregate:
-                  - AggregateFunction.average.as_:
-                      - rating
-                      - avg_rating
-            as_: average_rating
+          - AliasedExpression:
+              - Pipeline.to_scalar_expression:
+                  - Collection: books
+                  - Where:
+                      - FunctionExpression.equal:
+                          - Field: genre
+                          - FunctionExpression.map_get:
+                              - Variable: doc
+                              - Constant: "genre"
+                  - Aggregate:
+                      - AliasedExpression:
+                          - FunctionExpression.average:
+                              - Field: rating
+                          - avg_rating
+              - average_rating
+      - Select:
+          - Field: title
+          - Field: average_rating
     assert_results:
-      - title: 1984
-        average_rating: 4.5
+      - title: "1984"
+        average_rating: 4.15
diff --git a/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py b/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py
index 9023ea389aab..81bad2e8c3cf 100644
--- a/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py
+++ b/packages/google-cloud-firestore/tests/system/test_pipeline_acceptance.py
@@ -258,10 +258,16 @@ def _parse_expressions(client, yaml_element: Any):
             # find Pipeline objects for Union expressions
             other_ppl = yaml_element["Pipeline"]
             return parse_pipeline(client, other_ppl)
-        elif len(yaml_element) == 1 and list(yaml_element)[0] == "Pipeline.to_array_expression":
+        elif (
+            len(yaml_element) == 1
+            and list(yaml_element)[0] == "Pipeline.to_array_expression"
+        ):
             other_ppl = yaml_element["Pipeline.to_array_expression"]
             return parse_pipeline(client, other_ppl).to_array_expression()
-        elif len(yaml_element) == 1 and list(yaml_element)[0] == "Pipeline.to_scalar_expression":
+        elif (
+            len(yaml_element) == 1
+            and list(yaml_element)[0] == "Pipeline.to_scalar_expression"
+        ):
             other_ppl = yaml_element["Pipeline.to_scalar_expression"]
             return parse_pipeline(client, other_ppl).to_scalar_expression()
         else:
@@ -292,6 +298,8 @@ def _apply_yaml_args_to_callable(callable_obj, client, yaml_args):
     ):
         # yaml has an array of arguments. Treat as args
         return callable_obj(*_parse_expressions(client, yaml_args))
+    elif yaml_args is None:
+        return callable_obj()
     else:
         # yaml has a single argument
         return callable_obj(_parse_expressions(client, yaml_args))

From b8651561df6a19c5752739b1154ad5e61b8cb604 Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 26 Mar 2026 21:45:21 -0700
Subject: [PATCH 4/6] added assert proto blocks

---
 .../tests/system/pipeline_e2e/subqueries.yaml | 115 ++++++++++++++++++
 1 file changed, 115 insertions(+)

diff --git a/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml b/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
index ac2c854b84e8..7eed48ea8e73 100644
--- a/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
+++ b/packages/google-cloud-firestore/tests/system/pipeline_e2e/subqueries.yaml
@@ -29,6 +29,59 @@ tests:
         same_genre_books:
           - "1984"
           - "The Handmaid's Tale"
+    assert_proto:
+      pipeline:
+        stages:
+        - args:
+          - referenceValue: /books
+          name: collection
+        - args:
+          - functionValue:
+              args:
+              - fieldReferenceValue: title
+              - stringValue: '1984'
+              name: equal
+          name: where
+        - args:
+          - mapValue:
+              fields:
+                target_genre:
+                  fieldReferenceValue: genre
+          name: let
+        - args:
+          - mapValue:
+              fields:
+                same_genre_books:
+                  functionValue:
+                    args:
+                    - pipelineValue:
+                        stages:
+                        - args:
+                          - referenceValue: /books
+                          name: collection
+                        - args:
+                          - functionValue:
+                              args:
+                              - fieldReferenceValue: genre
+                              - variableReferenceValue: target_genre
+                              name: equal
+                          name: where
+                        - args:
+                          - mapValue:
+                              fields:
+                                title:
+                                  fieldReferenceValue: title
+                          name: select
+                    name: array
+          name: add_fields
+        - args:
+          - mapValue:
+              fields:
+                same_genre_books:
+                  fieldReferenceValue: same_genre_books
+                title:
+                  fieldReferenceValue: title
+          name: select
 
   - description: scalar_subquery_with_current_document
     pipeline:
@@ -63,3 +116,65 @@ tests:
     assert_results:
       - title: "1984"
         average_rating: 4.15
+    assert_proto:
+      pipeline:
+        stages:
+        - args:
+          - referenceValue: /books
+          name: collection
+        - args:
+          - functionValue:
+              args:
+              - fieldReferenceValue: title
+              - stringValue: '1984'
+              name: equal
+          name: where
+        - args:
+          - mapValue:
+              fields:
+                doc:
+                  functionValue:
+                    name: current_document
+          name: let
+        - args:
+          - mapValue:
+              fields:
+                average_rating:
+                  functionValue:
+                    args:
+                    - pipelineValue:
+                        stages:
+                        - args:
+                          - referenceValue: /books
+                          name: collection
+                        - args:
+                          - functionValue:
+                              args:
+                              - fieldReferenceValue: genre
+                              - functionValue:
+                                  args:
+                                  - variableReferenceValue: doc
+                                  - stringValue: genre
+                                  name: map_get
+                              name: equal
+                          name: where
+                        - args:
+                          - mapValue:
+                              fields:
+                                avg_rating:
+                                  functionValue:
+                                    args:
+                                    - fieldReferenceValue: rating
+                                    name: average
+                          - mapValue: {}
+                          name: aggregate
+                    name: scalar
+          name: add_fields
+        - args:
+          - mapValue:
+              fields:
+                average_rating:
+                  fieldReferenceValue: average_rating
+                title:
+                  fieldReferenceValue: title
+          name: select

From 6f37a5b763331ec21f6bdd50317ec41132219f0c Mon Sep 17 00:00:00 2001
From: Daniel Sanche <d.sanche14@gmail.com>
Date: Thu, 26 Mar 2026 21:53:47 -0700
Subject: [PATCH 5/6] fixed bug in tests

---
 .../google/cloud/firestore_v1/pipeline_expressions.py        | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
index a28ddeffc390..f119c091d424 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
@@ -30,7 +30,8 @@
 )
 
 from google.cloud.firestore_v1._helpers import GeoPoint, decode_value, encode_value
-from google.cloud.firestore_v1.types.document import Value, Pipeline
+from google.cloud.firestore_v1.types.document import Value
+from google.cloud.firestore_v1.types.document import Pipeline as Pipeline_pb
 from google.cloud.firestore_v1.types.query import StructuredQuery as Query_pb
 from google.cloud.firestore_v1.vector import Vector
 
@@ -2664,7 +2665,7 @@ def __init__(self, pipeline):
         self.pipeline = pipeline
 
     def _to_pb(self) -> Value:
-        pipeline_pb = Pipeline(stages=[s._to_pb() for s in self.pipeline.stages])
+        pipeline_pb = Pipeline_pb(stages=[s._to_pb() for s in self.pipeline.stages])
         return Value(pipeline_value=pipeline_pb)
 
 

From d94d96771675418918895546b8467965f8bc97db Mon Sep 17 00:00:00 2001
From: Daniel Sanche <sanche@google.com>
Date: Thu, 26 Mar 2026 22:07:07 -0700
Subject: [PATCH 6/6] Apply suggestion from @gemini-code-assist[bot]

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 .../google/cloud/firestore_v1/pipeline_expressions.py           | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
index f119c091d424..7bd5f6a60691 100644
--- a/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
+++ b/packages/google-cloud-firestore/google/cloud/firestore_v1/pipeline_expressions.py
@@ -2661,7 +2661,7 @@ def _to_pb(self) -> Value:
 class _PipelineValueExpression(Expression):
     """Internal wrapper to represent a pipeline as an expression."""
 
-    def __init__(self, pipeline):
+    def __init__(self, pipeline: "google.cloud.firestore_v1.base_pipeline._BasePipeline"):
         self.pipeline = pipeline
 
     def _to_pb(self) -> Value: