Add some missing geospatial scalar functions to support use in intermediate stages with the v2 query engine

pinot-core/src/main/java/org/apache/pinot/core/geospatial/transform/function/ScalarFunctions.java

@@ -188,4 +188,40 @@ public static double stDistance(byte[] firstPoint, byte[] secondPoint) {
       return firstGeometry.isEmpty() || secondGeometry.isEmpty() ? Double.NaN : firstGeometry.distance(secondGeometry);
     }
   }
+
+  @ScalarFunction(names = {"stContains", "st_contains"})
+  public static int stContains(byte[] first, byte[] second) {
+    Geometry firstGeometry = GeometrySerializer.deserialize(first);
+    Geometry secondGeometry = GeometrySerializer.deserialize(second);
+    if (GeometryUtils.isGeography(firstGeometry) != GeometryUtils.isGeography(secondGeometry)) {
+      throw new RuntimeException("The first and second arguments should either both be geometry or both be geography");
+    }
+    // TODO: to fully support Geography contains operation.
+    return firstGeometry.contains(secondGeometry) ? 1 : 0;
+  }
+
+  @ScalarFunction(names = {"stEquals", "st_equals"})
+  public static int stEquals(byte[] first, byte[] second) {
+    Geometry firstGeometry = GeometrySerializer.deserialize(first);
+    Geometry secondGeometry = GeometrySerializer.deserialize(second);
+
+    return firstGeometry.equals(secondGeometry) ? 1 : 0;
+  }
+
+  @ScalarFunction(names = {"stGeometryType", "st_geometry_type"})
+  public static String stGeometryType(byte[] bytes) {
+    Geometry geometry = GeometrySerializer.deserialize(bytes);
+    return geometry.getGeometryType();
+  }
+
+  @ScalarFunction(names = {"stWithin", "st_within"})
+  public static int stWithin(byte[] first, byte[] second) {
+    Geometry firstGeometry = GeometrySerializer.deserialize(first);
+    Geometry secondGeometry = GeometrySerializer.deserialize(second);
+    if (GeometryUtils.isGeography(firstGeometry) != GeometryUtils.isGeography(secondGeometry)) {
+      throw new RuntimeException("The first and second arguments should either both be geometry or both be geography");
+    }
+    // TODO: to fully support Geography within operation.
+    return firstGeometry.within(secondGeometry) ? 1 : 0;
+  }
 }

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/GeoSpatialTest.java

@@ -404,6 +404,39 @@ public void testStWithinQuery(boolean useMultiStageQueryEngine)
     }
   }
 
+  @Test(dataProvider = "useBothQueryEngines")
+  public void testStContainsQuery(boolean useMultiStageQueryEngine) throws Exception {
+    setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+    // contains is the converse of within
+    String query =
+        String.format("Select ST_Contains(ST_GeomFromText(%s), ST_GeomFromText(%s)), %s from %s",
+            WKT_2_NAME, WKT_1_NAME, ST_WITHIN_RESULT_NAME, getTableName());
+    JsonNode pinotResponse = postQuery(query);
+    JsonNode rows = pinotResponse.get("resultTable").get("rows");
+    for (int i = 0; i < rows.size(); i++) {
+      JsonNode row = rows.get(i);
+      boolean actualResult = row.get(0).intValue() == 1;
+      boolean expectedResult = row.get(1).booleanValue();
+      Assert.assertEquals(actualResult, expectedResult);
+    }
+  }
+
+  @Test(dataProvider = "useV2QueryEngine")
+  public void testStContainsQueryWithV2(boolean useMultiStageQueryEngine) throws Exception {
+    setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+
+    String query = String.format("Select ST_Contains(ST_GeomFromText('MULTIPOINT (20 20, 25 25)'), "
+        + "ST_GeomFromText('POINT (25 25)')) from %s a CROSS JOIN %s b LIMIT 5", getTableName(), getTableName());
+
+    JsonNode pinotResponse = postQuery(query);
+    JsonNode rows = pinotResponse.get("resultTable").get("rows");
+    for (int i = 0; i < rows.size(); i++) {
+      JsonNode row = rows.get(i);
+      boolean actualResult = row.get(0).intValue() == 1;
+      Assert.assertTrue(actualResult);
+    }
+  }
+
   @Test(dataProvider = "useV2QueryEngine")
   public void testStWithinLiteral(boolean useMultiStageQueryEngine)
       throws Exception {