fix(mongodb): update filter composition to handle MongoDB 5.x driver (#36004)

* fix(mongodb): update filter composition to handle MongoDB 5.x driver changes

Modify filter composition logic to avoid $and wrapping when possible by merging conditions directly into a composite Document. This addresses behavior changes in MongoDB driver 5.x while maintaining backward compatibility.

* fix the filter

* disable spotbugs for sql

Author: liferoad

.github/trigger_files/beam_PostCommit_SQL.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run ",
-  "modification": 3
+  "modification": 4
 }

sdks/java/extensions/sql/build.gradle

@@ -41,6 +41,8 @@ applyJavaNature(
   ],
   // javacc generated code produces lint warnings
   disableLintWarnings: ['dep-ann', 'rawtypes'],
+  // Disable SpotBugs due to ASM bytecode analysis issue with BeamCalcRel class
+  enableSpotbugs: false,
 )
 
 description = "Apache Beam :: SDKs :: Java :: Extensions :: SQL"

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/mongodb/MongoDbTable.java

@@ -21,6 +21,8 @@
 import static org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlKind.COMPARISON;
 import static org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.SqlKind.OR;
 
+import com.mongodb.BasicDBObject;
+import com.mongodb.MongoClientSettings;
 import com.mongodb.client.model.Filters;
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -64,6 +66,7 @@
 import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.bson.BsonDocument;
 import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.bson.json.JsonMode;
@@ -178,7 +181,21 @@ private Bson constructPredicate(List<RexNode> supported) {
     if (cnf.size() == 1) {
       return cnf.get(0);
     }
-    return Filters.and(cnf);
+    // Convert all filters to BsonDocument and merge them into a single Document
+    // This avoids wrapping in $and which changed behavior in MongoDB driver 5.x
+    Document compositeFilter = new Document();
+    for (Bson filter : cnf) {
+      // Convert any Bson filter to BsonDocument first
+      BsonDocument bsonDoc =
+          filter.toBsonDocument(BasicDBObject.class, MongoClientSettings.getDefaultCodecRegistry());
+      // Convert BsonDocument to Document for easier manipulation
+      Document doc = Document.parse(bsonDoc.toJson());
+      // Merge all top-level conditions into the composite filter
+      for (String key : doc.keySet()) {
+        compositeFilter.append(key, doc.get(key));
+      }
+    }
+    return compositeFilter;
   }
 
   /**