diff --git a/Firestore/Source/API/FIRPipelineBridge+Internal.h b/Firestore/Source/API/FIRPipelineBridge+Internal.h
index bfe7befe923..30bee14aa02 100644
--- a/Firestore/Source/API/FIRPipelineBridge+Internal.h
+++ b/Firestore/Source/API/FIRPipelineBridge+Internal.h
@@ -19,6 +19,7 @@
 #include <memory>
 
 #include "Firestore/core/src/api/expressions.h"
+#include "Firestore/core/src/api/firestore.h"
 #include "Firestore/core/src/api/pipeline.h"
 #include "Firestore/core/src/api/stages.h"
 
@@ -30,13 +31,13 @@ NS_ASSUME_NONNULL_BEGIN
 
 @interface FIRExprBridge (Internal)
 
-- (std::shared_ptr<api::Expr>)cpp_expr;
+- (std::shared_ptr<api::Expr>)cppExprWithReader:(FSTUserDataReader *)reader;
 
 @end
 
 @interface FIRStageBridge (Internal)
 
-- (std::shared_ptr<api::Stage>)cpp_stage;
+- (std::shared_ptr<api::Stage>)cppStageWithReader:(FSTUserDataReader *)reader;
 
 @end
 
@@ -46,4 +47,10 @@ NS_ASSUME_NONNULL_BEGIN
 
 @end
 
+@interface __FIRPipelineResultBridge (Internal)
+
+- (id)initWithCppResult:(api::PipelineResult)result db:(std::shared_ptr<api::Firestore>)db;
+
+@end
+
 NS_ASSUME_NONNULL_END
diff --git a/Firestore/Source/API/FIRPipelineBridge.mm b/Firestore/Source/API/FIRPipelineBridge.mm
index 013cedcccab..c10a05f4d88 100644
--- a/Firestore/Source/API/FIRPipelineBridge.mm
+++ b/Firestore/Source/API/FIRPipelineBridge.mm
@@ -18,9 +18,15 @@
 
 #include <memory>
 
+#import "Firestore/Source/API/FIRDocumentReference+Internal.h"
 #import "Firestore/Source/API/FIRFirestore+Internal.h"
 #import "Firestore/Source/API/FIRPipelineBridge+Internal.h"
+#import "Firestore/Source/API/FSTUserDataReader.h"
+#import "Firestore/Source/API/FSTUserDataWriter.h"
 
+#include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
+
+#include "Firestore/core/src/api/document_reference.h"
 #include "Firestore/core/src/api/expressions.h"
 #include "Firestore/core/src/api/pipeline.h"
 #include "Firestore/core/src/api/pipeline_result.h"
@@ -32,12 +38,14 @@
 
 using firebase::firestore::api::CollectionSource;
 using firebase::firestore::api::Constant;
+using firebase::firestore::api::DocumentReference;
 using firebase::firestore::api::Expr;
 using firebase::firestore::api::Field;
 using firebase::firestore::api::FunctionExpr;
 using firebase::firestore::api::Pipeline;
 using firebase::firestore::api::Where;
 using firebase::firestore::util::MakeCallback;
+using firebase::firestore::util::MakeNSString;
 using firebase::firestore::util::MakeString;
 
 NS_ASSUME_NONNULL_BEGIN
@@ -57,7 +65,7 @@ - (id)init:(NSString *)name {
   return self;
 }
 
-- (std::shared_ptr<api::Expr>)cpp_expr {
+- (std::shared_ptr<api::Expr>)cppExprWithReader:(FSTUserDataReader *)reader {
   return field;
 }
 
@@ -65,16 +73,22 @@ - (id)init:(NSString *)name {
 
 @implementation FIRConstantBridge {
   std::shared_ptr<Constant> constant;
+  id _input;
+  Boolean isUserDataRead;
 }
-- (id)init:(NSNumber *)value {
+- (id)init:(id)input {
   self = [super init];
-  if (self) {
-    constant = std::make_shared<Constant>(value.doubleValue);
-  }
+  _input = input;
+  isUserDataRead = NO;
   return self;
 }
 
-- (std::shared_ptr<api::Expr>)cpp_expr {
+- (std::shared_ptr<api::Expr>)cppExprWithReader:(FSTUserDataReader *)reader {
+  if (!isUserDataRead) {
+    constant = std::make_shared<Constant>([reader parsedQueryValue:_input]);
+  }
+
+  isUserDataRead = YES;
   return constant;
 }
 
@@ -82,22 +96,29 @@ - (id)init:(NSNumber *)value {
 
 @implementation FIRFunctionExprBridge {
   std::shared_ptr<FunctionExpr> eq;
+  NSString *_name;
+  NSArray<FIRExprBridge *> *_args;
+  Boolean isUserDataRead;
 }
 
 - (nonnull id)initWithName:(NSString *)name Args:(nonnull NSArray<FIRExprBridge *> *)args {
   self = [super init];
-  if (self) {
+  _name = name;
+  _args = args;
+  isUserDataRead = NO;
+  return self;
+}
+
+- (std::shared_ptr<api::Expr>)cppExprWithReader:(FSTUserDataReader *)reader {
+  if (!isUserDataRead) {
     std::vector<std::shared_ptr<Expr>> cpp_args;
-    for (FIRExprBridge *arg in args) {
-      cpp_args.push_back(arg.cpp_expr);
+    for (FIRExprBridge *arg in _args) {
+      cpp_args.push_back([arg cppExprWithReader:reader]);
     }
-
-    eq = std::make_shared<FunctionExpr>(MakeString(name), std::move(cpp_args));
+    eq = std::make_shared<FunctionExpr>(MakeString(_name), std::move(cpp_args));
   }
-  return self;
-}
 
-- (std::shared_ptr<api::Expr>)cpp_expr {
+  isUserDataRead = YES;
   return eq;
 }
 
@@ -118,63 +139,142 @@ - (id)initWithPath:(NSString *)path {
   return self;
 }
 
-- (std::shared_ptr<api::Stage>)cpp_stage {
+- (std::shared_ptr<api::Stage>)cppStageWithReader:(FSTUserDataReader *)reader {
   return collection_source;
 }
 
 @end
 
 @implementation FIRWhereStageBridge {
+  FIRExprBridge *_exprBridge;
+  Boolean isUserDataRead;
   std::shared_ptr<Where> where;
 }
 
 - (id)initWithExpr:(FIRExprBridge *)expr {
   self = [super init];
   if (self) {
-    where = std::make_shared<Where>(expr.cpp_expr);
+    _exprBridge = expr;
+    isUserDataRead = NO;
   }
   return self;
 }
 
-- (std::shared_ptr<api::Stage>)cpp_stage {
+- (std::shared_ptr<api::Stage>)cppStageWithReader:(FSTUserDataReader *)reader {
+  if (!isUserDataRead) {
+    where = std::make_shared<Where>([_exprBridge cppExprWithReader:reader]);
+  }
+
+  isUserDataRead = YES;
   return where;
 }
 
 @end
 
+@interface __FIRPipelineSnapshotBridge ()
+
+@property(nonatomic, strong, readwrite) NSArray<__FIRPipelineSnapshotBridge *> *results;
+
+@end
+
 @implementation __FIRPipelineSnapshotBridge {
-  absl::optional<api::PipelineSnapshot> pipeline;
+  absl::optional<api::PipelineSnapshot> snapshot_;
+  NSMutableArray<__FIRPipelineResultBridge *> *results_;
 }
 
 - (id)initWithCppSnapshot:(api::PipelineSnapshot)snapshot {
   self = [super init];
   if (self) {
-    pipeline = std::move(snapshot);
+    snapshot_ = std::move(snapshot);
+    if (!snapshot_.has_value()) {
+      results_ = nil;
+    } else {
+      NSMutableArray<__FIRPipelineResultBridge *> *results = [NSMutableArray array];
+      for (auto &result : snapshot_.value().results()) {
+        [results addObject:[[__FIRPipelineResultBridge alloc]
+                               initWithCppResult:result
+                                              db:snapshot_.value().firestore()]];
+      }
+      results_ = results;
+    }
   }
 
   return self;
 }
 
+- (NSArray<__FIRPipelineResultBridge *> *)results {
+  return results_;
+}
+
 @end
 
-@implementation FIRPipelineBridge {
-  std::shared_ptr<Pipeline> pipeline;
+@implementation __FIRPipelineResultBridge {
+  api::PipelineResult _result;
+  std::shared_ptr<api::Firestore> _db;
 }
 
-- (id)initWithStages:(NSArray<FIRStageBridge *> *)stages db:(FIRFirestore *)db {
+- (FIRDocumentReference *)reference {
+  if (!_result.internal_key().has_value()) return nil;
+
+  return [[FIRDocumentReference alloc] initWithKey:_result.internal_key().value() firestore:_db];
+}
+
+- (NSString *)documentID {
+  if (!_result.document_id().has_value()) {
+    return nil;
+  }
+
+  return MakeNSString(_result.document_id().value());
+}
+
+- (id)initWithCppResult:(api::PipelineResult)result db:(std::shared_ptr<api::Firestore>)db {
   self = [super init];
   if (self) {
-    std::vector<std::shared_ptr<firebase::firestore::api::Stage>> cpp_stages;
-    for (FIRStageBridge *stage in stages) {
-      cpp_stages.push_back(stage.cpp_stage);
-    }
-    pipeline = std::make_shared<Pipeline>(cpp_stages, db.wrapped);
+    _result = std::move(result);
+    _db = std::move(db);
   }
+
   return self;
 }
 
+- (nullable NSDictionary<NSString *, id> *)data {
+  return [self dataWithServerTimestampBehavior:FIRServerTimestampBehaviorNone];
+}
+
+- (nullable NSDictionary<NSString *, id> *)dataWithServerTimestampBehavior:
+    (FIRServerTimestampBehavior)serverTimestampBehavior {
+  absl::optional<firebase::firestore::google_firestore_v1_Value> data =
+      _result.internal_value()->Get();
+  if (!data) return nil;
+
+  FSTUserDataWriter *dataWriter =
+      [[FSTUserDataWriter alloc] initWithFirestore:_db
+                           serverTimestampBehavior:serverTimestampBehavior];
+  return [dataWriter convertedValue:*data];
+}
+
+@end
+
+@implementation FIRPipelineBridge {
+  NSArray<FIRStageBridge *> *_stages;
+  FIRFirestore *firestore;
+  std::shared_ptr<Pipeline> pipeline;
+}
+
+- (id)initWithStages:(NSArray<FIRStageBridge *> *)stages db:(FIRFirestore *)db {
+  _stages = stages;
+  firestore = db;
+  return [super init];
+}
+
 - (void)executeWithCompletion:(void (^)(__FIRPipelineSnapshotBridge *_Nullable result,
                                         NSError *_Nullable error))completion {
+  std::vector<std::shared_ptr<firebase::firestore::api::Stage>> cpp_stages;
+  for (FIRStageBridge *stage in _stages) {
+    cpp_stages.push_back([stage cppStageWithReader:firestore.dataReader]);
+  }
+  pipeline = std::make_shared<Pipeline>(cpp_stages, firestore.wrapped);
+
   pipeline->execute([completion](StatusOr<api::PipelineSnapshot> maybe_value) {
     if (maybe_value.ok()) {
       __FIRPipelineSnapshotBridge *bridge = [[__FIRPipelineSnapshotBridge alloc]
diff --git a/Firestore/Source/Public/FirebaseFirestore/FIRPipelineBridge.h b/Firestore/Source/Public/FirebaseFirestore/FIRPipelineBridge.h
index fa7472e3292..a27b2b7aa18 100644
--- a/Firestore/Source/Public/FirebaseFirestore/FIRPipelineBridge.h
+++ b/Firestore/Source/Public/FirebaseFirestore/FIRPipelineBridge.h
@@ -18,6 +18,8 @@
 
 #import <Foundation/Foundation.h>
 
+#import "FIRDocumentSnapshot.h"
+
 NS_ASSUME_NONNULL_BEGIN
 
 NS_SWIFT_NAME(ExprBridge)
@@ -31,7 +33,7 @@ NS_SWIFT_NAME(FieldBridge)
 
 NS_SWIFT_NAME(ConstantBridge)
 @interface FIRConstantBridge : FIRExprBridge
-- (id)init:(NSNumber *)value;
+- (id)init:(id)input;
 @end
 
 NS_SWIFT_NAME(FunctionExprBridge)
@@ -72,6 +74,8 @@ NS_SWIFT_NAME(__PipelineResultBridge)
 @property(nonatomic, copy, readonly) NSString *documentID;
 
 - (nullable NSDictionary<NSString *, id> *)data;
+- (nullable NSDictionary<NSString *, id> *)dataWithServerTimestampBehavior:
+    (FIRServerTimestampBehavior)serverTimestampBehavior;
 
 @end
 
diff --git a/Firestore/Swift/Source/SwiftAPI/Expressions.swift b/Firestore/Swift/Source/SwiftAPI/Expressions.swift
index 729b5c9fb67..22af7ae6471 100644
--- a/Firestore/Swift/Source/SwiftAPI/Expressions.swift
+++ b/Firestore/Swift/Source/SwiftAPI/Expressions.swift
@@ -16,26 +16,28 @@
 
 import Foundation
 
-public protocol Expr {
+public protocol Expr {}
+
+protocol BridgeWrapper {
   var bridge: ExprBridge { get }
 }
 
-public struct Constant: Expr {
-  public var bridge: ExprBridge
+public struct Constant: Expr, BridgeWrapper {
+  var bridge: ExprBridge
 
-  var value: any Numeric
-  init(value: any Numeric) {
+  var value: Any
+  init(value: Any) {
     self.value = value
-    bridge = ConstantBridge(value as! NSNumber)
+    bridge = ConstantBridge(value)
   }
 }
 
-public func constant(_ number: any Numeric) -> Constant {
+public func constant(_ number: Any) -> Constant {
   return Constant(value: number)
 }
 
-public struct Field: Expr {
-  public var bridge: ExprBridge
+public struct Field: Expr, BridgeWrapper {
+  var bridge: ExprBridge
 
   var name: String
   init(name: String) {
@@ -52,8 +54,8 @@ protocol Function: Expr {
   var name: String { get }
 }
 
-public struct FunctionExpr: Function {
-  public var bridge: ExprBridge
+public struct FunctionExpr: Function, BridgeWrapper {
+  var bridge: ExprBridge
 
   var name: String
   private var args: [Expr]
@@ -61,7 +63,11 @@ public struct FunctionExpr: Function {
   init(name: String, args: [Expr]) {
     self.name = name
     self.args = args
-    bridge = FunctionExprBridge(name: name, args: args.map { $0.bridge })
+    bridge = FunctionExprBridge(
+      name: name,
+      args: args.map { ($0 as! (Expr & BridgeWrapper)).bridge
+      }
+    )
   }
 }
 
diff --git a/Firestore/Swift/Source/SwiftAPI/Stages.swift b/Firestore/Swift/Source/SwiftAPI/Stages.swift
index df3c163e803..c5de0c00e52 100644
--- a/Firestore/Swift/Source/SwiftAPI/Stages.swift
+++ b/Firestore/Swift/Source/SwiftAPI/Stages.swift
@@ -42,6 +42,6 @@ class Where: Stage {
 
   init(condition: Expr) {
     self.condition = condition
-    bridge = WhereStageBridge(expr: condition.bridge)
+    bridge = WhereStageBridge(expr: (condition as! (Expr & BridgeWrapper)).bridge)
   }
 }
diff --git a/Firestore/Swift/Tests/Integration/PipelineTests.swift b/Firestore/Swift/Tests/Integration/PipelineTests.swift
index 79185762b91..a2252488312 100644
--- a/Firestore/Swift/Tests/Integration/PipelineTests.swift
+++ b/Firestore/Swift/Tests/Integration/PipelineTests.swift
@@ -20,10 +20,11 @@ import Foundation
 @available(iOS 13, tvOS 13, macOS 10.15, macCatalyst 13, watchOS 7, *)
 class PipelineIntegrationTests: FSTIntegrationTestCase {
   func testCount() async throws {
+    try await firestore().collection("foo").document("bar").setData(["foo": "bar", "x": 42])
     let snapshot = try await firestore()
       .pipeline()
-      .collection(path: "foo")
-      .where(eq(field("foo"), constant(42)))
+      .collection(path: "/foo")
+      .where(eq(field("foo"), constant("bar")))
       .execute()
 
     print(snapshot)
diff --git a/Firestore/core/src/api/aggregate_expressions.cc b/Firestore/core/src/api/aggregate_expressions.cc
new file mode 100644
index 00000000000..87fc69c368a
--- /dev/null
+++ b/Firestore/core/src/api/aggregate_expressions.cc
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2025 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.
+ */
+
+#include "Firestore/core/src/api/aggregate_expressions.h"
+
+#include "Firestore/core/src/nanopb/nanopb_util.h"
+
+namespace firebase {
+namespace firestore {
+namespace api {
+
+google_firestore_v1_Value AggregateExpr::to_proto() const {
+  google_firestore_v1_Value result;
+  result.which_value_type = google_firestore_v1_Value_function_value_tag;
+
+  result.function_value.name = nanopb::MakeBytesArray(name_);
+  result.function_value.args_count = static_cast<pb_size_t>(params_.size());
+  result.function_value.args = nanopb::MakeArray<google_firestore_v1_Value>(
+      result.function_value.args_count);
+
+  for (size_t i = 0; i < params_.size(); ++i) {
+    result.function_value.args[i] = params_[i]->to_proto();
+  }
+
+  return result;
+}
+
+}  // namespace api
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/src/api/aggregate_expressions.h b/Firestore/core/src/api/aggregate_expressions.h
new file mode 100644
index 00000000000..119198b2abd
--- /dev/null
+++ b/Firestore/core/src/api/aggregate_expressions.h
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2025 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.
+ */
+
+#ifndef FIRESTORE_CORE_SRC_API_AGGREGATE_EXPRESSIONS_H_
+#define FIRESTORE_CORE_SRC_API_AGGREGATE_EXPRESSIONS_H_
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
+#include "Firestore/core/src/api/expressions.h"
+
+namespace firebase {
+namespace firestore {
+namespace api {
+
+class AggregateExpr {
+ public:
+  AggregateExpr(std::string name, std::vector<std::shared_ptr<Expr>> params)
+      : name_(std::move(name)), params_(std::move(params)) {
+  }
+  ~AggregateExpr() = default;
+
+  google_firestore_v1_Value to_proto() const;
+
+ private:
+  std::string name_;
+  std::vector<std::shared_ptr<Expr>> params_;
+};
+
+}  // namespace api
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_SRC_API_AGGREGATE_EXPRESSIONS_H_
diff --git a/Firestore/core/src/api/expressions.cc b/Firestore/core/src/api/expressions.cc
index 07e99b1e848..7ec517f2aab 100644
--- a/Firestore/core/src/api/expressions.cc
+++ b/Firestore/core/src/api/expressions.cc
@@ -19,6 +19,7 @@
 #include <memory>
 
 #include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
+#include "Firestore/core/src/model/value_util.h"
 #include "Firestore/core/src/nanopb/nanopb_util.h"
 
 namespace firebase {
@@ -35,12 +36,8 @@ google_firestore_v1_Value Field::to_proto() const {
 }
 
 google_firestore_v1_Value Constant::to_proto() const {
-  google_firestore_v1_Value result;
-
-  result.which_value_type = google_firestore_v1_Value_double_value_tag;
-  result.double_value = this->value_;
-
-  return result;
+  // Return a copy of the value proto to avoid double delete.
+  return *model::DeepClone(*value_).release();
 }
 
 google_firestore_v1_Value FunctionExpr::to_proto() const {
diff --git a/Firestore/core/src/api/expressions.h b/Firestore/core/src/api/expressions.h
index 2ab134249cf..5b08a277e3b 100644
--- a/Firestore/core/src/api/expressions.h
+++ b/Firestore/core/src/api/expressions.h
@@ -23,6 +23,7 @@
 #include <vector>
 
 #include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
+#include "Firestore/core/src/nanopb/message.h"
 
 namespace firebase {
 namespace firestore {
@@ -35,11 +36,20 @@ class Expr {
   virtual google_firestore_v1_Value to_proto() const = 0;
 };
 
-class Field : public Expr {
+class Selectable : public Expr {
+ public:
+  virtual ~Selectable() = default;
+  virtual const std::string& alias() const = 0;
+};
+
+class Field : public Selectable {
  public:
   explicit Field(std::string name) : name_(std::move(name)) {
   }
   google_firestore_v1_Value to_proto() const override;
+  const std::string& alias() const override {
+    return name_;
+  }
 
  private:
   std::string name_;
@@ -47,12 +57,13 @@ class Field : public Expr {
 
 class Constant : public Expr {
  public:
-  explicit Constant(double value) : value_(value) {
+  explicit Constant(nanopb::SharedMessage<google_firestore_v1_Value> value)
+      : value_(std::move(value)) {
   }
   google_firestore_v1_Value to_proto() const override;
 
  private:
-  double value_;
+  nanopb::SharedMessage<google_firestore_v1_Value> value_;
 };
 
 class FunctionExpr : public Expr {
diff --git a/Firestore/core/src/api/ordering.cc b/Firestore/core/src/api/ordering.cc
new file mode 100644
index 00000000000..6520cea5b6f
--- /dev/null
+++ b/Firestore/core/src/api/ordering.cc
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2025 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.
+ */
+
+#include "Firestore/core/src/api/ordering.h"
+
+#include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
+#include "Firestore/core/src/nanopb/nanopb_util.h"
+
+namespace firebase {
+namespace firestore {
+namespace api {
+
+google_firestore_v1_Value Ordering::to_proto() const {
+  google_firestore_v1_Value result;
+  result.which_value_type = google_firestore_v1_Value_map_value_tag;
+
+  result.map_value.fields_count = 2;
+  result.map_value.fields =
+      nanopb::MakeArray<google_firestore_v1_MapValue_FieldsEntry>(2);
+  result.map_value.fields[0].key = nanopb::MakeBytesArray("expression");
+  result.map_value.fields[0].value = field_.to_proto();
+  result.map_value.fields[1].key = nanopb::MakeBytesArray("direction");
+  google_firestore_v1_Value direction;
+  direction.which_value_type = google_firestore_v1_Value_string_value_tag;
+  direction.string_value = nanopb::MakeBytesArray(
+      this->direction_ == ASCENDING ? "ascending" : "descending");
+  result.map_value.fields[1].value = direction;
+
+  return result;
+}
+
+}  // namespace api
+}  // namespace firestore
+}  // namespace firebase
diff --git a/Firestore/core/src/api/ordering.h b/Firestore/core/src/api/ordering.h
new file mode 100644
index 00000000000..130dda12b19
--- /dev/null
+++ b/Firestore/core/src/api/ordering.h
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2025 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.
+ */
+
+#ifndef FIRESTORE_CORE_SRC_API_ORDERING_H_
+#define FIRESTORE_CORE_SRC_API_ORDERING_H_
+
+#include <utility>
+
+#include "Firestore/core/src/api/expressions.h"
+
+namespace firebase {
+namespace firestore {
+namespace api {
+
+class UserDataReader;  // forward declaration
+
+class Ordering {
+ public:
+  enum Direction {
+    ASCENDING,
+    DESCENDING,
+  };
+
+  Ordering(Field field, Direction direction)
+      : field_(std::move(field)), direction_(direction) {
+  }
+
+  google_firestore_v1_Value to_proto() const;
+
+ private:
+  Field field_;
+  Direction direction_;
+};
+
+}  // namespace api
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIRESTORE_CORE_SRC_API_ORDERING_H_
diff --git a/Firestore/core/src/api/pipeline_result.h b/Firestore/core/src/api/pipeline_result.h
index 4680d058c7b..53761752cdc 100644
--- a/Firestore/core/src/api/pipeline_result.h
+++ b/Firestore/core/src/api/pipeline_result.h
@@ -53,6 +53,10 @@ class PipelineResult {
   std::shared_ptr<model::ObjectValue> internal_value() const;
   absl::optional<absl::string_view> document_id() const;
 
+  const absl::optional<model::DocumentKey>& internal_key() const {
+    return internal_key_;
+  }
+
  private:
   absl::optional<model::DocumentKey> internal_key_;
   // Using a shared pointer to ObjectValue makes PipelineResult copy-assignable
diff --git a/Firestore/core/src/api/pipeline_snapshot.h b/Firestore/core/src/api/pipeline_snapshot.h
index a19e76138a7..079f2d57375 100644
--- a/Firestore/core/src/api/pipeline_snapshot.h
+++ b/Firestore/core/src/api/pipeline_snapshot.h
@@ -41,9 +41,18 @@ class PipelineSnapshot {
     return results_;
   }
 
+  const std::shared_ptr<Firestore> firestore() const {
+    return firestore_;
+  }
+
+  void SetFirestore(std::shared_ptr<Firestore> db) {
+    firestore_ = std::move(db);
+  }
+
  private:
   std::vector<PipelineResult> results_;
   model::SnapshotVersion execution_time_;
+  std::shared_ptr<Firestore> firestore_;
 };
 
 }  // namespace api
diff --git a/Firestore/core/src/api/stages.cc b/Firestore/core/src/api/stages.cc
index 6843a1b4ce5..eaa19cb03bd 100644
--- a/Firestore/core/src/api/stages.cc
+++ b/Firestore/core/src/api/stages.cc
@@ -16,6 +16,11 @@
 
 #include "Firestore/core/src/api/stages.h"
 
+#include <unordered_map>
+#include <utility>
+
+#include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
+#include "Firestore/core/src/nanopb/message.h"
 #include "Firestore/core/src/nanopb/nanopb_util.h"
 
 namespace firebase {
@@ -39,6 +44,112 @@ google_firestore_v1_Pipeline_Stage CollectionSource::to_proto() const {
   return result;
 }
 
+google_firestore_v1_Pipeline_Stage DatabaseSource::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+
+  result.name = nanopb::MakeBytesArray("database");
+  result.args_count = 0;
+  result.args = nullptr;
+  result.options_count = 0;
+  result.options = nullptr;
+
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage CollectionGroupSource::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+
+  result.name = nanopb::MakeBytesArray("collection_group");
+
+  result.args_count = 2;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(2);
+  // First argument is an empty reference value.
+  result.args[0].which_value_type =
+      google_firestore_v1_Value_reference_value_tag;
+  result.args[0].reference_value = nanopb::MakeBytesArray("");
+
+  // Second argument is the collection ID (encoded as a string value).
+  result.args[1].which_value_type = google_firestore_v1_Value_string_value_tag;
+  result.args[1].string_value = nanopb::MakeBytesArray(collection_id_);
+
+  result.options_count = 0;
+  result.options = nullptr;
+
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage DocumentsSource::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+
+  result.name = nanopb::MakeBytesArray("documents");
+
+  result.args_count = documents_.size();
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(result.args_count);
+
+  for (size_t i = 0; i < documents_.size(); ++i) {
+    result.args[i].which_value_type =
+        google_firestore_v1_Value_string_value_tag;
+    result.args[i].string_value = nanopb::MakeBytesArray(documents_[i]);
+  }
+
+  result.options_count = 0;
+  result.options = nullptr;
+
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage AddFields::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("add_fields");
+
+  result.args_count = 1;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
+
+  result.args[0].which_value_type = google_firestore_v1_Value_map_value_tag;
+  nanopb::SetRepeatedField(
+      &result.args[0].map_value.fields, &result.args[0].map_value.fields_count,
+      fields_, [](const std::shared_ptr<Selectable>& entry) {
+        return _google_firestore_v1_MapValue_FieldsEntry{
+            nanopb::MakeBytesArray(entry->alias()), entry->to_proto()};
+      });
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage AggregateStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("aggregate");
+
+  result.args_count = 2;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(2);
+
+  // Encode accumulators map.
+  result.args[0].which_value_type = google_firestore_v1_Value_map_value_tag;
+  nanopb::SetRepeatedField(
+      &result.args[0].map_value.fields, &result.args[0].map_value.fields_count,
+      this->accumulators_,
+      [](const std::pair<std::string, std::shared_ptr<AggregateExpr>>& entry) {
+        return _google_firestore_v1_MapValue_FieldsEntry{
+            nanopb::MakeBytesArray(entry.first), entry.second->to_proto()};
+      });
+
+  // Encode groups map.
+  result.args[1].which_value_type = google_firestore_v1_Value_map_value_tag;
+  nanopb::SetRepeatedField(
+      &result.args[1].map_value.fields, &result.args[1].map_value.fields_count,
+      this->groups_,
+      [](const std::pair<std::string, std::shared_ptr<Expr>>& entry) {
+        return _google_firestore_v1_MapValue_FieldsEntry{
+            nanopb::MakeBytesArray(entry.first), entry.second->to_proto()};
+      });
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
 google_firestore_v1_Pipeline_Stage Where::to_proto() const {
   google_firestore_v1_Pipeline_Stage result;
 
@@ -54,6 +165,145 @@ google_firestore_v1_Pipeline_Stage Where::to_proto() const {
   return result;
 }
 
+google_firestore_v1_Value FindNearestStage::DistanceMeasure::proto() const {
+  google_firestore_v1_Value result;
+  result.which_value_type = google_firestore_v1_Value_string_value_tag;
+  switch (measure_) {
+    case EUCLIDEAN:
+      result.string_value = nanopb::MakeBytesArray("euclidean");
+      break;
+    case COSINE:
+      result.string_value = nanopb::MakeBytesArray("cosine");
+      break;
+    case DOT_PRODUCT:
+      result.string_value = nanopb::MakeBytesArray("dot_product");
+      break;
+  }
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage FindNearestStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("find_nearest");
+
+  result.args_count = 3;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(3);
+  result.args[0] = property_->to_proto();
+  result.args[1] = *vector_;
+  result.args[2] = distance_measure_.proto();
+
+  nanopb::SetRepeatedField(
+      &result.options, &result.options_count, options_,
+      [](const std::pair<std::string,
+                         nanopb::SharedMessage<google_firestore_v1_Value>>&
+             entry) {
+        return _google_firestore_v1_Pipeline_Stage_OptionsEntry{
+            nanopb::MakeBytesArray(entry.first), *entry.second};
+      });
+
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage LimitStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("limit");
+
+  result.args_count = 1;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
+  result.args[0].which_value_type = google_firestore_v1_Value_integer_value_tag;
+  result.args[0].integer_value = limit_;
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage OffsetStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("offset");
+
+  result.args_count = 1;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
+  result.args[0].which_value_type = google_firestore_v1_Value_integer_value_tag;
+  result.args[0].integer_value = offset_;
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage SelectStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("select");
+
+  result.args_count = 1;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
+
+  result.args[0].which_value_type = google_firestore_v1_Value_map_value_tag;
+  nanopb::SetRepeatedField(
+      &result.args[0].map_value.fields, &result.args[0].map_value.fields_count,
+      fields_, [](const std::shared_ptr<Selectable>& entry) {
+        return _google_firestore_v1_MapValue_FieldsEntry{
+            nanopb::MakeBytesArray(entry->alias()), entry->to_proto()};
+      });
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage SortStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("sort");
+
+  result.args_count = static_cast<pb_size_t>(orders_.size());
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(result.args_count);
+
+  for (size_t i = 0; i < orders_.size(); ++i) {
+    result.args[i] = orders_[i].to_proto();
+  }
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage DistinctStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("distinct");
+
+  result.args_count = 1;
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
+
+  result.args[0].which_value_type = google_firestore_v1_Value_map_value_tag;
+  nanopb::SetRepeatedField(
+      &result.args[0].map_value.fields, &result.args[0].map_value.fields_count,
+      groups_, [](const std::shared_ptr<Selectable>& entry) {
+        return _google_firestore_v1_MapValue_FieldsEntry{
+            nanopb::MakeBytesArray(entry->alias()), entry->to_proto()};
+      });
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
+google_firestore_v1_Pipeline_Stage RemoveFieldsStage::to_proto() const {
+  google_firestore_v1_Pipeline_Stage result;
+  result.name = nanopb::MakeBytesArray("remove_fields");
+
+  result.args_count = static_cast<pb_size_t>(fields_.size());
+  result.args = nanopb::MakeArray<google_firestore_v1_Value>(result.args_count);
+
+  for (size_t i = 0; i < fields_.size(); ++i) {
+    result.args[i] = fields_[i].to_proto();
+  }
+
+  result.options_count = 0;
+  result.options = nullptr;
+  return result;
+}
+
 }  // namespace api
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/src/api/stages.h b/Firestore/core/src/api/stages.h
index f037a70408e..11534278002 100644
--- a/Firestore/core/src/api/stages.h
+++ b/Firestore/core/src/api/stages.h
@@ -19,9 +19,15 @@
 
 #include <memory>
 #include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
 
 #include "Firestore/Protos/nanopb/google/firestore/v1/document.nanopb.h"
+#include "Firestore/core/src/api/aggregate_expressions.h"
 #include "Firestore/core/src/api/expressions.h"
+#include "Firestore/core/src/api/ordering.h"
+#include "Firestore/core/src/nanopb/message.h"
 
 namespace firebase {
 namespace firestore {
@@ -37,7 +43,7 @@ class Stage {
 
 class CollectionSource : public Stage {
  public:
-  explicit CollectionSource(std::string path) : path_(path) {
+  explicit CollectionSource(std::string path) : path_(std::move(path)) {
   }
   ~CollectionSource() override = default;
 
@@ -47,9 +53,71 @@ class CollectionSource : public Stage {
   std::string path_;
 };
 
+class DatabaseSource : public Stage {
+ public:
+  DatabaseSource() = default;
+  ~DatabaseSource() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+};
+
+class CollectionGroupSource : public Stage {
+ public:
+  explicit CollectionGroupSource(std::string collection_id)
+      : collection_id_(std::move(collection_id)) {
+  }
+  ~CollectionGroupSource() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::string collection_id_;
+};
+
+class DocumentsSource : public Stage {
+ public:
+  explicit DocumentsSource(std::vector<std::string> documents)
+      : documents_(std::move(documents)) {
+  }
+  ~DocumentsSource() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::vector<std::string> documents_;
+};
+
+class AddFields : public Stage {
+ public:
+  explicit AddFields(std::vector<std::shared_ptr<Selectable>> fields)
+      : fields_(std::move(fields)) {
+  }
+  ~AddFields() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::vector<std::shared_ptr<Selectable>> fields_;
+};
+
+class AggregateStage : public Stage {
+ public:
+  AggregateStage(std::unordered_map<std::string, std::shared_ptr<AggregateExpr>>
+                     accumulators,
+                 std::unordered_map<std::string, std::shared_ptr<Expr>> groups)
+      : accumulators_(std::move(accumulators)), groups_(std::move(groups)) {
+  }
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::unordered_map<std::string, std::shared_ptr<AggregateExpr>> accumulators_;
+  std::unordered_map<std::string, std::shared_ptr<Expr>> groups_;
+};
+
 class Where : public Stage {
  public:
-  explicit Where(std::shared_ptr<Expr> expr) : expr_(expr) {
+  explicit Where(std::shared_ptr<Expr> expr) : expr_(std::move(expr)) {
   }
   ~Where() override = default;
 
@@ -59,6 +127,122 @@ class Where : public Stage {
   std::shared_ptr<Expr> expr_;
 };
 
+class FindNearestStage : public Stage {
+ public:
+  class DistanceMeasure {
+   public:
+    enum Measure { EUCLIDEAN, COSINE, DOT_PRODUCT };
+
+    explicit DistanceMeasure(Measure measure) : measure_(measure) {
+    }
+    google_firestore_v1_Value proto() const;
+
+   private:
+    Measure measure_;
+  };
+
+  FindNearestStage(
+      std::shared_ptr<Expr> property,
+      nanopb::SharedMessage<google_firestore_v1_Value> vector,
+      DistanceMeasure distance_measure,
+      std::unordered_map<std::string,
+                         nanopb::SharedMessage<google_firestore_v1_Value>>
+          options)
+      : property_(std::move(property)),
+        vector_(std::move(vector)),
+        distance_measure_(distance_measure),
+        options_(options) {
+  }
+
+  ~FindNearestStage() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::shared_ptr<Expr> property_;
+  nanopb::SharedMessage<google_firestore_v1_Value> vector_;
+  DistanceMeasure distance_measure_;
+  std::unordered_map<std::string,
+                     nanopb::SharedMessage<google_firestore_v1_Value>>
+      options_;
+};
+
+class LimitStage : public Stage {
+ public:
+  explicit LimitStage(int64_t limit) : limit_(limit) {
+  }
+  ~LimitStage() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  int64_t limit_;
+};
+
+class OffsetStage : public Stage {
+ public:
+  explicit OffsetStage(int64_t offset) : offset_(offset) {
+  }
+  ~OffsetStage() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  int64_t offset_;
+};
+
+class SelectStage : public Stage {
+ public:
+  explicit SelectStage(std::vector<std::shared_ptr<Selectable>> fields)
+      : fields_(std::move(fields)) {
+  }
+  ~SelectStage() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::vector<std::shared_ptr<Selectable>> fields_;
+};
+
+class SortStage : public Stage {
+ public:
+  explicit SortStage(std::vector<Ordering> orders)
+      : orders_(std::move(orders)) {
+  }
+  ~SortStage() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::vector<Ordering> orders_;
+};
+
+class DistinctStage : public Stage {
+ public:
+  explicit DistinctStage(std::vector<std::shared_ptr<Selectable>> groups)
+      : groups_(std::move(groups)) {
+  }
+  ~DistinctStage() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::vector<std::shared_ptr<Selectable>> groups_;
+};
+
+class RemoveFieldsStage : public Stage {
+ public:
+  explicit RemoveFieldsStage(std::vector<Field> fields)
+      : fields_(std::move(fields)) {
+  }
+  ~RemoveFieldsStage() override = default;
+
+  google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+ private:
+  std::vector<Field> fields_;
+};
+
 }  // namespace api
 }  // namespace firestore
 }  // namespace firebase
diff --git a/Firestore/core/src/remote/datastore.cc b/Firestore/core/src/remote/datastore.cc
index d5950ca09c6..504beadbd99 100644
--- a/Firestore/core/src/remote/datastore.cc
+++ b/Firestore/core/src/remote/datastore.cc
@@ -340,20 +340,22 @@ void Datastore::RunPipelineWithCredentials(
   GrpcUnaryCall* call = call_owning.get();
   active_calls_.push_back(std::move(call_owning));
 
-  call->Start([this, call, callback = std::move(callback)](
-                  const StatusOr<grpc::ByteBuffer>& result) {
-    LogGrpcCallFinished("ExecutePipeline", call, result.status());
-    HandleCallStatus(result.status());
+  call->Start(
+      [this, db = pipeline.firestore(), call, callback = std::move(callback)](
+          const StatusOr<grpc::ByteBuffer>& result) {
+        LogGrpcCallFinished("ExecutePipeline", call, result.status());
+        HandleCallStatus(result.status());
 
-    if (result.ok()) {
-      callback(datastore_serializer_.DecodeExecutePipelineResponse(
-          result.ValueOrDie()));
-    } else {
-      callback(result.status());
-    }
+        if (result.ok()) {
+          auto response = datastore_serializer_.DecodeExecutePipelineResponse(
+              result.ValueOrDie(), std::move(db));
+          callback(response);
+        } else {
+          callback(result.status());
+        }
 
-    RemoveGrpcCall(call);
-  });
+        RemoveGrpcCall(call);
+      });
 }
 
 void Datastore::ResumeRpcWithCredentials(const OnCredentials& on_credentials) {
diff --git a/Firestore/core/src/remote/remote_objc_bridge.cc b/Firestore/core/src/remote/remote_objc_bridge.cc
index 466ed1229cc..6cc675d4f7f 100644
--- a/Firestore/core/src/remote/remote_objc_bridge.cc
+++ b/Firestore/core/src/remote/remote_objc_bridge.cc
@@ -33,6 +33,7 @@
 #include "Firestore/core/src/remote/grpc_util.h"
 #include "Firestore/core/src/remote/watch_change.h"
 #include "Firestore/core/src/util/hard_assert.h"
+#include "Firestore/core/src/util/log.h"
 #include "Firestore/core/src/util/status.h"
 #include "Firestore/core/src/util/statusor.h"
 #include "grpcpp/support/status.h"
@@ -398,7 +399,8 @@ DatastoreSerializer::EncodeExecutePipelineRequest(
 
 util::StatusOr<api::PipelineSnapshot>
 DatastoreSerializer::DecodeExecutePipelineResponse(
-    const grpc::ByteBuffer& response) const {
+    const grpc::ByteBuffer& response,
+    std::shared_ptr<api::Firestore> db) const {
   ByteBufferReader reader{response};
   auto message =
       Message<google_firestore_v1_ExecutePipelineResponse>::TryParse(&reader);
@@ -406,7 +408,15 @@ DatastoreSerializer::DecodeExecutePipelineResponse(
     return reader.status();
   }
 
-  return serializer_.DecodePipelineResponse(reader.context(), message);
+  LOG_DEBUG("Pipeline Response: %s", message.ToString());
+
+  auto snapshot = serializer_.DecodePipelineResponse(reader.context(), message);
+  if (!reader.ok()) {
+    return reader.status();
+  }
+
+  snapshot.SetFirestore(std::move(db));
+  return snapshot;
 }
 
 }  // namespace remote
diff --git a/Firestore/core/src/remote/remote_objc_bridge.h b/Firestore/core/src/remote/remote_objc_bridge.h
index f6615003eed..96329c1ae25 100644
--- a/Firestore/core/src/remote/remote_objc_bridge.h
+++ b/Firestore/core/src/remote/remote_objc_bridge.h
@@ -156,7 +156,8 @@ class DatastoreSerializer {
       const firebase::firestore::api::Pipeline& pipeline) const;
 
   util::StatusOr<api::PipelineSnapshot> DecodeExecutePipelineResponse(
-      const grpc::ByteBuffer& response) const;
+      const grpc::ByteBuffer& response,
+      std::shared_ptr<api::Firestore> db) const;
 
  private:
   Serializer serializer_;