feat(firestore): Implement Search pipeline stage and related expressions (#14383)

**Summary**
This PR implements the `search` pipeline stage and its associated
expressions in the Go Firestore SDK, providing parity with the Java
(googleapis/java-firestore#2346) and Node.js
(googleapis/google-cloud-node#7824) SDKs for
full-text and geospatial search capabilities within Firestore Pipelines.

**Key Changes**

*   **Search Pipeline Stage**:
* Implemented the `Search` stage in `Pipeline` with a functional options
pattern.
* Introduced `WithSearchQuery`, `WithSearchSort`, `WithSearchAddFields`,
and `WithSearchRetrievalDepth` options.
* Ensured protobuf encoding places the `query` within the `Options` map
(not `Args`) to align with backend requirements and the Java SDK
implementation.
*   **Search Expressions**:
* Implemented `DocumentMatches(query string)` and `Matches(field, query
string)` for full-text search.
* Implemented `GeoDistance(field, location)` for geospatial distance
calculations.
    *   Implemented `Score()` to retrieve the search topicality score.
* Implemented `Snippet(field, query)` for highlighted search result
snippets.
*   **Fluent API**:
* Added `GeoDistance(location)`, `Matches(query)`, and `Snippet(query)`
methods to the `Expression` interface to support fluent chaining (e.g.,
`FieldOf("location").GeoDistance(loc)`).
*   **Testing**:
* Added comprehensive unit tests in `pipeline_test.go` and
`pipeline_function_test.go`.
* Verified backend request structure consistency via JSON dump
comparisons.

**Usage Example**
```go
pipeline := client.Pipeline().
    Collection("restaurants").
    Search(
        WithSearchQuery(DocumentMatches("waffles OR pancakes")),
        WithSearchSort(Descending(Score())),
        WithSearchAddFields(Snippet("menu", "waffles").As("highlight")),
    )
```

Author: bhshkh

firestore/integration_test.go

@@ -184,6 +184,8 @@ func (RawOptions) isLiteralsOption() {}
 
 func (RawOptions) isDefineOption() {}
 
+func (RawOptions) isSearchOption() {}
+
 func (r RawOptions) apply(eo *executeSettings) {
 	if eo.RawOptions == nil {
 		eo.RawOptions = make(map[string]any)
@@ -1125,6 +1127,120 @@ func (p *Pipeline) FindNearest(vectorField any, queryVector any, measure Pipelin
 	return p.append(stage)
 }
 
+// SearchOption is an option for a Search pipeline stage.
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+type SearchOption interface {
+	StageOption
+	isSearchOption()
+}
+
+type funcSearchOption struct {
+	f func(map[string]any)
+}
+
+func (fso *funcSearchOption) applyStage(so map[string]any) {
+	fso.f(so)
+}
+
+func (*funcSearchOption) isSearchOption() {}
+
+func newFuncSearchOption(f func(map[string]any)) *funcSearchOption {
+	return &funcSearchOption{
+		f: f,
+	}
+}
+
+// WithSearchQuery specifies the search query that will be used to query and score documents by the search stage.
+// It can be a string (automatically wrapped in DocumentMatches) or a BooleanExpression.
+//
+// Example:
+//
+//	client.Pipeline().Collection("restaurants").Search(
+//		WithSearchQuery("waffles"),
+//	)
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func WithSearchQuery(query any) SearchOption {
+	return newFuncSearchOption(func(so map[string]any) {
+		so["query"] = query
+	})
+}
+
+// WithSearchSort specifies how the returned documents are sorted. One or more ordering are required.
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func WithSearchSort(orders ...Ordering) SearchOption {
+	return newFuncSearchOption(func(so map[string]any) {
+		t, ok := so["sort"].([]Ordering)
+		if !ok {
+			t = []Ordering{}
+		}
+		so["sort"] = append(t, orders...)
+	})
+}
+
+// WithSearchAddFields specifies the fields to add to each document.
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func WithSearchAddFields(fields ...Selectable) SearchOption {
+	return newFuncSearchOption(func(so map[string]any) {
+		t, ok := so["add_fields"].([]Selectable)
+		if !ok {
+			t = []Selectable{}
+		}
+		so["add_fields"] = append(t, fields...)
+	})
+}
+
+// WithSearchRetrievalDepth specifies the maximum number of documents to retrieve. Documents will be retrieved in the
+// pre-sort order specified by the search index.
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func WithSearchRetrievalDepth(depth int64) SearchOption {
+	return newFuncSearchOption(func(so map[string]any) {
+		so["retrieval_depth"] = depth
+	})
+}
+
+// Search adds a search stage to the Pipeline.
+// This must be the first stage of the pipeline.
+// A limited set of expressions are supported in the search stage.
+// Use [WithSearchQuery] to specify the search query.
+//
+// Example:
+//
+//	client.Pipeline().Collection("restaurants").Search(
+//		WithSearchQuery(DocumentMatches("waffles OR pancakes")),
+//		WithSearchSort(Descending(Score())),
+//		WithSearchRetrievalDepth(10),
+//	)
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func (p *Pipeline) Search(opts ...SearchOption) *Pipeline {
+	if p.err != nil {
+		return p
+	}
+	options := make(map[string]any)
+	for _, opt := range opts {
+		if opt != nil {
+			opt.applyStage(options)
+		}
+	}
+	stage, err := newSearchStage(options)
+	if err != nil {
+		p.err = err
+		return p
+	}
+	return p.append(stage)
+}
+
 // RawStage adds a generic stage to the pipeline.
 // This method provides a flexible way to extend the pipeline's functionality by adding custom stages.
 //

firestore/pipeline.go

@@ -16,6 +16,7 @@ package firestore
 
 import (
 	pb "cloud.google.com/go/firestore/apiv1/firestorepb"
+	"google.golang.org/genproto/googleapis/type/latlng"
 )
 
 // Selectable is an interface for expressions that can be selected in a pipeline.
@@ -508,6 +509,22 @@ type Expression interface {
 	// Descending creates an ordering expression for descending order.
 	Descending() Ordering
 
+	// GeoDistance creates an expression that evaluates to the distance in meters between the location in the expression and the query location.
+	//
+	// The parameter 'location' is the query location.
+	//
+	// Example:
+	//
+	//	client.Pipeline().Collection("restaurants").
+	//		Search(
+	//			WithSearchQuery("waffles"),
+	//			WithSearchSort(Ascending(FieldOf("location").GeoDistance(&latlng.LatLng{Latitude: 37.0, Longitude: -122.0}))),
+	//		)
+	//
+	// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+	// regardless of any other documented package stability guarantees.
+	GeoDistance(location *latlng.LatLng) Expression
+
 	// As assigns an alias to an expression.
 	// Aliases are useful for renaming fields in the output of a stage.
 	As(alias string) *AliasedExpression
@@ -760,6 +777,10 @@ func (b *baseExpression) VectorLength() Expression               { return Vector
 func (b *baseExpression) Ascending() Ordering  { return Ascending(b) }
 func (b *baseExpression) Descending() Ordering { return Descending(b) }
 
+func (b *baseExpression) GeoDistance(location *latlng.LatLng) Expression {
+	return GeoDistance(b, location)
+}
+
 func (b *baseExpression) As(alias string) *AliasedExpression {
 	return newAliasedExpr(b, alias)
 }

firestore/pipeline_expression.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 
 	pb "cloud.google.com/go/firestore/apiv1/firestorepb"
+	"google.golang.org/genproto/googleapis/type/latlng"
 )
 
 // FunctionExpression represents Firestore [Pipeline] functions, which can be evaluated within pipeline
@@ -1154,3 +1155,57 @@ func RegexFindAll(exprOrField any, pattern any) Expression {
 func Rand() Expression {
 	return newBaseFunction("rand", nil)
 }
+
+// DocumentMatches creates a boolean expression that performs a full-text search on all indexed search fields in the document.
+//
+// This Expression can only be used within a Search stage.
+//
+// Example:
+//
+//	client.Pipeline().Collection("restaurants").
+//		Search(WithSearchQuery(DocumentMatches("waffles OR pancakes")))
+//
+// - query: Define the search query using the search domain-specific language (DSL).
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func DocumentMatches(query string) BooleanExpression {
+	return &baseBooleanExpression{baseFunction: newBaseFunction("document_matches", []Expression{ConstantOf(query)})}
+}
+
+// GeoDistance creates an expression that evaluates to the distance in meters between the location in the specified field and the query location.
+//
+// This Expression can only be used within a Search stage.
+//
+// Example:
+//
+//	client.Pipeline().Collection("restaurants").
+//		Search(
+//			WithSearchQuery("waffles"),
+//			WithSearchSort(Ascending(GeoDistance("location", &latlng.LatLng{Latitude: 37.0, Longitude: -122.0}))),
+//		)
+//
+// - field: Specifies the field in the document which contains the GeoPoint for distance computation. It can be a field path string, [FieldPath] or [Expression].
+// - location: Compute distance to this GeoPoint.
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func GeoDistance(field any, location *latlng.LatLng) Expression {
+	return newBaseFunction("geo_distance", []Expression{asFieldExpr(field), ConstantOf(location)})
+}
+
+// Score creates an expression that evaluates to the search score that reflects the topicality of the document to all of the text
+// predicates (for example: DocumentMatches) in the search query.
+//
+// This Expression can only be used within a Search stage.
+//
+// Example:
+//
+//	client.Pipeline().Collection("restaurants").
+//		Search(WithSearchQuery("waffles"), WithSearchSort(Descending(Score())))
+//
+// Experimental: Firestore Pipelines is currently in preview and is subject to potential breaking changes in future versions,
+// regardless of any other documented package stability guarantees.
+func Score() Expression {
+	return newBaseFunction("score", nil)
+}

firestore/pipeline_function.go

@@ -19,6 +19,7 @@ import (
 
 	pb "cloud.google.com/go/firestore/apiv1/firestorepb"
 	"cloud.google.com/go/internal/testutil"
+	"google.golang.org/genproto/googleapis/type/latlng"
 )
 
 func TestTruncFunctions(t *testing.T) {
@@ -1118,3 +1119,76 @@ func TestGetFieldVariations(t *testing.T) {
 		t.Fatal("expected expr6 not to be nil")
 	}
 }
+
+func TestSearchFunctions(t *testing.T) {
+	// 1. DocumentMatches
+	// 2. GeoDistance
+	// 3. Score
+	// 4. GeoDistance method
+	// 8. Matches method
+	testcases := []struct {
+		desc string
+		expr Expression
+		want *pb.Value
+	}{
+		{
+			desc: "DocumentMatches",
+			expr: DocumentMatches("waffles"),
+			want: &pb.Value{ValueType: &pb.Value_FunctionValue{
+				FunctionValue: &pb.Function{
+					Name: "document_matches",
+					Args: []*pb.Value{
+						{ValueType: &pb.Value_StringValue{StringValue: "waffles"}},
+					},
+				},
+			}},
+		},
+		{
+			desc: "GeoDistance",
+			expr: GeoDistance("location", &latlng.LatLng{Latitude: 37.0, Longitude: -122.0}),
+			want: &pb.Value{ValueType: &pb.Value_FunctionValue{
+				FunctionValue: &pb.Function{
+					Name: "geo_distance",
+					Args: []*pb.Value{
+						{ValueType: &pb.Value_FieldReferenceValue{FieldReferenceValue: "location"}},
+						{ValueType: &pb.Value_GeoPointValue{GeoPointValue: &latlng.LatLng{Latitude: 37.0, Longitude: -122.0}}},
+					},
+				},
+			}},
+		},
+		{
+			desc: "Score",
+			expr: Score(),
+			want: &pb.Value{ValueType: &pb.Value_FunctionValue{
+				FunctionValue: &pb.Function{
+					Name: "score",
+				},
+			}},
+		},
+		{
+			desc: "GeoDistance method",
+			expr: FieldOf("location").GeoDistance(&latlng.LatLng{Latitude: 37.0, Longitude: -122.0}),
+			want: &pb.Value{ValueType: &pb.Value_FunctionValue{
+				FunctionValue: &pb.Function{
+					Name: "geo_distance",
+					Args: []*pb.Value{
+						{ValueType: &pb.Value_FieldReferenceValue{FieldReferenceValue: "location"}},
+						{ValueType: &pb.Value_GeoPointValue{GeoPointValue: &latlng.LatLng{Latitude: 37.0, Longitude: -122.0}}},
+					},
+				},
+			}},
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.desc, func(t *testing.T) {
+			got, err := tc.expr.toProto()
+			if err != nil {
+				t.Fatalf("toProto() failed: %v", err)
+			}
+			if diff := testutil.Diff(got, tc.want); diff != "" {
+				t.Errorf("toProto() returned diff (-got +want): %s", diff)
+			}
+		})
+	}
+}

firestore/pipeline_function_test.go

@@ -856,7 +856,6 @@ func TestIntegration_PipelineStages(t *testing.T) {
 		}
 	})
 	t.Run("Update", func(t *testing.T) {
-		t.Skip("Skipping test until feature is available in PROD")
 		updateIter := client.Pipeline().Collection(coll.ID).
 			Where(Equal(FieldOf("author.country"), "UK")).
 			Update(WithUpdateTransformations(ConstantOf("Active").As("status"))).
@@ -878,7 +877,6 @@ func TestIntegration_PipelineStages(t *testing.T) {
 		}
 	})
 	t.Run("Delete", func(t *testing.T) {
-		t.Skip("Skipping test until feature is available in PROD")
 		deleteIter := client.Pipeline().Collection(coll.ID).Where(Equal(FieldOf("title"), "The Great Gatsby")).Delete().Execute(ctx).Results()
 		defer deleteIter.Stop()
 		_, err := deleteIter.GetAll()
@@ -903,7 +901,6 @@ func TestIntegration_PipelineFunctions(t *testing.T) {
 	t.Run("arrayFuncs", arrayFuncs)
 	t.Run("stringFuncs", stringFuncs)
 	t.Run("vectorFuncs", vectorFuncs)
-
 	t.Run("timestampFuncs", timestampFuncs)
 	t.Run("arithmeticFuncs", arithmeticFuncs)
 	t.Run("aggregateFuncs", aggregateFuncs)
@@ -945,6 +942,8 @@ func aggregationFuncs(t *testing.T) {
 		Aggregate(Accumulators(
 			First("val").As("first_val"),
 			Last("val").As("last_val"),
+			Maximum("val").As("max_val"),
+			Minimum("val").As("min_val"),
 			ArrayAgg("val").As("all_vals"),
 			ArrayAggDistinct("val").As("distinct_vals"),
 			CountDistinct("val").As("distinct_count_val"),
@@ -961,6 +960,13 @@ func aggregationFuncs(t *testing.T) {
 
 	data := res.Data()
 
+	if data["max_val"] != int64(2) {
+		t.Errorf("got max_val %v, want 2", data["max_val"])
+	}
+	if data["min_val"] != int64(1) {
+		t.Errorf("got min_val %v, want 1", data["min_val"])
+	}
+
 	// Check ArrayAgg "all_vals" -> [1, 2, 1] (order irrelevant)
 	allValsRaw, ok := data["all_vals"].([]interface{})
 	if !ok {
@@ -1326,6 +1332,11 @@ func objectFuncs(t *testing.T) {
 			pipeline: client.Pipeline().Collection(coll.ID).Select(Fields(MapGet("m1", "a").As("value"))),
 			want:     map[string]interface{}{"value": int64(1)},
 		},
+		{
+			name:     "GetField",
+			pipeline: client.Pipeline().Collection(coll.ID).Select(Fields(GetField("m1", "a").As("value"))),
+			want:     map[string]interface{}{"value": int64(1)},
+		},
 		{
 			name:     "MapMerge",
 			pipeline: client.Pipeline().Collection(coll.ID).Select(Fields(MapMerge("m1", FieldOf("m2")).As("merged"))),
@@ -2427,6 +2438,13 @@ func comparisonFuncs(t *testing.T) {
 				Where(NotEqual("a", 2)),
 			want: []map[string]interface{}{doc1want},
 		},
+		{
+			name: "GreaterThan",
+			pipeline: client.Pipeline().
+				Collection(coll.ID).
+				Where(GreaterThan("a", 1)),
+			want: []map[string]interface{}{{"a": int64(2), "b": int64(2), "c": int64(-3), "d": float64(4.5), "e": float64(-5.5), "timestamp": now.Truncate(time.Microsecond)}},
+		},
 		{
 			name: "LessThan",
 			pipeline: client.Pipeline().