Add documentation for ppl

.gitignore

@@ -61,6 +61,8 @@ profile
 DerivedData
 *.hmap
 *.ipa
+# Xcode index build files
+.index-build/
 
 # Swift Package Manager
 Package.resolved
@@ -166,3 +168,4 @@ Firestore/Example/GoogleService-Info.plist
 
 # FirebaseVertexAI test data
 vertexai-sdk-test-data
+

Firestore/Example/Tests/Integration/API/FIRAggregateTests.mm

@@ -516,7 +516,9 @@ - (void)testCannotPerformMoreThanMaxAggregations {
   [self awaitExpectation:expectation];
 
   XCTAssertNotNil(result);
-  XCTAssertTrue([[result localizedDescription] containsString:@"maximum number of aggregations"]);
+  if (!FSTIntegrationTestCase.isRunningAgainstEmulator) {
+    XCTAssertTrue([[result localizedDescription] containsString:@"maximum number of aggregations"]);
+  }
 }
 
 - (void)testThrowsAnErrorWhenGettingTheResultOfAnUnrequestedAggregation {

Firestore/Example/Tests/SpecTests/FSTLevelDBSpecTests.mm

@@ -62,4 +62,18 @@ - (BOOL)shouldRunWithTags:(NSArray<NSString *> *)tags {
 
 @end
 
+/**
+ * An implementation of FSTLevelDBSpecTests that runs tests in pipeline mode.
+ */
+@interface FSTLevelDBPipelineSpecTests : FSTLevelDBSpecTests
+@end
+
+@implementation FSTLevelDBPipelineSpecTests
+
+- (BOOL)usePipelineMode {
+  return YES;
+}
+
+@end
+
 NS_ASSUME_NONNULL_END

Firestore/Example/Tests/SpecTests/FSTMemorySpecTests.mm

@@ -57,4 +57,18 @@ - (BOOL)shouldRunWithTags:(NSArray<NSString *> *)tags {
 
 @end
 
+/**
+ * An implementation of FSTMemorySpecTests that runs tests in pipeline mode.
+ */
+@interface FSTMemoryPipelineSpecTests : FSTMemorySpecTests
+@end
+
+@implementation FSTMemoryPipelineSpecTests
+
+- (BOOL)usePipelineMode {
+  return YES;
+}
+
+@end
+
 NS_ASSUME_NONNULL_END

Firestore/Example/Tests/SpecTests/FSTSpecTests.h

@@ -37,14 +37,20 @@ extern NSString *const kDurablePersistence;
  * + Subclass FSTSpecTests
  * + override -persistence to create and return an appropriate Persistence implementation.
  */
-@interface FSTSpecTests : XCTestCase
+@interface FSTSpecTests : XCTestCase {
+ @protected
+  BOOL _convertToPipeline;
+}
 
 /** Based on its tags, determine whether the test case should run. */
 - (BOOL)shouldRunWithTags:(NSArray<NSString *> *)tags;
 
 /** Do any necessary setup for a single spec test */
 - (void)setUpForSpecWithConfig:(NSDictionary *)config;
 
+/** Determines if tests should run in pipeline mode. Subclasses can override. */
+- (BOOL)usePipelineMode;
+
 @end
 
 NS_ASSUME_NONNULL_END

Firestore/Example/Tests/SpecTests/FSTSpecTests.mm

@@ -158,6 +158,9 @@
 // if `kRunBenchmarkTests` is set to 'YES'.
 static NSString *const kBenchmarkTag = @"benchmark";
 
+// A tag for tests that should skip its pipeline run.
+static NSString *const kNoPipelineConversion = @"no-pipeline-conversion";
+
 NSString *const kEagerGC = @"eager-gc";
 
 NSString *const kDurablePersistence = @"durable-persistence";
@@ -236,11 +239,14 @@ - (BOOL)shouldRunWithTags:(NSArray<NSString *> *)tags {
     return NO;
   } else if (!kRunBenchmarkTests && [tags containsObject:kBenchmarkTag]) {
     return NO;
+  } else if (self.usePipelineMode && [tags containsObject:kNoPipelineConversion]) {
+    return NO;
   }
   return YES;
 }
 
 - (void)setUpForSpecWithConfig:(NSDictionary *)config {
+  _convertToPipeline = [self usePipelineMode];  // Call new method
   _reader = FSTTestUserDataReader();
   std::unique_ptr<Executor> user_executor = Executor::CreateSerial("user executor");
   user_executor_ = absl::ShareUniquePtr(std::move(user_executor));
@@ -261,6 +267,7 @@ - (void)setUpForSpecWithConfig:(NSDictionary *)config {
   self.driver =
       [[FSTSyncEngineTestDriver alloc] initWithPersistence:std::move(persistence)
                                                    eagerGC:_useEagerGCForMemory
+                                         convertToPipeline:_convertToPipeline  // Pass the flag
                                                initialUser:User::Unauthenticated()
                                          outstandingWrites:{}
                              maxConcurrentLimboResolutions:_maxConcurrentLimboResolutions];
@@ -282,6 +289,11 @@ - (BOOL)isTestBaseClass {
   return [self class] == [FSTSpecTests class];
 }
 
+// Default implementation for pipeline mode. Subclasses can override.
+- (BOOL)usePipelineMode {
+  return NO;
+}
+
 #pragma mark - Methods for constructing objects from specs.
 
 - (Query)parseQuery:(id)querySpec {
@@ -645,6 +657,7 @@ - (void)doRestart {
   self.driver =
       [[FSTSyncEngineTestDriver alloc] initWithPersistence:std::move(persistence)
                                                    eagerGC:_useEagerGCForMemory
+                                         convertToPipeline:_convertToPipeline  // Pass the flag
                                                initialUser:currentUser
                                          outstandingWrites:outstandingWrites
                              maxConcurrentLimboResolutions:_maxConcurrentLimboResolutions];
@@ -721,8 +734,42 @@ - (void)doStep:(NSDictionary *)step {
 }
 
 - (void)validateEvent:(FSTQueryEvent *)actual matches:(NSDictionary *)expected {
-  Query expectedQuery = [self parseQuery:expected[@"query"]];
-  XCTAssertEqual(actual.query, expectedQuery);
+  // The 'expected' query from JSON is always a standard Query.
+  Query expectedJSONQuery = [self parseQuery:expected[@"query"]];
+  core::QueryOrPipeline actualQueryOrPipeline = actual.queryOrPipeline;
+
+  if (_convertToPipeline) {
+    XCTAssertTrue(actualQueryOrPipeline.IsPipeline(),
+                  @"In pipeline mode, actual event query should be a pipeline. Actual: %@",
+                  MakeNSString(actualQueryOrPipeline.ToString()));
+
+    // Convert the expected JSON Query to a RealtimePipeline for comparison.
+    std::vector<std::shared_ptr<api::EvaluableStage>> expectedStages =
+        core::ToPipelineStages(expectedJSONQuery);
+    // TODO(specstest): Need access to the database_id for the serializer.
+    // Assuming self.driver.databaseInfo is accessible and provides it.
+    // This might require making databaseInfo public or providing a getter in
+    // FSTSyncEngineTestDriver. For now, proceeding with the assumption it's available.
+    auto serializer = absl::make_unique<remote::Serializer>(self.driver.databaseInfo.database_id());
+    api::RealtimePipeline expectedPipeline(std::move(expectedStages), std::move(serializer));
+    auto expectedQoPForComparison =
+        core::QueryOrPipeline(expectedPipeline);  // Wrap expected pipeline
+
+    XCTAssertEqual(actualQueryOrPipeline.CanonicalId(), expectedQoPForComparison.CanonicalId(),
+                   @"Pipeline canonical IDs do not match. Actual: %@, Expected: %@",
+                   MakeNSString(actualQueryOrPipeline.CanonicalId()),
+                   MakeNSString(expectedQoPForComparison.CanonicalId()));
+
+  } else {
+    XCTAssertFalse(actualQueryOrPipeline.IsPipeline(),
+                   @"In non-pipeline mode, actual event query should be a Query. Actual: %@",
+                   MakeNSString(actualQueryOrPipeline.ToString()));
+    XCTAssertTrue(actualQueryOrPipeline.query() == expectedJSONQuery,
+                  @"Queries do not match. Actual: %@, Expected: %@",
+                  MakeNSString(actualQueryOrPipeline.query().ToString()),
+                  MakeNSString(expectedJSONQuery.ToString()));
+  }
+
   if ([expected[@"errorCode"] integerValue] != 0) {
     XCTAssertNotNil(actual.error);
     XCTAssertEqual(actual.error.code, [expected[@"errorCode"] integerValue]);
@@ -787,14 +834,43 @@ - (void)validateExpectedSnapshotEvents:(NSArray *_Nullable)expectedEvents {
   XCTAssertEqual(events.count, expectedEvents.count);
   events =
       [events sortedArrayUsingComparator:^NSComparisonResult(FSTQueryEvent *q1, FSTQueryEvent *q2) {
-        return WrapCompare(q1.query.CanonicalId(), q2.query.CanonicalId());
-      }];
-  expectedEvents = [expectedEvents
-      sortedArrayUsingComparator:^NSComparisonResult(NSDictionary *left, NSDictionary *right) {
-        Query leftQuery = [self parseQuery:left[@"query"]];
-        Query rightQuery = [self parseQuery:right[@"query"]];
-        return WrapCompare(leftQuery.CanonicalId(), rightQuery.CanonicalId());
+        // Use QueryOrPipeline's CanonicalId for sorting
+        return WrapCompare(q1.queryOrPipeline.CanonicalId(), q2.queryOrPipeline.CanonicalId());
       }];
+  expectedEvents = [expectedEvents sortedArrayUsingComparator:^NSComparisonResult(
+                                       NSDictionary *left, NSDictionary *right) {
+    // Expected query from JSON is always a core::Query.
+    // For sorting consistency with actual events (which might be pipelines),
+    // we convert the expected query to QueryOrPipeline then get its CanonicalId.
+    // If _convertToPipeline is true, this will effectively sort expected items
+    // by their pipeline canonical ID.
+    Query leftJSONQuery = [self parseQuery:left[@"query"]];
+    core::QueryOrPipeline leftQoP;
+    if (self->_convertToPipeline) {
+      std::vector<std::shared_ptr<api::EvaluableStage>> stages =
+          core::ToPipelineStages(leftJSONQuery);
+      auto serializer =
+          absl::make_unique<remote::Serializer>(self.driver.databaseInfo.database_id());
+      leftQoP =
+          core::QueryOrPipeline(api::RealtimePipeline(std::move(stages), std::move(serializer)));
+    } else {
+      leftQoP = core::QueryOrPipeline(leftJSONQuery);
+    }
+
+    Query rightJSONQuery = [self parseQuery:right[@"query"]];
+    core::QueryOrPipeline rightQoP;
+    if (self->_convertToPipeline) {
+      std::vector<std::shared_ptr<api::EvaluableStage>> stages =
+          core::ToPipelineStages(rightJSONQuery);
+      auto serializer =
+          absl::make_unique<remote::Serializer>(self.driver.databaseInfo.database_id());
+      rightQoP =
+          core::QueryOrPipeline(api::RealtimePipeline(std::move(stages), std::move(serializer)));
+    } else {
+      rightQoP = core::QueryOrPipeline(rightJSONQuery);
+    }
+    return WrapCompare(leftQoP.CanonicalId(), rightQoP.CanonicalId());
+  }];
 
   NSUInteger i = 0;
   for (; i < expectedEvents.count && i < events.count; ++i) {
@@ -842,36 +918,49 @@ - (void)validateExpectedState:(nullable NSDictionary *)expectedState {
     }
     if (expectedState[@"activeTargets"]) {
       __block ActiveTargetMap expectedActiveTargets;
-      [expectedState[@"activeTargets"] enumerateKeysAndObjectsUsingBlock:^(NSString *targetIDString,
-                                                                           NSDictionary *queryData,
-                                                                           BOOL *) {
-        TargetId targetID = [targetIDString intValue];
-        NSArray *queriesJson = queryData[@"queries"];
-        std::vector<TargetData> queries;
-        for (id queryJson in queriesJson) {
-          Query query = [self parseQuery:queryJson];
-
-          QueryPurpose purpose = QueryPurpose::Listen;
-          if ([queryData objectForKey:@"targetPurpose"] != nil) {
-            purpose = [self parseQueryPurpose:queryData[@"targetPurpose"]];
-          }
-
-          TargetData target_data(core::TargetOrPipeline(query.ToTarget()), targetID, 0, purpose);
-          if ([queryData objectForKey:@"resumeToken"] != nil) {
-            target_data = target_data.WithResumeToken(MakeResumeToken(queryData[@"resumeToken"]),
-                                                      SnapshotVersion::None());
-          } else {
-            target_data = target_data.WithResumeToken(ByteString(),
-                                                      [self parseVersion:queryData[@"readTime"]]);
-          }
-
-          if ([queryData objectForKey:@"expectedCount"] != nil) {
-            target_data = target_data.WithExpectedCount([queryData[@"expectedCount"] intValue]);
-          }
-          queries.push_back(std::move(target_data));
-        }
-        expectedActiveTargets[targetID] = std::move(queries);
-      }];
+      [expectedState[@"activeTargets"]
+          enumerateKeysAndObjectsUsingBlock:^(NSString *targetIDString, NSDictionary *queryData,
+                                              BOOL *) {
+            TargetId targetID = [targetIDString intValue];
+            NSArray *queriesJson = queryData[@"queries"];
+            std::vector<TargetData> queries;
+            for (id queryJson in queriesJson) {
+              QueryPurpose purpose = QueryPurpose::Listen;
+              if ([queryData objectForKey:@"targetPurpose"] != nil) {
+                purpose = [self parseQueryPurpose:queryData[@"targetPurpose"]];
+              }
+
+              core::TargetOrPipeline top;
+              Query query = [self parseQuery:queryJson];
+
+              if (self->_convertToPipeline &&
+                  purpose != firebase::firestore::local::QueryPurpose::LimboResolution) {
+                std::vector<std::shared_ptr<api::EvaluableStage>> stages =
+                    core::ToPipelineStages(query);
+                auto serializer =
+                    absl::make_unique<remote::Serializer>(self.driver.databaseInfo.database_id());
+                top = core::TargetOrPipeline(
+                    api::RealtimePipeline(std::move(stages), std::move(serializer)));
+              } else {
+                top = core::TargetOrPipeline(query.ToTarget());
+              }
+
+              TargetData target_data(top, targetID, 0, purpose);
+              if ([queryData objectForKey:@"resumeToken"] != nil) {
+                target_data = target_data.WithResumeToken(
+                    MakeResumeToken(queryData[@"resumeToken"]), SnapshotVersion::None());
+              } else {
+                target_data = target_data.WithResumeToken(
+                    ByteString(), [self parseVersion:queryData[@"readTime"]]);
+              }
+
+              if ([queryData objectForKey:@"expectedCount"] != nil) {
+                target_data = target_data.WithExpectedCount([queryData[@"expectedCount"] intValue]);
+              }
+              queries.push_back(std::move(target_data));
+            }
+            expectedActiveTargets[targetID] = std::move(queries);
+          }];
       [self.driver setExpectedActiveTargets:std::move(expectedActiveTargets)];
     }
   }
@@ -980,9 +1069,13 @@ - (void)validateActiveTargets {
     // is ever made to be consistent.
     // XCTAssertEqualObjects(actualTargets[targetID], TargetData);
     const TargetData &actual = found->second;
-
+    auto left = actual.target_or_pipeline();
+    auto right = targetData.target_or_pipeline();
+    auto left_p = left.IsPipeline();
+    auto right_p = right.IsPipeline();
+    XCTAssertEqual(left_p, right_p);
+    XCTAssertEqual(left, right);
     XCTAssertEqual(actual.purpose(), targetData.purpose());
-    XCTAssertEqual(actual.target_or_pipeline(), targetData.target_or_pipeline());
     XCTAssertEqual(actual.target_id(), targetData.target_id());
     XCTAssertEqual(actual.snapshot_version(), targetData.snapshot_version());
     XCTAssertEqual(actual.resume_token(), targetData.resume_token());
@@ -1032,6 +1125,8 @@ - (void)runSpecTestSteps:(NSArray *)steps config:(NSDictionary *)config {
 - (void)testSpecTests {
   if ([self isTestBaseClass]) return;
 
+  // LogSetLevel(firebase::firestore::util::kLogLevelDebug);
+
   // Enumerate the .json files containing the spec tests.
   NSMutableArray<NSString *> *specFiles = [NSMutableArray array];
   NSMutableArray<NSDictionary *> *parsedSpecs = [NSMutableArray array];
@@ -1121,10 +1216,10 @@ - (void)testSpecTests {
         ++testPassCount;
       } else {
         ++testSkipCount;
-        NSLog(@"  [SKIPPED] Spec test: %@", name);
+        // NSLog(@"  [SKIPPED] Spec test: %@", name);
         NSString *comment = testDescription[@"comment"];
         if (comment) {
-          NSLog(@"    %@", comment);
+          // NSLog(@"    %@", comment);
         }
       }
     }];

Firestore/Example/Tests/SpecTests/FSTSyncEngineTestDriver.h

@@ -26,6 +26,7 @@
 #include "Firestore/core/src/api/load_bundle_task.h"
 #include "Firestore/core/src/bundle/bundle_reader.h"
 #include "Firestore/core/src/core/database_info.h"
+#include "Firestore/core/src/core/pipeline_util.h"  // For QueryOrPipeline
 #include "Firestore/core/src/core/query.h"
 #include "Firestore/core/src/core/view_snapshot.h"
 #include "Firestore/core/src/credentials/user.h"
@@ -66,7 +67,7 @@ NS_ASSUME_NONNULL_BEGIN
  * given query.
  */
 @interface FSTQueryEvent : NSObject
-@property(nonatomic, assign) core::Query query;
+@property(nonatomic, assign) core::QueryOrPipeline queryOrPipeline;
 @property(nonatomic, strong, nullable) NSError *error;
 
 - (const absl::optional<core::ViewSnapshot> &)viewSnapshot;
@@ -115,7 +116,10 @@ typedef std::
  *
  * Each method on the driver injects a different event into the system.
  */
-@interface FSTSyncEngineTestDriver : NSObject
+@interface FSTSyncEngineTestDriver : NSObject {
+ @protected
+  BOOL _convertToPipeline;
+}
 
 /**
  * Initializes the underlying FSTSyncEngine with the given local persistence implementation and
@@ -124,6 +128,7 @@ typedef std::
  */
 - (instancetype)initWithPersistence:(std::unique_ptr<local::Persistence>)persistence
                             eagerGC:(BOOL)eagerGC
+                  convertToPipeline:(BOOL)convertToPipeline
                         initialUser:(const credentials::User &)initialUser
                   outstandingWrites:(const FSTOutstandingWriteQueues &)outstandingWrites
       maxConcurrentLimboResolutions:(size_t)maxConcurrentLimboResolutions NS_DESIGNATED_INITIALIZER;