如何在MongoDB Scala驱动的聚合投影中使用$filter操作符

Hey there! I totally get wanting to filter embedded document arrays during the aggregation projection stage—let's break this down with clear, actionable examples using the MongoDB Scala driver.

First, let's set up a sample scenario to make this concrete. Suppose we have a users collection where each document looks like this:

{
  "_id": ObjectId("60d21b4667d0d8992e610c85"),
  "name": "Alice",
  "orders": [
    {"orderId": 1, "status": "completed", "amount": 50},
    {"orderId": 2, "status": "pending", "amount": 30},
    {"orderId": 3, "status": "completed", "amount": 75}
  ]
}

Step 1: Import Required Packages #

First, make sure you've imported the necessary classes from the Scala driver:

import org.mongodb.scala._
import org.mongodb.scala.model.Aggregates._
import org.mongodb.scala.model.Projections._
import org.mongodb.scala.model.expressions._
import org.mongodb.scala.model.Filters._

Step 2: Basic $filter Implementation #

To filter the orders array to only include completed orders, we'll use the filter method from the expressions package to construct the $filter expression, then include it in our projection stage.

Here's the full code:

// Get your collection instance (adjust based on your setup)
val mongoClient = MongoClient("mongodb://localhost:27017")
val database = mongoClient.getDatabase("your_db")
val collection = database.getCollection("users")

// Construct the $filter expression
val completedOrdersFilter = filter(
  Expression("orders"),       // The array field we want to filter
  "order",                    // Variable name to reference each array element
  eq(Expression("order.status"), "completed")  // Filter condition
)

// Build the aggregation pipeline
val pipeline = Seq(
  project(
    fields(
      include("name"),  // Keep the name field in the output
      computed("completedOrders", completedOrdersFilter)  // Create a new field with filtered results
    )
  )
)

// Execute the aggregation and handle results
collection.aggregate(pipeline).subscribe(
  result => println(s"Filtered user document: $result"),
  error => println(s"Error during aggregation: ${error.getMessage}"),
  () => println("Aggregation finished successfully")
)

This will return documents where the completedOrders field only contains orders with a status of "completed". If you want to replace the original orders field instead of creating a new one, just change computed("completedOrders", ...) to computed("orders", ...).

Step 3: Complex Filter Conditions #

You can also combine multiple conditions using logical operators like and or or. For example, let's filter for completed orders with an amount greater than 50:

val highValueCompletedFilter = filter(
  Expression("orders"),
  "order",
  and(
    eq(Expression("order.status"), "completed"),
    gt(Expression("order.amount"), 50)
  )
)

val pipeline = Seq(
  project(
    fields(
      include("name"),
      computed("highValueCompletedOrders", highValueCompletedFilter)
    )
  )
)

Key Things to Remember #

• Variable Names: The second parameter in the filter method is the variable name used to reference each array element—pick something descriptive (like "order" here) and avoid conflicting with existing field names.
• Empty Arrays: If no elements match the filter condition, the resulting array will be empty (not null).
• Imports: Don't forget to import org.mongodb.scala.model.expressions._—that's where the filter method lives.

内容的提问来源于stack exchange，提问作者thiaguten