[realppl 6] offline ppl evaluation and tests

Author: wu-hui

Firestore/Example/Firestore.xcodeproj/project.pbxproj

@@ -29,6 +29,7 @@ namespace api {
 
 Field::Field(std::string name) {
   field_path_ = model::FieldPath::FromDotSeparatedString(name);
+  alias_ = field_path_.CanonicalString();
 }
 
 google_firestore_v1_Value Field::to_proto() const {

Firestore/core/src/api/expressions.cc

@@ -34,12 +34,12 @@ class Ordering {
     DESCENDING,
   };
 
-  Ordering(std::unique_ptr<Expr> expr, Direction direction)
+  Ordering(std::shared_ptr<Expr> expr, Direction direction)
       : expr_(std::move(expr)), direction_(direction) {
   }
 
-  const Expr& expr() const {
-    return *expr_;
+  const Expr* expr() const {
+    return expr_.get();
   }
 
   Direction direction() const {
@@ -49,7 +49,7 @@ class Ordering {
   google_firestore_v1_Value to_proto() const;
 
  private:
-  std::unique_ptr<Expr> expr_;
+  std::shared_ptr<Expr> expr_;
   Direction direction_;
 };
 

Firestore/core/src/api/ordering.h

@@ -44,6 +44,16 @@ const std::vector<std::shared_ptr<EvaluableStage>>& RealtimePipeline::stages()
   return this->stages_;
 }
 
+const std::vector<std::shared_ptr<EvaluableStage>>&
+RealtimePipeline::rewritten_stages() const {
+  return this->rewritten_stages_;
+}
+
+void RealtimePipeline::SetRewrittentStages(
+    std::vector<std::shared_ptr<EvaluableStage>> stages) {
+  this->rewritten_stages_ = std::move(stages);
+}
+
 EvaluateContext RealtimePipeline::evaluate_context() {
   return EvaluateContext(&serializer_);
 }

Firestore/core/src/api/realtime_pipeline.cc

@@ -38,11 +38,15 @@ class RealtimePipeline {
   RealtimePipeline AddingStage(std::shared_ptr<EvaluableStage> stage);
 
   const std::vector<std::shared_ptr<EvaluableStage>>& stages() const;
+  const std::vector<std::shared_ptr<EvaluableStage>>& rewritten_stages() const;
+
+  void SetRewrittentStages(std::vector<std::shared_ptr<EvaluableStage>>);
 
   EvaluateContext evaluate_context();
 
  private:
   std::vector<std::shared_ptr<EvaluableStage>> stages_;
+  std::vector<std::shared_ptr<EvaluableStage>> rewritten_stages_;
   remote::Serializer serializer_;
 };
 

Firestore/core/src/api/realtime_pipeline.h

@@ -40,6 +40,10 @@ namespace firebase {
 namespace firestore {
 namespace api {
 
+CollectionSource::CollectionSource(std::string path)
+    : path_(model::ResourcePath::FromStringView(path)) {
+}
+
 google_firestore_v1_Pipeline_Stage CollectionSource::to_proto() const {
   google_firestore_v1_Pipeline_Stage result;
 
@@ -49,7 +53,9 @@ google_firestore_v1_Pipeline_Stage CollectionSource::to_proto() const {
   result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
   result.args[0].which_value_type =
       google_firestore_v1_Value_reference_value_tag;
-  result.args[0].reference_value = nanopb::MakeBytesArray(this->path_);
+  // TODO: use EncodeResourceName instead
+  result.args[0].reference_value =
+      nanopb::MakeBytesArray(this->path_.CanonicalString());
 
   result.options_count = 0;
   result.options = nullptr;
@@ -324,7 +330,20 @@ model::PipelineInputOutputVector CollectionSource::Evaluate(
   std::copy_if(inputs.begin(), inputs.end(), std::back_inserter(results),
                [this](const model::MutableDocument& doc) {
                  return doc.is_found_document() &&
-                        doc.key().path().PopLast().CanonicalString() == path_;
+                        doc.key().path().PopLast().CanonicalString() ==
+                            path_.CanonicalString();
+               });
+  return results;
+}
+
+model::PipelineInputOutputVector CollectionGroupSource::Evaluate(
+    const EvaluateContext& /*context*/,
+    const model::PipelineInputOutputVector& inputs) const {
+  model::PipelineInputOutputVector results;
+  std::copy_if(inputs.begin(), inputs.end(), std::back_inserter(results),
+               [this](const model::MutableDocument& doc) {
+                 return doc.is_found_document() &&
+                        doc.key().GetCollectionGroup() == collection_id_;
                });
   return results;
 }
@@ -373,6 +392,39 @@ model::PipelineInputOutputVector LimitStage::Evaluate(
                                           inputs.begin() + count);
 }
 
+model::PipelineInputOutputVector SortStage::Evaluate(
+    const EvaluateContext& context,
+    const model::PipelineInputOutputVector& inputs) const {
+  model::PipelineInputOutputVector input_copy = inputs;
+  std::sort(
+      input_copy.begin(), input_copy.end(),
+      [this, &context](const model::PipelineInputOutput& left,
+                       const model::PipelineInputOutput& right) -> bool {
+        for (const auto& ordering : this->orders_) {
+          const auto left_result =
+              ordering.expr()->ToEvaluable()->Evaluate(context, left);
+          const auto right_result =
+              ordering.expr()->ToEvaluable()->Evaluate(context, right);
+
+          auto left_val = left_result.IsErrorOrUnset() ? model::MinValue()
+                                                       : *left_result.value();
+          auto right_val = right_result.IsErrorOrUnset()
+                               ? model::MinValue()
+                               : *right_result.value();
+          const auto compare_result = model::Compare(left_val, right_val);
+          if (compare_result != util::ComparisonResult::Same) {
+            return ordering.direction() == Ordering::ASCENDING
+                       ? compare_result == util::ComparisonResult::Ascending
+                       : compare_result == util::ComparisonResult::Descending;
+          }
+        }
+
+        return false;
+      });
+
+  return input_copy;
+}
+
 }  // namespace api
 }  // namespace firestore
 }  // namespace firebase

Firestore/core/src/api/stages.cc

@@ -28,6 +28,7 @@
 #include "Firestore/core/src/api/expressions.h"
 #include "Firestore/core/src/api/ordering.h"
 #include "Firestore/core/src/model/model_fwd.h"
+#include "Firestore/core/src/model/resource_path.h"
 #include "Firestore/core/src/nanopb/message.h"
 
 namespace firebase {
@@ -65,25 +66,29 @@ class EvaluableStage : public Stage {
   EvaluableStage() = default;
   virtual ~EvaluableStage() = default;
 
+  virtual absl::string_view name() const = 0;
   virtual model::PipelineInputOutputVector Evaluate(
       const EvaluateContext& context,
       const model::PipelineInputOutputVector& inputs) const = 0;
 };
 
 class CollectionSource : public EvaluableStage {
  public:
-  explicit CollectionSource(std::string path) : path_(std::move(path)) {
-  }
+  explicit CollectionSource(std::string path);
   ~CollectionSource() override = default;
 
   google_firestore_v1_Pipeline_Stage to_proto() const override;
 
+  absl::string_view name() const override {
+    return "collection";
+  }
+
   model::PipelineInputOutputVector Evaluate(
       const EvaluateContext& context,
       const model::PipelineInputOutputVector& inputs) const override;
 
  private:
-  std::string path_;
+  model::ResourcePath path_;
 };
 
 class DatabaseSource : public EvaluableStage {
@@ -92,6 +97,11 @@ class DatabaseSource : public EvaluableStage {
   ~DatabaseSource() override = default;
 
   google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+  absl::string_view name() const override {
+    return "database";
+  }
+
   model::PipelineInputOutputVector Evaluate(
       const EvaluateContext& context,
       const model::PipelineInputOutputVector& inputs) const override;
@@ -106,6 +116,14 @@ class CollectionGroupSource : public EvaluableStage {
 
   google_firestore_v1_Pipeline_Stage to_proto() const override;
 
+  absl::string_view name() const override {
+    return "collection_group";
+  }
+
+  model::PipelineInputOutputVector Evaluate(
+      const EvaluateContext& context,
+      const model::PipelineInputOutputVector& inputs) const override;
+
  private:
   std::string collection_id_;
 };
@@ -119,6 +137,10 @@ class DocumentsSource : public EvaluableStage {
 
   google_firestore_v1_Pipeline_Stage to_proto() const override;
 
+  absl::string_view name() const override {
+    return "documents";
+  }
+
  private:
   std::vector<std::string> documents_;
 };
@@ -158,6 +180,11 @@ class Where : public EvaluableStage {
   ~Where() override = default;
 
   google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+  absl::string_view name() const override {
+    return "where";
+  }
+
   model::PipelineInputOutputVector Evaluate(
       const EvaluateContext& context,
       const model::PipelineInputOutputVector& inputs) const override;
@@ -213,6 +240,11 @@ class LimitStage : public EvaluableStage {
   ~LimitStage() override = default;
 
   google_firestore_v1_Pipeline_Stage to_proto() const override;
+
+  absl::string_view name() const override {
+    return "limit";
+  }
+
   model::PipelineInputOutputVector Evaluate(
       const EvaluateContext& context,
       const model::PipelineInputOutputVector& inputs) const override;
@@ -255,6 +287,18 @@ class SortStage : public EvaluableStage {
 
   google_firestore_v1_Pipeline_Stage to_proto() const override;
 
+  absl::string_view name() const override {
+    return "sort";
+  }
+
+  model::PipelineInputOutputVector Evaluate(
+      const EvaluateContext& context,
+      const model::PipelineInputOutputVector& inputs) const override;
+
+  const std::vector<Ordering>& orders() const {
+    return orders_;
+  }
+
  private:
   std::vector<Ordering> orders_;
 };

Firestore/core/src/api/stages.h

@@ -24,9 +24,7 @@
 
 #include "Firestore/core/src/api/expressions.h"
 #include "Firestore/core/src/api/stages.h"
-#include "Firestore/core/src/model/value_util.h"
 #include "Firestore/core/src/nanopb/message.h"
-#include "Firestore/core/src/util/hard_assert.h"
 #include "absl/types/optional.h"
 
 namespace firebase {

Firestore/core/src/core/expressions_eval.h

@@ -21,6 +21,7 @@
 #include "Firestore/core/src/api/stages.h"
 #include "Firestore/core/src/model/mutable_document.h"
 #include "Firestore/core/src/remote/serializer.h"
+#include "pipeline_util.h"
 
 namespace firebase {
 namespace firestore {
@@ -29,8 +30,12 @@ namespace core {
 std::vector<model::MutableDocument> RunPipeline(
     api::RealtimePipeline& pipeline,
     const std::vector<model::MutableDocument>& inputs) {
-  auto& current = const_cast<std::vector<model::MutableDocument>&>(inputs);
-  for (const auto& stage : pipeline.stages()) {
+  if (pipeline.rewritten_stages().empty()) {
+    pipeline.SetRewrittentStages(RewriteStages(pipeline.stages()));
+  }
+
+  auto current = std::vector<model::MutableDocument>(inputs);
+  for (const auto& stage : pipeline.rewritten_stages()) {
     current = stage->Evaluate(pipeline.evaluate_context(), current);
   }
 

Firestore/core/src/core/pipeline_run.cc

@@ -0,0 +1,92 @@
+/*
+ * 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/core/pipeline_util.h"
+
+#include "Firestore/core/src/api/expressions.h"
+#include "Firestore/core/src/api/realtime_pipeline.h"
+#include "Firestore/core/src/api/stages.h"
+#include "Firestore/core/src/model/mutable_document.h"
+#include "Firestore/core/src/remote/serializer.h"
+
+namespace firebase {
+namespace firestore {
+namespace core {
+
+namespace {
+
+auto NewKeyOrdering() {
+  return api::Ordering(
+      std::make_shared<api::Field>(model::FieldPath::KeyFieldPath()),
+      api::Ordering::Direction::ASCENDING);
+}
+
+}  // namespace
+
+std::vector<std::shared_ptr<api::EvaluableStage>> RewriteStages(
+    const std::vector<std::shared_ptr<api::EvaluableStage>>& stages) {
+  bool has_order = false;
+  std::vector<std::shared_ptr<api::EvaluableStage>> new_stages;
+  for (const auto& stage : stages) {
+    // For stages that provide ordering semantics
+    if (stage->name() == "sort") {
+      auto sort_stage = std::static_pointer_cast<api::SortStage>(stage);
+      has_order = true;
+
+      // Ensure we have a stable ordering
+      bool includes_key_ordering = false;
+      for (const auto& order : sort_stage->orders()) {
+        auto field = dynamic_cast<const api::Field*>(order.expr());
+        if (field != nullptr && field->field_path().IsKeyFieldPath()) {
+          includes_key_ordering = true;
+          break;
+        }
+      }
+
+      if (includes_key_ordering) {
+        new_stages.push_back(stage);
+      } else {
+        auto copy = sort_stage->orders();
+        copy.push_back(NewKeyOrdering());
+        new_stages.push_back(std::make_shared<api::SortStage>(std::move(copy)));
+      }
+    }
+    // For stages whose semantics depend on ordering
+    else if (stage->name() == "limit") {
+      if (!has_order) {
+        new_stages.push_back(std::make_shared<api::SortStage>(
+            std::vector<api::Ordering>{NewKeyOrdering()}));
+        has_order = true;
+      }
+      new_stages.push_back(stage);
+    }
+    // TODO(wuandy): Handle add_fields and select and such
+    else {
+      new_stages.push_back(stage);
+    }
+  }
+
+  if (!has_order) {
+    new_stages.push_back(std::make_shared<api::SortStage>(
+        std::vector<api::Ordering>{NewKeyOrdering()}));
+  }
+
+  return new_stages;
+}
+
+}  // namespace core
+}  // namespace firestore
+}  // namespace firebase