How to Run $lookup with an Atlas Search $search Query

1
using MongoDB.Bson;
2
using MongoDB.Bson.Serialization.Attributes;
3
using MongoDB.Bson.Serialization.Conventions;
4
using MongoDB.Driver;
5
using MongoDB.Driver.Core;
6
using MongoDB.Driver.Search;
7
8
public class LookupWithSearch{
9
10
  static void Main(string[] args) {// allow automapping of the camelCase database fields to our MovieDocument
11
    var camelCaseConvention = new ConventionPack { new CamelCaseElementNameConvention() };
12
    ConventionRegistry.Register("CamelCase", camelCaseConvention, type => true);
13
14
    // connect to your Atlas cluster
15
    var mongoClient = new MongoClient("<connection-string>");
16
    
17
    // define namespace
18
    var analyticsDatabase = mongoClient.GetDatabase("sample_analytics");
19
    var accountsCollection = analyticsDatabase.GetCollection<AccountDocument>("accounts");
20
    var customersCollection = analyticsDatabase.GetCollection<CustomerDocument>("customers");
21
22
    // define pipeline stages
23
    var lookupStage = new BsonDocument("$lookup", new BsonDocument{
24
      { "from", "accounts" }, { "localField", "accounts" }, { "foreignField", "account_id" }, 
25
      { "as", "purchases" }, { "pipeline", new BsonArray{
26
        new BsonDocument("$search", new BsonDocument{
27
          { "index", "lookup-with-search-tutorial" }, { "compound", new BsonDocument{
28
            { "must", new BsonArray{
29
              new BsonDocument("queryString", new BsonDocument{
30
                { "defaultPath", "products" }, { "query", "products: (CurrencyService AND InvestmentStock)" }
31
              })
32
            }}, 
33
            { "should", new BsonArray{
34
              new BsonDocument("range", new BsonDocument{
35
                { "path", "limit" }, { "gte", 5000 }, { "lte", 10000 }
36
              })
37
            }}
38
          }}
39
        })
40
      }}
41
    });
42
    var projectStage1 = new BsonDocument("$project", new BsonDocument("_id", 0));
43
    var limitStage = new BsonDocument("$limit", 5);
44
    var projectStage2 = new BsonDocument("$project", new BsonDocument{
45
      { "_id", 0 }, { "address", 0 }, { "birthdate", 0 }, { "username", 0 }, { "tier_and_details", 0 }
46
    });
47
    var aggregationPipeline = new List<BsonDocument> {lookupStage, projectStage1, limitStage, projectStage2};
48
49
    // run pipeline
50
    var results = customersCollection.Aggregate<BsonDocument>(aggregationPipeline).ToList();
51
52
    // print results
53
    foreach (var acct in results) {
54
        Console.WriteLine(acct.ToJson());
55
    }
56
  }            
57
}
58
59
// define fields in the accounts collection
60
[BsonIgnoreExtraElements]
61
public class AccountDocument {
62
    [BsonId]
63
    [BsonRepresentation(BsonType.ObjectId)]
64
    [BsonElement("_id")]
65
    public string Id { get; set; }
66
67
    [BsonElement("account_id")]
68
    public int AccountId { get; set; }
69
70
    [BsonElement("limit")]
71
    public int Limit { get; set; }
72
}
73
74
// define fields in the customers collection
75
[BsonIgnoreExtraElements]
76
public class CustomerDocument {
77
    [BsonId]
78
    [BsonRepresentation(BsonType.ObjectId)]
79
    [BsonElement("_id")]
80
    public ObjectId Id { get; set; }
81
82
    [BsonElement("name")]
83
    public string Name { get; set; }
84
85
    [BsonElement("email")]
86
    public string Email { get; set; }
87
88
    [BsonElement("active")]
89
    public bool Active { get; set; }
90
91
    [BsonElement("accounts")]
92
    public List<int> Accounts { get; set; }
93
}
94
95
// define new array field for matching documents
96
public class CustomerLookedUp: CustomerDocument{
97
    public List<CustomerDocument> Purchases { get; set; }
98
}

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_analytics").Collection("customers")
23
  // define pipeline
24
  lookupStage := bson.D{{"$lookup", bson.D{
25
	{"from", "accounts"},
26
	{"localField", "accounts"},
27
	{"foreignField", "account_id"},
28
	{"as", "purchases"},
29
	{"pipeline", bson.A{
30
	  bson.D{
31
		{"$search", bson.D{
32
		  {"index", "lookup-with-search-tutorial"},
33
		  {"compound", bson.D{
34
			{"must", bson.A{
35
			  bson.D{{"queryString", bson.D{
36
				{"defaultPath", "products"},
37
				{"query", "products: (CurrencyService AND InvestmentStock)"},
38
			  }}},
39
			}},
40
			{"should", bson.A{
41
			  bson.D{{"range", bson.D{
42
				{"path", "limit"},
43
				{"gte", 5000},
44
				{"lte", 10000},
45
			  }}},
46
			}},
47
		  }},
48
		}},
49
	  },
50
	  bson.D{{"$project", bson.D{
51
		{"_id", 0},
52
		{"address", 0},
53
        {"birthdate", 0},
54
	    {"username", 0},
55
	    {"tier_and_details", 0},
56
	  }}},
57
	}},
58
  }}}
59
  limitStage := bson.D{{"$limit", 5}}
60
  projectStage := bson.D{{"$project", bson.D{
61
    {"name", 1},
62
	{"email", 1},
63
	{"active", 1},
64
	{"accounts", 1},
65
	{"purchases", 1},
66
  }}}
67
  // specify the amount of time the operation can run on the server
68
  opts := options.Aggregate().SetMaxTime(5 * time.Second)
69
  // run pipeline
70
  cursor, err := collection.Aggregate(ctx, mongo.Pipeline{lookupStage, limitStage, projectStage}, opts)
71
  if err != nil {
72
	  panic(err)
73
  }
74
  // print results
75
  var results []bson.D
76
  if err = cursor.All(context.TODO(), &results); err != nil {
77
	  panic(err)
78
  }
79
  for _, result := range results {
80
	  fmt.Println(result)
81
  }
82
}

1
import static com.mongodb.client.model.Aggregates.limit;
2
import static com.mongodb.client.model.Aggregates.project;
3
import static com.mongodb.client.model.Projections.*;
4
import java.util.Arrays;
5
import com.mongodb.client.MongoClient;
6
import com.mongodb.client.MongoClients;
7
import com.mongodb.client.MongoCollection;
8
import com.mongodb.client.MongoDatabase;
9
import org.bson.Document;
10
11
public class LookupWithSearchQuery {
12
13
  public static void main(String[] args) {
14
    // connect to your Atlas cluster
15
    String uri = "<connection-string>";
16
        
17
    try (MongoClient mongoClient = MongoClients.create(uri)) {
18
      // set namespace
19
      MongoDatabase database = mongoClient.getDatabase("sample_analytics");
20
      MongoCollection<Document> collection = database.getCollection("customers");
21
            
22
      // define pipeline
23
      Document agg = new Document("$lookup", 
24
        new Document("from", "accounts")
25
          .append("localField", "accounts")
26
          .append("foreignField", "account_id")
27
          .append("as", "purchases")
28
          .append("pipeline", Arrays.asList(new Document("$search", 
29
            new Document("index", "lookup-with-search-tutorial")
30
              .append("compound", 
31
              new Document("must", Arrays.asList(new Document("queryString", 
32
                new Document("defaultPath", "products")
33
                  .append("query", "products: (CurrencyService AND InvestmentStock)"))))
34
                  .append("should", Arrays.asList(new Document("range", 
35
                    new Document("path", "limit")
36
                      .append("gte", 5000L)
37
                      .append("lte", 10000L)
38
                  )))
39
              )
40
          ), 
41
          new Document("$limit", 5L),
42
          new Document("$project", 
43
          new Document("_id", 0L)
44
            .append("address", 0L)
45
            .append("birthdate", 0L)
46
            .append("username", 0L)
47
            .append("tier_and_details", 0L)
48
        )))
49
      );
50
      // run pipeline and print results
51
      collection.aggregate(Arrays.asList(agg,
52
        limit(5),
53
        project(fields(excludeId(), include("name", "email", "active", "accounts", "purchases")))
54
      ))
55
      .forEach(doc -> System.out.println(doc.toJson()));
56
    }
57
  }
58
}

1
import com.mongodb.client.model.Aggregates.limit
2
import com.mongodb.client.model.Aggregates.project
3
import com.mongodb.client.model.Projections.*
4
import com.mongodb.kotlin.client.coroutine.MongoClient
5
import kotlinx.coroutines.runBlocking
6
import org.bson.Document
7
import java.util.*
8
9
fun main() {
10
    // connect to your Atlas cluster
11
    val uri = "<connection-string>"
12
    val mongoClient = MongoClient.create(uri)
13
14
    // set namespace
15
    val database = mongoClient.getDatabase("sample_analytics")
16
    val collection = database.getCollection<Document>("customers")
17
18
    runBlocking {
19
        // define pipeline
20
        val agg = Document(
21
            "\$lookup",
22
            Document("from", "accounts")
23
                .append("localField", "accounts")
24
                .append("foreignField", "account_id")
25
                .append("as", "purchases")
26
                .append(
27
                    "pipeline", Arrays.asList(
28
                        Document(
29
                            "\$search",
30
                            Document("index", "lookup-with-search-tutorial")
31
                                .append(
32
                                    "compound",
33
                                    Document(
34
                                        "must", Arrays.asList(
35
                                            Document(
36
                                                "queryString",
37
                                                Document("defaultPath", "products")
38
                                                    .append("query", "products: (CurrencyService AND InvestmentStock)")
39
                                            )
40
                                        )
41
                                    )
42
                                        .append(
43
                                            "should", Arrays.asList(
44
                                                Document(
45
                                                    "range",
46
                                                    Document("path", "limit")
47
                                                        .append("gte", 5000)
48
                                                        .append("lte", 10000)
49
                                                )
50
                                            )
51
                                        )
52
                                )
53
                        ),
54
                        Document("\$limit", 5),
55
                        Document(
56
                            "\$project",
57
                            Document("_id", 0)
58
                                .append("address", 0)
59
                                .append("birthdate", 0)
60
                                .append("username", 0)
61
                                .append("tier_and_details", 0)
62
                        )
63
                    )
64
                )
65
        )
66
67
        // run pipeline and print results
68
        val resultsFlow = collection.aggregate<Document>(
69
            listOf(
70
                agg,
71
                limit(5),
72
                project(fields(excludeId(), include("name", "email", "active", "accounts", "purchases")))
73
            )
74
        )
75
        resultsFlow.collect { println(it) }
76
    }
77
    mongoClient.close()
78
}

1
const MongoClient = require("mongodb").MongoClient;
2
const assert = require("assert");
3
4
const agg = [
5
    {
6
      '$lookup': {
7
        'from': 'accounts', 
8
        'localField': 'accounts', 
9
        'foreignField': 'account_id', 
10
        'as': 'purchases', 
11
        'pipeline': [
12
          {
13
            '$search': {
14
              'index': 'lookup-with-search-tutorial', 
15
              'compound': {
16
                'must': [
17
                  {
18
                    'queryString': {
19
                      'defaultPath': 'products', 
20
                      'query': 'products: (CurrencyService AND InvestmentStock)'
21
                    }
22
                  }
23
                ], 
24
                'should': [
25
                  {
26
                    'range': {
27
                      'path': 'limit', 
28
                      'gte': 5000, 
29
                      'lte': 10000
30
                    }
31
                  }
32
                ]
33
              }
34
            }
35
          }, {
36
            '$project': {
37
              '_id': 0
38
            }
39
          }
40
        ]
41
      }
42
    }, {
43
      '$limit': 5
44
    }, {
45
      '$project': {
46
        '_id': 0, 
47
        'address': 0, 
48
        'birthdate': 0, 
49
        'username': 0, 
50
        'tier_and_details': 0
51
      }
52
    }
53
  ];
54
55
MongoClient.connect(
56
  "<connection-string>",
57
  { useNewUrlParser: true, useUnifiedTopology: true },
58
  async function (connectErr, client) {
59
    assert.equal(null, connectErr);
60
    const coll = client.db("sample_analytics").collection("customers");
61
    let cursor = await coll.aggregate(agg);
62
    await cursor.forEach((doc) => console.log(doc));
63
    client.close();
64
  }
65
);

1
import datetime
2
import pymongo
3
4
# connect to your Atlas cluster
5
client = pymongo.MongoClient('<connection-string>')
6
7
# define pipeline
8
pipeline = [
9
  {
10
    '$lookup': {
11
      'from': 'accounts', 
12
      'localField': 'accounts', 
13
      'foreignField': 'account_id', 
14
      'as': 'purchases', 
15
      'pipeline': [
16
        {
17
          '$search': {
18
            'index': 'lookup-with-search-tutorial', 
19
            'compound': {
20
              'must': [
21
                {
22
                  'queryString': {
23
                    'defaultPath': 'products', 
24
                    'query': 'products: (CurrencyService AND InvestmentStock)'
25
                  }
26
                }
27
              ], 
28
              'should': [
29
                {
30
                  'range': {
31
                    'path': 'limit', 
32
                    'gte': 5000, 
33
                    'lte': 10000
34
                  }
35
                }
36
              ]
37
            }
38
          }
39
        }, 
40
        { '$project': { '_id': 0 }  }
41
      ]
42
    }
43
  }, 
44
  { '$limit': 5 }, 
45
  {
46
    '$project': {
47
      '_id': 0, 
48
      'address': 0, 
49
      'birthdate': 0, 
50
      'username': 0, 
51
      'tier_and_details': 0
52
    }
53
  }
54
]
55
56
# run pipeline
57
result = client['sample_analytics']['customers'].aggregate(pipeline)
58
59
# print results
60
for i in result:
61
    print(i)