feat: support IN/NOT_IN/NOT_EQUAL operators (#688)

* feat: add support for IN/NOT_IN/NOT_EQUAL operator

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: Vincent-Weng

google-cloud-datastore/src/main/java/com/google/cloud/datastore/StructuredQuery.java

@@ -272,7 +272,10 @@ public Operator apply(String constant) {
       static final Operator GREATER_THAN = type.createAndRegister("GREATER_THAN");
       static final Operator GREATER_THAN_OR_EQUAL = type.createAndRegister("GREATER_THAN_OR_EQUAL");
       static final Operator EQUAL = type.createAndRegister("EQUAL");
+      static final Operator IN = type.createAndRegister("IN");
+      static final Operator NOT_EQUAL = type.createAndRegister("NOT_EQUAL");
       static final Operator HAS_ANCESTOR = type.createAndRegister("HAS_ANCESTOR");
+      static final Operator NOT_IN = type.createAndRegister("NOT_IN");
 
       com.google.datastore.v1.PropertyFilter.Operator toPb() {
         return com.google.datastore.v1.PropertyFilter.Operator.valueOf(name());
@@ -502,6 +505,46 @@ public static PropertyFilter eq(String property, Blob value) {
       return new PropertyFilter(property, Operator.EQUAL, of(value));
     }
 
+    public static PropertyFilter neq(String property, Value<?> value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, value);
+    }
+
+    public static PropertyFilter neq(String property, String value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, of(value));
+    }
+
+    public static PropertyFilter neq(String property, long value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, of(value));
+    }
+
+    public static PropertyFilter neq(String property, double value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, of(value));
+    }
+
+    public static PropertyFilter neq(String property, boolean value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, of(value));
+    }
+
+    public static PropertyFilter neq(String property, Timestamp value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, of(value));
+    }
+
+    public static PropertyFilter neq(String property, Key value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, of(value));
+    }
+
+    public static PropertyFilter neq(String property, Blob value) {
+      return new PropertyFilter(property, Operator.NOT_EQUAL, of(value));
+    }
+
+    public static PropertyFilter in(String property, ListValue value) {
+      return new PropertyFilter(property, Operator.IN, value);
+    }
+
+    public static PropertyFilter not_in(String property, ListValue value) {
+      return new PropertyFilter(property, Operator.NOT_IN, value);
+    }
+
     public static PropertyFilter hasAncestor(Key key) {
       return new PropertyFilter(KEY_PROPERTY_NAME, Operator.HAS_ANCESTOR, of(key));
     }

google-cloud-datastore/src/test/java/com/google/cloud/datastore/it/ITDatastoreTest.java

@@ -583,6 +583,83 @@ public void testRunStructuredQuery() throws InterruptedException {
     assertFalse(results4.hasNext());
   }
 
+  @Test
+  public void testInNotInNeqFilters() throws InterruptedException {
+    Entity e1 =
+        Entity.newBuilder(ENTITY1)
+            .setKey(Key.newBuilder(INCOMPLETE_KEY1, "e1").build())
+            .set("v_int", 10)
+            .build();
+    Entity e2 =
+        Entity.newBuilder(ENTITY1)
+            .setKey(Key.newBuilder(INCOMPLETE_KEY1, "e2").build())
+            .set("v_int", 20)
+            .build();
+    DATASTORE.put(e1, e2);
+
+    Query<Entity> queryIn =
+        Query.newEntityQueryBuilder()
+            .setKind(KIND1)
+            .setFilter(PropertyFilter.in("v_int", ListValue.of(10, 20)))
+            .build();
+
+    Query<Entity> scQueryIn =
+        Query.newEntityQueryBuilder()
+            .setKind(KIND1)
+            .setFilter(PropertyFilter.hasAncestor(ROOT_KEY))
+            .setFilter(PropertyFilter.in("v_int", ListValue.of(10, 20)))
+            .build();
+
+    Iterator<Entity> resultIn = getStronglyConsistentResults(scQueryIn, queryIn);
+
+    assertTrue(resultIn.hasNext());
+    assertEquals(e1, resultIn.next());
+    assertTrue(resultIn.hasNext());
+    assertEquals(e2, resultIn.next());
+    assertFalse(resultIn.hasNext());
+
+    Query<Entity> queryNotIn =
+        Query.newEntityQueryBuilder()
+            .setKind(KIND1)
+            .setFilter(PropertyFilter.not_in("v_int", ListValue.of(20, 30)))
+            .build();
+
+    Query<Entity> scQueryNotIn =
+        Query.newEntityQueryBuilder()
+            .setKind(KIND1)
+            .setFilter(PropertyFilter.hasAncestor(ROOT_KEY))
+            .setFilter(PropertyFilter.not_in("v_int", ListValue.of(20, 30)))
+            .build();
+
+    Iterator<Entity> resultNotIn = getStronglyConsistentResults(scQueryNotIn, queryNotIn);
+
+    assertTrue(resultNotIn.hasNext());
+    assertEquals(e1, resultNotIn.next());
+    assertFalse(resultNotIn.hasNext());
+
+    Query<Entity> queryNeq =
+        Query.newEntityQueryBuilder()
+            .setKind(KIND1)
+            .setFilter(PropertyFilter.neq("v_int", 10))
+            .build();
+
+    Query<Entity> scQueryNeq =
+        Query.newEntityQueryBuilder()
+            .setKind(KIND1)
+            .setFilter(PropertyFilter.hasAncestor(ROOT_KEY))
+            .setFilter(PropertyFilter.neq("v_int", 10))
+            .build();
+
+    Iterator<Entity> resultNeq = getStronglyConsistentResults(scQueryNeq, queryNeq);
+
+    assertTrue(resultNeq.hasNext());
+    assertEquals(e2, resultNeq.next());
+    assertFalse(resultNeq.hasNext());
+
+    DATASTORE.delete(e1.getKey());
+    DATASTORE.delete(e2.getKey());
+  }
+
   @Test
   public void testAllocateId() {
     KeyFactory keyFactory = DATASTORE.newKeyFactory().setKind(KIND1);