如何使用 Atlas Search $search 查询运行 $unionWith

1
using MongoDB.Bson;
2
using MongoDB.Driver;
3
using MongoDB.Driver.Search;
4
5
public class Program
6
{
7
    public static void Main(string[] args)
8
    {
9
        // connect to your Atlas cluster
10
        string connectionString = "<connection-string>";
11
        var client = new MongoClient(connectionString);
12
        
13
        // define namespace
14
        var database = client.GetDatabase("sample_training");
15
        var collection = database.GetCollection<BsonDocument>("companies");
16
        
17
        // define pipeline stage
18
        var searchStage1 = new BsonDocument("$search", new BsonDocument{{ "text", new BsonDocument
19
            {{ "query", "Mobile" },{ "path", "name" }}
20
        }});
21
        var projectStage1 = new BsonDocument("$project", new BsonDocument{
22
            { "score", new BsonDocument("$meta", "searchScore") },
23
            { "_id", 0 },{ "number_of_employees", 1 },{ "founded_year", 1 },{ "name", 1 }
24
        });
25
        var setStage1 = new BsonDocument("$set", new BsonDocument{{ "source", "companies" }});
26
        var limitStage1 = new BsonDocument("$limit", 3);
27
        
28
        // define subpipeline
29
        var searchStage2 = new BsonDocument("$search", new BsonDocument{{ "text", new BsonDocument
30
            {{ "query", "Mobile" },{ "path", "business_name" }}
31
        }});
32
        var setStage2 = new BsonDocument("$set", new BsonDocument{ { "source", "inspections" } });
33
        var projectStage2 = new BsonDocument("$project", new BsonDocument{
34
            { "score", new BsonDocument("$meta", "searchScore") }, 
35
            { "source", 1 }, { "_id", 0 }, { "business_name", 1 }, { "address", 1 }
36
        });
37
        var limitStage2 = new BsonDocument("$limit", 3);
38
        var sortStage2 = new BsonDocument("$sort", new BsonDocument{{ "score", -1 }});
39
        var unionWithPipeline = new List<BsonDocument>{searchStage2, setStage2, projectStage2, limitStage2, sortStage2};
40
        var unionWithStage = new BsonDocument("$unionWith", new BsonDocument
41
        {
42
            { "coll", "inspections" },
43
            { "pipeline", new BsonArray(unionWithPipeline) }
44
        });
45
        var aggregationPipeline = new List<BsonDocument> {searchStage1, projectStage1, setStage1, limitStage1,unionWithStage};
46
        
47
        // run pipeline
48
        var result = collection.Aggregate<BsonDocument>(aggregationPipeline).ToList();
49
        
50
        //print results
51
        foreach (var document in result)
52
        {
53
            Console.WriteLine(document);
54
        }
55
    }
56
}

1
using MongoDB.Bson;
2
using MongoDB.Driver;
3
4
public class Program
5
{
6
    public static void Main(string[] args)
7
    {
8
        // connect to your Atlas cluster
9
        var client = new MongoClient("<connection-string>");
10
11
        // define namespace
12
        var database = client.GetDatabase("sample_training");
13
        var collection = database.GetCollection<BsonDocument>("companies");
14
15
        // define pipeline
16
        var pipeline = new BsonDocument[]
17
        {
18
        new BsonDocument("$search", new BsonDocument{
19
          { "text", new BsonDocument{
20
            { "query", "mobile" }, { "path", "name" },
21
            { "score", new BsonDocument{
22
                { "boost", new BsonDocument{ { "value", 1.6 } }}
23
            }}
24
          }}
25
        }),
26
        new BsonDocument("$project", new BsonDocument{
27
          { "score", new BsonDocument("$meta", "searchScore") },
28
          { "_id", 0 }, 
29
          { "number_of_employees", 1 }, { "founded_year", 1 }, { "name", 1 }
30
        }),
31
        new BsonDocument("$addFields", new BsonDocument{
32
          { "source", "companies" },
33
          { "source_count", "$$SEARCH_META.count.lowerBound" }
34
        }),
35
        new BsonDocument("$limit", 3),
36
        new BsonDocument("$unionWith", new BsonDocument{
37
          { "coll", "inspections" },
38
          { "pipeline", new BsonArray{
39
            new BsonDocument("$search", new BsonDocument{
40
                { "text", new BsonDocument{
41
                  { "query", "mobile" },
42
                  { "path", "business_name" }
43
                }}
44
            }),
45
            new BsonDocument("$project", new BsonDocument{
46
                { "score", new BsonDocument("$meta", "searchScore") },
47
                { "business_name", 1 }, { "address", 1 }, { "_id", 0 }
48
            }),
49
            new BsonDocument("$limit", 3),
50
            new BsonDocument("$set", new BsonDocument{
51
                { "source", "inspections" },
52
                { "source_count", "$$SEARCH_META.count.lowerBound" }
53
            }),
54
            new BsonDocument("$sort", new BsonDocument{
55
                { "score", -1 }
56
            })
57
          }}
58
        }),
59
        new BsonDocument("$facet", new BsonDocument{
60
          { "allDocs", new BsonArray() },
61
          { "totalCount", new BsonArray{
62
            new BsonDocument("$group", new BsonDocument{
63
                { "_id", "$source" },
64
                { "firstCount", new BsonDocument("$first", "$source_count") }
65
            }),
66
            new BsonDocument("$project", new BsonDocument{
67
                { "totalCount", new BsonDocument("$sum", "$firstCount") }
68
            })
69
          }}
70
        })
71
        };
72
73
        // run pipeline
74
        var result = collection.Aggregate<BsonDocument>(pipeline).ToList();
75
76
        //print results
77
        foreach (var document in result)
78
        {
79
            Console.WriteLine(document);
80
        }
81
    }
82
}

1
package main
2
import (
3
	"context"
4
	"fmt"
5
	"time"
6
7
	"go.mongodb.org/mongo-driver/bson"
8
	"go.mongodb.org/mongo-driver/mongo"
9
	"go.mongodb.org/mongo-driver/mongo/options"
10
)
11
12
func main() {
13
  var err error
14
  // connect to the Atlas cluster
15
  ctx := context.Background()
16
  client, err := mongo.Connect(ctx, options.Client().ApplyURI("<connection-string>"))
17
  if err != nil {
18
	  panic(err)
19
  }
20
  defer client.Disconnect(ctx)
21
  
22
  // set namespace
23
  collection := client.Database("sample_training").Collection("companies")
24
  // define pipeline
25
  searchStage := bson.D{{"$search", bson.D{
26
	{"text", bson.D{
27
	  {"query", "Mobile"}, {"path", "name"},
28
	}},
29
  }}}
30
  projectStage := bson.D{{"$project", bson.D{
31
    {"score", bson.D{{"$meta", "searchScore"}}},
32
    {"_id", 0},
33
    {"number_of_employees", 1},
34
    {"founded_year", 1},
35
    {"name", 1},
36
  }}}
37
  setStage := bson.D{{"$set", bson.D{{"source", "companies"}}}}
38
  limitStage := bson.D{{"$limit", 5}}
39
  uinionWithStage := bson.D{{"$unionWith", bson.D{
40
	{"coll", "inspections"},
41
	{"pipeline", bson.A{
42
	  bson.D{{"$search", bson.D{
43
		{"text", bson.D{
44
		  {"query", "Mobile"}, {"path", "business_name"},
45
		}},
46
	  }}},
47
	  bson.D{{"$set", bson.D{{"source", "inspections"}}}},
48
	  bson.D{{"$project", bson.D{
49
		{"score", bson.D{{"$meta", "searchScore"}}},
50
		{"source", 1},
51
		{"_id", 0},
52
		{"business_name", 1},
53
		{"address", 1},
54
	  }}},
55
	  bson.D{{"$limit", 3}},
56
	  bson.D{{"$sort", bson.D{{"score", -1}}}},
57
	}},
58
  }}}
59
  // specify the amount of time the operation can run on the server
60
  opts := options.Aggregate().SetMaxTime(5 * time.Second)
61
  // run pipeline
62
  cursor, err := collection.Aggregate(ctx, mongo.Pipeline{searchStage, projectStage, setStage, limitStage, uinionWithStage}, opts)
63
  if err != nil {
64
	  panic(err)
65
  }
66
  // print results
67
  var results []bson.D
68
  if err = cursor.All(context.TODO(), &results); err != nil {
69
	  panic(err)
70
  }
71
  for _, result := range results {
72
	  fmt.Println(result)
73
  }
74
}

1
package main
2
import (
3
	"context"
4
	"fmt"
5
	"time"
6
7
	"go.mongodb.org/mongo-driver/bson"
8
	"go.mongodb.org/mongo-driver/mongo"
9
	"go.mongodb.org/mongo-driver/mongo/options"
10
)
11
12
func main() {
13
  var err error
14
  // connect to the Atlas cluster
15
  ctx := context.Background()
16
  client, err := mongo.Connect(ctx, options.Client().ApplyURI("<connection-string>"))
17
  if err != nil {
18
	  panic(err)
19
  }
20
  defer client.Disconnect(ctx)
21
  // set namespace
22
  collection := client.Database("sample_training").Collection("companies")
23
  // define pipeline
24
  searchStage := bson.D{{"$search", bson.D{
25
	{"text", bson.D{
26
	  {"query", "Mobile"}, {"path", "name"}, {"score", bson.D{{"boost", bson.D{{"value", 1.6}}}}},
27
	}},
28
  }}}
29
  projectStage := bson.D{{"$project", bson.D{
30
    {"score", bson.D{{"$meta", "searchScore"}}},
31
    {"_id", 0},
32
    {"number_of_employees", 1},
33
    {"founded_year", 1},
34
    {"name", 1},
35
  }}}
36
  addFieldsStage := bson.D{{"$set", bson.D{
37
	{"source", "companies"},
38
    {"source_count", "$$SEARCH_META.count.lowerBound"},
39
  }}}
40
  limitStage := bson.D{{"$limit", 3}}
41
  uinionWithStage := bson.D{{"$unionWith", bson.D{
42
	{"coll", "inspections"},
43
	{"pipeline", bson.A{
44
      bson.D{{"$search", bson.D{
45
        {"text", bson.D{
46
          {"query", "mobile"}, {"path", "business_name"},
47
        }},
48
      }}},
49
      bson.D{{"$project", bson.D{
50
        {"score", bson.D{{"$meta", "searchScore"}}},
51
        {"business_name", 1},
52
        {"address", 1},
53
        {"_id", 0},
54
      }}},
55
      bson.D{{"$limit", 3}},
56
      bson.D{{"$set", bson.D{
57
        {"source", "inspections"},
58
        {"source_count", "$$SEARCH_META.count.lowerBound"},
59
      }}},
60
      bson.D{{"$sort", bson.D{{"score", -1}}}},
61
    }},
62
  }}}
63
  facetStage := bson.D{{"$facet", bson.D{
64
	{"allDocs", bson.A{}},
65
	{"totalCount",  bson.A{
66
	  bson.D{
67
		{"$group", bson.D{
68
		  {"_id", "$source"},
69
		  {"firstCount", bson.D{{"$first", "$source_count"}}},
70
		}},
71
	  },
72
	  bson.D{{"$project", bson.D{{"totalCount", bson.D{{"$sum", "$firstCount"}}}}}},
73
	}},
74
  }}}
75
  // specify the amount of time the operation can run on the server
76
  opts := options.Aggregate().SetMaxTime(5 * time.Second)
77
  // run pipeline
78
  cursor, err := collection.Aggregate(ctx, mongo.Pipeline{searchStage, projectStage, addFieldsStage, limitStage, uinionWithStage, facetStage}, opts)
79
  if err != nil {
80
	  panic(err)
81
  }
82
  // print results
83
  var results []bson.D
84
  if err = cursor.All(context.TODO(), &results); err != nil {
85
	  panic(err)
86
  }
87
  for _, result := range results {
88
	  fmt.Println(result)
89
  }
90
}

1
import com.mongodb.client.MongoClients;
2
import com.mongodb.client.MongoClient;
3
import com.mongodb.client.MongoDatabase;
4
import org.bson.Document;
5
import java.util.ArrayList;
6
import java.util.Arrays;
7
import java.util.List;
8
9
public class SearchWithUnionwith {
10
  public static void main(String[] args) {
11
  // connect to Atlas cluster
12
    try (MongoClient mongoClient = MongoClients.create("<connection-string>")) {
13
      // get database name
14
      MongoDatabase database = mongoClient.getDatabase("sample_training");
15
      // define pipeline
16
      List<Document> pipeline1 = Arrays.asList(
17
        new Document("$search", new Document("text",
18
          new Document("query", "Mobile")
19
            .append("path", "name"))),
20
        new Document("$project", new Document("score",
21
          new Document("$meta", "searchScore"))
22
            .append("_id", 0)
23
            .append("number_of_employees", 1)
24
            .append("founded_year", 1)
25
            .append("name", 1)),
26
        new Document("$set", new Document("source", "companies")),
27
        new Document("$limit", 3)
28
      );
29
30
      List<Document> pipeline2 = Arrays.asList(
31
        new Document("$search", new Document("text",
32
          new Document("query", "Mobile")
33
            .append("path", "business_name"))),
34
        new Document("$set", new Document("source", "inspections")),
35
        new Document("$project", new Document("score",
36
          new Document("$meta", "searchScore"))
37
            .append("source", 1)
38
            .append("_id", 0)
39
            .append("business_name", 1)
40
            .append("address", 1)),
41
        new Document("$limit", 3),
42
        new Document("$sort", new Document("score", -1))
43
      );
44
45
      List<Document> unionWithStage = new ArrayList<>();
46
      Document unionWith = new Document("$unionWith", new Document("coll", "inspections")
47
        .append("pipeline", pipeline2));
48
      unionWithStage.add(unionWith);
49
50
      List<Document> finalPipeline = new ArrayList<>(pipeline1);
51
      finalPipeline.addAll(unionWithStage);
52
      // run pipeline and print results
53
      database.getCollection("companies").aggregate(finalPipeline)
54
        .forEach(doc -> System.out.println(doc.toJson()));
55
    }
56
  }
57
}

1
import com.mongodb.client.MongoClients;
2
import com.mongodb.client.MongoCollection;
3
import com.mongodb.client.MongoClient;
4
import org.bson.Document;
5
6
public class SearchWithUnionwith {
7
  public static void main(String[] args) {
8
    // connect to Atlas cluster
9
    try (MongoClient mongoClient = MongoClients.create("<connection-string>")) {
10
      // define namespace
11
      MongoCollection<Document> collection = mongoClient.getDatabase("sample_training").getCollection("companies");
12
      // define pipeline
13
      Document searchStage = new Document("$search", new Document("text",
14
        new Document("query", "mobile")
15
          .append("path", "name")
16
          .append("score", new Document("boost", new Document("value", 1.6)))
17
        )
18
      );
19
20
      Document projectStage = new Document("$project", new Document("score", new Document("$meta", "searchScore"))
21
        .append("_id", 0)
22
        .append("number_of_employees", 1)
23
        .append("founded_year", 1)
24
        .append("name", 1)
25
      );
26
27
      Document addFieldsStage = new Document("$addFields", new Document("source", "companies")
28
        .append("source_count", "$$SEARCH_META.count.lowerBound")
29
      );
30
31
      Document limitStage = new Document("$limit", 3);
32
33
      Document unionWithStage = new Document("$unionWith", new Document("coll", "inspections")
34
        .append("pipeline", java.util.Arrays.asList(
35
          new Document("$search", new Document("text",
36
            new Document("query", "mobile")
37
              .append("path", "business_name")
38
          )),
39
          new Document("$project", new Document("score", new Document("$meta", "searchScore"))
40
            .append("business_name", 1)
41
            .append("address", 1)
42
            .append("_id", 0)
43
          ),
44
          new Document("$limit", 3),
45
          new Document("$set", new Document("source", "inspections")
46
            .append("source_count", "$$SEARCH_META.count.lowerBound")
47
          ),
48
          new Document("$sort", new Document("score", -1))
49
        ))
50
      );
51
52
      Document facetStage = new Document("$facet", new Document("allDocs", java.util.Arrays.asList())
53
        .append("totalCount", java.util.Arrays.asList(
54
          new Document("$group", new Document("_id", "$source")
55
            .append("firstCount", new Document("$first", "$source_count"))
56
          ),
57
          new Document("$project", new Document("totalCount",
58
            new Document("$sum", "$firstCount")
59
          ))
60
        ))
61
      );
62
      // run pipeline and print results
63
      collection.aggregate(java.util.Arrays.asList(
64
        searchStage, projectStage, addFieldsStage, limitStage, unionWithStage, facetStage
65
      )).forEach(doc -> System.out.println(doc.toJson()));
66
    }
67
  }
68
}

1
import com.mongodb.kotlin.client.coroutine.MongoClient
2
import kotlinx.coroutines.runBlocking
3
import org.bson.Document
4
5
fun main() {
6
    // connect to Atlas cluster
7
    val uri = "<connection-string>"
8
    val mongoClient = MongoClient.create(uri)
9
10
    // set namespace
11
    val database = mongoClient.getDatabase("sample_training")
12
    val collection = database.getCollection<Document>("companies")
13
14
    runBlocking {
15
        // define pipeline
16
        val pipeline1 = listOf(
17
            Document("\$search", Document("text",
18
                Document("query", "Mobile")
19
                    .append("path", "name"))), Document("\$project", Document("score",
20
                Document("\$meta", "searchScore"))
21
                .append("_id", 0)
22
                .append("number_of_employees", 1)
23
                .append("founded_year", 1)
24
                .append("name", 1)), Document("\$set", Document("source", "companies")),
25
            Document("\$limit", 3)
26
        )
27
28
        val pipeline2 = listOf(
29
            Document(
30
                "\$search", Document(
31
                    "text",
32
                    Document("query", "Mobile")
33
                        .append("path", "business_name")
34
                )
35
            ),
36
            Document("\$set", Document("source", "inspections")),
37
            Document(
38
                "\$project", Document(
39
                    "score",
40
                    Document("\$meta", "searchScore")
41
                )
42
                    .append("source", 1)
43
                    .append("_id", 0)
44
                    .append("business_name", 1)
45
                    .append("address", 1)
46
            ),
47
            Document("\$limit", 3),
48
            Document("\$sort", Document("score", -1))
49
        )
50
51
        val unionWithStage: MutableList<Document> = ArrayList()
52
        val unionWith = Document(
53
            "\$unionWith", Document("coll", "inspections")
54
                .append("pipeline", pipeline2)
55
        )
56
        unionWithStage.add(unionWith)
57
        val finalPipeline: MutableList<Document> = ArrayList(pipeline1)
58
        finalPipeline.addAll(unionWithStage)
59
60
        // run pipeline and print results
61
        val resultsFlow = collection.aggregate<Document>(finalPipeline)
62
        resultsFlow.collect { println(it) }
63
64
    }
65
    mongoClient.close()
66
}

1
import com.mongodb.kotlin.client.coroutine.MongoClient
2
import kotlinx.coroutines.runBlocking
3
import org.bson.Document
4
import java.util.*
5
6
fun main() {
7
    // connect to Atlas cluster
8
    val uri = "<connection-string>"
9
    val mongoClient = MongoClient.create(uri)
10
11
    // set namespace
12
    val database = mongoClient.getDatabase("sample_training")
13
    val collection = database.getCollection<Document>("companies")
14
15
    runBlocking {
16
        // define pipeline stages
17
        val searchStage = Document(
18
            "\$search", Document(
19
                "text",
20
                Document("query", "mobile")
21
                    .append("path", "name")
22
                    .append("score", Document("boost", Document("value", 1.6)))
23
            )
24
        )
25
26
        val projectStage = Document(
27
            "\$project", Document("score", Document("\$meta", "searchScore"))
28
                .append("_id", 0)
29
                .append("number_of_employees", 1)
30
                .append("founded_year", 1)
31
                .append("name", 1)
32
        )
33
34
        val addFieldsStage = Document(
35
            "\$addFields", Document("source", "companies")
36
                .append("source_count", "$\$SEARCH_META.count.lowerBound")
37
        )
38
39
        val limitStage = Document("\$limit", 3)
40
41
        val unionWithStage = Document(
42
            "\$unionWith", Document("coll", "inspections")
43
                .append(
44
                    "pipeline", Arrays.asList(
45
                        Document(
46
                            "\$search", Document(
47
                                "text",
48
                                Document("query", "mobile")
49
                                    .append("path", "business_name")
50
                            )
51
                        ),
52
                        Document(
53
                            "\$project", Document("score", Document("\$meta", "searchScore"))
54
                                .append("business_name", 1)
55
                                .append("address", 1)
56
                                .append("_id", 0)
57
                        ),
58
                        Document("\$limit", 3),
59
                        Document(
60
                            "\$set", Document("source", "inspections")
61
                                .append("source_count", "$\$SEARCH_META.count.lowerBound")
62
                        ),
63
                        Document("\$sort", Document("score", -1))
64
                    )
65
                )
66
        )
67
68
        val facetStage = Document(
69
            "\$facet", Document("allDocs", Arrays.asList<Any>())
70
                .append(
71
                    "totalCount", Arrays.asList(
72
                        Document(
73
                            "\$group", Document("_id", "\$source")
74
                                .append("firstCount", Document("\$first", "\$source_count"))
75
                        ),
76
                        Document(
77
                            "\$project", Document(
78
                                "totalCount",
79
                                Document("\$sum", "\$firstCount")
80
                            )
81
                        )
82
                    )
83
                )
84
        )
85
86
        // run pipeline and print results
87
        val resultsFlow = collection.aggregate<Document>(
88
            listOf(
89
                searchStage,
90
                projectStage,
91
                addFieldsStage,
92
                limitStage,
93
                unionWithStage,
94
                facetStage
95
            )
96
        )
97
        resultsFlow.collect { println(it) }
98
99
    }
100
    mongoClient.close()
101
}

1
const MongoClient = require("mongodb").MongoClient;
2
const assert = require("assert");
3
4
const agg = [
5
    {
6
      '$search': {
7
        'text': { 'query': 'Mobile', 'path': 'name'  }
8
      }
9
    }, {
10
      '$project': {
11
        'score': { '$meta': 'searchScore' }, 
12
        '_id': 0, 'number_of_employees': 1, 'founded_year': 1, 'name': 1
13
      }
14
    }, {
15
      '$set': { 'source': 'companies' }
16
    }, {
17
      '$limit': 3
18
    }, {
19
      '$unionWith': {
20
        'coll': 'inspections', 
21
        'pipeline': [
22
          {
23
            '$search': {
24
              'text': { 'query': 'Mobile', 'path': 'business_name' }
25
            }
26
          }, {
27
            '$set': { 'source': 'inspections' }
28
          }, {
29
            '$project': {
30
              'score': { '$meta': 'searchScore' }, 
31
              'source': 1, '_id': 0, 'business_name': 1, 'address': 1
32
            }
33
          }, {
34
            '$limit': 3
35
          }, {
36
            '$sort': { 'score': -1 }
37
          }
38
        ]
39
      }
40
    }
41
  ];
42
43
MongoClient.connect(
44
  "<connection-string>",
45
  { useNewUrlParser: true, useUnifiedTopology: true },
46
  async function (connectErr, client) {
47
    assert.equal(null, connectErr);
48
    const coll = client.db("sample_training").collection("companies");
49
    let cursor = await coll.aggregate(agg);
50
    await cursor.forEach((doc) => console.log(doc));
51
    client.close();
52
  }
53
);