Skip to content

Commit 7f35bbd

Browse files
wu-huiwu-hui
committed
[realppl 6] offline ppl evaluation and tests
1 parent 89aa149 commit 7f35bbd

27 files changed

+8511
-54
lines changed

‎Firestore/core/src/api/expressions.cc‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ namespace api {
2929

3030
Field::Field(std::string name) {
3131
field_path_ = model::FieldPath::FromDotSeparatedString(name);
32+
alias_ = field_path_.CanonicalString();
3233
}
3334

3435
google_firestore_v1_Value Field::to_proto() const {

‎Firestore/core/src/api/ordering.h‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,12 +34,12 @@ class Ordering {
3434
DESCENDING,
3535
};
3636

37-
Ordering(std::unique_ptr<Expr> expr, Direction direction)
37+
Ordering(std::shared_ptr<Expr> expr, Direction direction)
3838
: expr_(std::move(expr)), direction_(direction) {
3939
}
4040

41-
const Expr& expr() const {
42-
return *expr_;
41+
const Expr* expr() const {
42+
return expr_.get();
4343
}
4444

4545
Direction direction() const {
@@ -49,7 +49,7 @@ class Ordering {
4949
google_firestore_v1_Value to_proto() const;
5050

5151
private:
52-
std::unique_ptr<Expr> expr_;
52+
std::shared_ptr<Expr> expr_;
5353
Direction direction_;
5454
};
5555

‎Firestore/core/src/api/realtime_pipeline.cc‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,16 @@ const std::vector<std::shared_ptr<EvaluableStage>>& RealtimePipeline::stages()
4545
return this->stages_;
4646
}
4747

48+
const std::vector<std::shared_ptr<EvaluableStage>>&
49+
RealtimePipeline::rewritten_stages() const {
50+
return this->rewritten_stages_;
51+
}
52+
53+
void RealtimePipeline::SetRewrittentStages(
54+
std::vector<std::shared_ptr<EvaluableStage>> stages) {
55+
this->rewritten_stages_ = std::move(stages);
56+
}
57+
4858
EvaluateContext RealtimePipeline::evaluate_context() {
4959
return EvaluateContext(&serializer_);
5060
}

‎Firestore/core/src/api/realtime_pipeline.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,11 +38,15 @@ class RealtimePipeline {
3838
RealtimePipeline AddingStage(std::shared_ptr<EvaluableStage> stage);
3939

4040
const std::vector<std::shared_ptr<EvaluableStage>>& stages() const;
41+
const std::vector<std::shared_ptr<EvaluableStage>>& rewritten_stages() const;
42+
43+
void SetRewrittentStages(std::vector<std::shared_ptr<EvaluableStage>>);
4144

4245
EvaluateContext evaluate_context();
4346

4447
private:
4548
std::vector<std::shared_ptr<EvaluableStage>> stages_;
49+
std::vector<std::shared_ptr<EvaluableStage>> rewritten_stages_;
4650
remote::Serializer serializer_;
4751
};
4852

‎Firestore/core/src/api/stages.cc‎

Lines changed: 54 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ namespace firebase {
4040
namespace firestore {
4141
namespace api {
4242

43+
CollectionSource::CollectionSource(std::string path)
44+
: path_(model::ResourcePath::FromStringView(path)) {
45+
}
46+
4347
google_firestore_v1_Pipeline_Stage CollectionSource::to_proto() const {
4448
google_firestore_v1_Pipeline_Stage result;
4549

@@ -49,7 +53,9 @@ google_firestore_v1_Pipeline_Stage CollectionSource::to_proto() const {
4953
result.args = nanopb::MakeArray<google_firestore_v1_Value>(1);
5054
result.args[0].which_value_type =
5155
google_firestore_v1_Value_reference_value_tag;
52-
result.args[0].reference_value = nanopb::MakeBytesArray(this->path_);
56+
// TODO: use EncodeResourceName instead
57+
result.args[0].reference_value =
58+
nanopb::MakeBytesArray(this->path_.CanonicalString());
5359

5460
result.options_count = 0;
5561
result.options = nullptr;
@@ -324,7 +330,20 @@ model::PipelineInputOutputVector CollectionSource::Evaluate(
324330
std::copy_if(inputs.begin(), inputs.end(), std::back_inserter(results),
325331
[this](const model::MutableDocument& doc) {
326332
return doc.is_found_document() &&
327-
doc.key().path().PopLast().CanonicalString() == path_;
333+
doc.key().path().PopLast().CanonicalString() ==
334+
path_.CanonicalString();
335+
});
336+
return results;
337+
}
338+
339+
model::PipelineInputOutputVector CollectionGroupSource::Evaluate(
340+
const EvaluateContext& /*context*/,
341+
const model::PipelineInputOutputVector& inputs) const {
342+
model::PipelineInputOutputVector results;
343+
std::copy_if(inputs.begin(), inputs.end(), std::back_inserter(results),
344+
[this](const model::MutableDocument& doc) {
345+
return doc.is_found_document() &&
346+
doc.key().GetCollectionGroup() == collection_id_;
328347
});
329348
return results;
330349
}
@@ -373,6 +392,39 @@ model::PipelineInputOutputVector LimitStage::Evaluate(
373392
inputs.begin() + count);
374393
}
375394

395+
model::PipelineInputOutputVector SortStage::Evaluate(
396+
const EvaluateContext& context,
397+
const model::PipelineInputOutputVector& inputs) const {
398+
model::PipelineInputOutputVector input_copy = inputs;
399+
std::sort(
400+
input_copy.begin(), input_copy.end(),
401+
[this, &context](const model::PipelineInputOutput& left,
402+
const model::PipelineInputOutput& right) -> bool {
403+
for (const auto& ordering : this->orders_) {
404+
const auto left_result =
405+
ordering.expr()->ToEvaluable()->Evaluate(context, left);
406+
const auto right_result =
407+
ordering.expr()->ToEvaluable()->Evaluate(context, right);
408+
409+
auto left_val = left_result.IsErrorOrUnset() ? model::MinValue()
410+
: *left_result.value();
411+
auto right_val = right_result.IsErrorOrUnset()
412+
? model::MinValue()
413+
: *right_result.value();
414+
const auto compare_result = model::Compare(left_val, right_val);
415+
if (compare_result != util::ComparisonResult::Same) {
416+
return ordering.direction() == Ordering::ASCENDING
417+
? compare_result == util::ComparisonResult::Ascending
418+
: compare_result == util::ComparisonResult::Descending;
419+
}
420+
}
421+
422+
return false;
423+
});
424+
425+
return input_copy;
426+
}
427+
376428
} // namespace api
377429
} // namespace firestore
378430
} // namespace firebase

‎Firestore/core/src/api/stages.h‎

Lines changed: 47 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
#include "Firestore/core/src/api/expressions.h"
2929
#include "Firestore/core/src/api/ordering.h"
3030
#include "Firestore/core/src/model/model_fwd.h"
31+
#include "Firestore/core/src/model/resource_path.h"
3132
#include "Firestore/core/src/nanopb/message.h"
3233

3334
namespace firebase {
@@ -65,25 +66,29 @@ class EvaluableStage : public Stage {
6566
EvaluableStage() = default;
6667
virtual ~EvaluableStage() = default;
6768

69+
virtual absl::string_view name() const = 0;
6870
virtual model::PipelineInputOutputVector Evaluate(
6971
const EvaluateContext& context,
7072
const model::PipelineInputOutputVector& inputs) const = 0;
7173
};
7274

7375
class CollectionSource : public EvaluableStage {
7476
public:
75-
explicit CollectionSource(std::string path) : path_(std::move(path)) {
76-
}
77+
explicit CollectionSource(std::string path);
7778
~CollectionSource() override = default;
7879

7980
google_firestore_v1_Pipeline_Stage to_proto() const override;
8081

82+
absl::string_view name() const override {
83+
return "collection";
84+
}
85+
8186
model::PipelineInputOutputVector Evaluate(
8287
const EvaluateContext& context,
8388
const model::PipelineInputOutputVector& inputs) const override;
8489

8590
private:
86-
std::string path_;
91+
model::ResourcePath path_;
8792
};
8893

8994
class DatabaseSource : public EvaluableStage {
@@ -92,6 +97,11 @@ class DatabaseSource : public EvaluableStage {
9297
~DatabaseSource() override = default;
9398

9499
google_firestore_v1_Pipeline_Stage to_proto() const override;
100+
101+
absl::string_view name() const override {
102+
return "database";
103+
}
104+
95105
model::PipelineInputOutputVector Evaluate(
96106
const EvaluateContext& context,
97107
const model::PipelineInputOutputVector& inputs) const override;
@@ -106,6 +116,14 @@ class CollectionGroupSource : public EvaluableStage {
106116

107117
google_firestore_v1_Pipeline_Stage to_proto() const override;
108118

119+
absl::string_view name() const override {
120+
return "collection_group";
121+
}
122+
123+
model::PipelineInputOutputVector Evaluate(
124+
const EvaluateContext& context,
125+
const model::PipelineInputOutputVector& inputs) const override;
126+
109127
private:
110128
std::string collection_id_;
111129
};
@@ -119,6 +137,10 @@ class DocumentsSource : public EvaluableStage {
119137

120138
google_firestore_v1_Pipeline_Stage to_proto() const override;
121139

140+
absl::string_view name() const override {
141+
return "documents";
142+
}
143+
122144
private:
123145
std::vector<std::string> documents_;
124146
};
@@ -158,6 +180,11 @@ class Where : public EvaluableStage {
158180
~Where() override = default;
159181

160182
google_firestore_v1_Pipeline_Stage to_proto() const override;
183+
184+
absl::string_view name() const override {
185+
return "where";
186+
}
187+
161188
model::PipelineInputOutputVector Evaluate(
162189
const EvaluateContext& context,
163190
const model::PipelineInputOutputVector& inputs) const override;
@@ -213,6 +240,11 @@ class LimitStage : public EvaluableStage {
213240
~LimitStage() override = default;
214241

215242
google_firestore_v1_Pipeline_Stage to_proto() const override;
243+
244+
absl::string_view name() const override {
245+
return "limit";
246+
}
247+
216248
model::PipelineInputOutputVector Evaluate(
217249
const EvaluateContext& context,
218250
const model::PipelineInputOutputVector& inputs) const override;
@@ -255,6 +287,18 @@ class SortStage : public EvaluableStage {
255287

256288
google_firestore_v1_Pipeline_Stage to_proto() const override;
257289

290+
absl::string_view name() const override {
291+
return "sort";
292+
}
293+
294+
model::PipelineInputOutputVector Evaluate(
295+
const EvaluateContext& context,
296+
const model::PipelineInputOutputVector& inputs) const override;
297+
298+
const std::vector<Ordering>& orders() const {
299+
return orders_;
300+
}
301+
258302
private:
259303
std::vector<Ordering> orders_;
260304
};

‎Firestore/core/src/core/expressions_eval.h‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,7 @@
2424

2525
#include "Firestore/core/src/api/expressions.h"
2626
#include "Firestore/core/src/api/stages.h"
27-
#include "Firestore/core/src/model/value_util.h"
2827
#include "Firestore/core/src/nanopb/message.h"
29-
#include "Firestore/core/src/util/hard_assert.h"
3028
#include "absl/types/optional.h"
3129

3230
namespace firebase {

‎Firestore/core/src/core/pipeline_run.cc‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
#include "Firestore/core/src/api/stages.h"
2222
#include "Firestore/core/src/model/mutable_document.h"
2323
#include "Firestore/core/src/remote/serializer.h"
24+
#include "pipeline_util.h"
2425

2526
namespace firebase {
2627
namespace firestore {
@@ -29,8 +30,12 @@ namespace core {
2930
std::vector<model::MutableDocument> RunPipeline(
3031
api::RealtimePipeline& pipeline,
3132
const std::vector<model::MutableDocument>& inputs) {
32-
auto& current = const_cast<std::vector<model::MutableDocument>&>(inputs);
33-
for (const auto& stage : pipeline.stages()) {
33+
if (pipeline.rewritten_stages().empty()) {
34+
pipeline.SetRewrittentStages(RewriteStages(pipeline.stages()));
35+
}
36+
37+
auto current = std::vector<model::MutableDocument>(inputs);
38+
for (const auto& stage : pipeline.rewritten_stages()) {
3439
current = stage->Evaluate(pipeline.evaluate_context(), current);
3540
}
3641

‎Firestore/core/src/core/pipeline_util.cc‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
/*
2+
* Copyright 2025 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "Firestore/core/src/core/pipeline_util.h"
18+
19+
#include "Firestore/core/src/api/expressions.h"
20+
#include "Firestore/core/src/api/realtime_pipeline.h"
21+
#include "Firestore/core/src/api/stages.h"
22+
#include "Firestore/core/src/model/mutable_document.h"
23+
#include "Firestore/core/src/remote/serializer.h"
24+
25+
namespace firebase {
26+
namespace firestore {
27+
namespace core {
28+
29+
namespace {
30+
31+
auto NewKeyOrdering() {
32+
return api::Ordering(
33+
std::make_shared<api::Field>(model::FieldPath::KeyFieldPath()),
34+
api::Ordering::Direction::ASCENDING);
35+
}
36+
37+
} // namespace
38+
39+
std::vector<std::shared_ptr<api::EvaluableStage>> RewriteStages(
40+
const std::vector<std::shared_ptr<api::EvaluableStage>>& stages) {
41+
bool has_order = false;
42+
std::vector<std::shared_ptr<api::EvaluableStage>> new_stages;
43+
for (const auto& stage : stages) {
44+
// For stages that provide ordering semantics
45+
if (stage->name() == "sort") {
46+
auto sort_stage = std::static_pointer_cast<api::SortStage>(stage);
47+
has_order = true;
48+
49+
// Ensure we have a stable ordering
50+
bool includes_key_ordering = false;
51+
for (const auto& order : sort_stage->orders()) {
52+
auto field = dynamic_cast<const api::Field*>(order.expr());
53+
if (field != nullptr && field->field_path().IsKeyFieldPath()) {
54+
includes_key_ordering = true;
55+
break;
56+
}
57+
}
58+
59+
if (includes_key_ordering) {
60+
new_stages.push_back(stage);
61+
} else {
62+
auto copy = sort_stage->orders();
63+
copy.push_back(NewKeyOrdering());
64+
new_stages.push_back(std::make_shared<api::SortStage>(std::move(copy)));
65+
}
66+
}
67+
// For stages whose semantics depend on ordering
68+
else if (stage->name() == "limit") {
69+
if (!has_order) {
70+
new_stages.push_back(std::make_shared<api::SortStage>(
71+
std::vector<api::Ordering>{NewKeyOrdering()}));
72+
has_order = true;
73+
}
74+
new_stages.push_back(stage);
75+
}
76+
// TODO(wuandy): Handle add_fields and select and such
77+
else {
78+
new_stages.push_back(stage);
79+
}
80+
}
81+
82+
if (!has_order) {
83+
new_stages.push_back(std::make_shared<api::SortStage>(
84+
std::vector<api::Ordering>{NewKeyOrdering()}));
85+
}
86+
87+
return new_stages;
88+
}
89+
90+
} // namespace core
91+
} // namespace firestore
92+
} // namespace firebase

0 commit comments

Comments
 (0)