[GH-3156][GH-3157][GH-3158] Implement distributed GeoSeries methods

common/src/main/java/org/apache/sedona/common/Predicates.java

@@ -158,6 +158,11 @@ public static boolean equals(Geometry leftGeometry, Geometry rightGeometry) {
     return leftGeometry.equalsTopo(rightGeometry);
   }
 
+  public static boolean equalsExact(
+      Geometry leftGeometry, Geometry rightGeometry, double tolerance) {
+    return leftGeometry.equalsExact(rightGeometry, tolerance);
+  }
+
   public static boolean disjoint(Geometry leftGeometry, Geometry rightGeometry) {
     return leftGeometry.disjoint(rightGeometry);
   }

common/src/test/java/org/apache/sedona/common/PredicatesTest.java

@@ -338,6 +338,42 @@ public void testEqualsEmptyGeometries() throws ParseException {
     assertFalse(Predicates.equals(pointEmpty, point));
   }
 
+  @Test
+  public void testEqualsExact() throws ParseException {
+    Geometry origin = geomFromEWKT("POINT(0 0)");
+    Geometry nearby = geomFromEWKT("POINT(0.03 0.04)");
+    assertTrue(Predicates.equalsExact(origin, origin, 0.0));
+    assertTrue(Predicates.equalsExact(origin, nearby, 0.05));
+    assertTrue(Predicates.equalsExact(origin, nearby, 0.051));
+    assertFalse(Predicates.equalsExact(origin, nearby, 0.049));
+    assertFalse(Predicates.equalsExact(origin, origin, -1.0));
+    assertFalse(Predicates.equalsExact(origin, origin, Double.NaN));
+    assertTrue(Predicates.equalsExact(origin, nearby, Double.POSITIVE_INFINITY));
+
+    Geometry line = geomFromEWKT("LINESTRING(0 0, 1 1, 2 0)");
+    Geometry reversed = geomFromEWKT("LINESTRING(2 0, 1 1, 0 0)");
+    assertFalse(Predicates.equalsExact(line, reversed, 0.0));
+
+    Geometry pointZ1 = geomFromEWKT("POINT Z (1 2 3)");
+    Geometry pointZ2 = geomFromEWKT("POINT Z (1 2 99)");
+    assertTrue(Predicates.equalsExact(pointZ1, pointZ2, 0.0));
+
+    Geometry pointM1 = geomFromEWKT("POINT M (1 2 3)");
+    Geometry pointM2 = geomFromEWKT("POINT M (1 2 99)");
+    assertTrue(Predicates.equalsExact(pointM1, pointM2, 0.0));
+  }
+
+  @Test
+  public void testEqualsExactEmptyAndCollectionStructure() throws ParseException {
+    Geometry pointEmpty = geomFromEWKT("POINT EMPTY");
+    assertTrue(Predicates.equalsExact(pointEmpty, geomFromEWKT("POINT EMPTY"), 0.0));
+    assertFalse(Predicates.equalsExact(pointEmpty, geomFromEWKT("POLYGON EMPTY"), 0.0));
+
+    Geometry collection = geomFromEWKT("GEOMETRYCOLLECTION(POINT(0 0), LINESTRING(0 0, 1 1))");
+    Geometry reordered = geomFromEWKT("GEOMETRYCOLLECTION(LINESTRING(0 0, 1 1), POINT(0 0))");
+    assertFalse(Predicates.equalsExact(collection, reordered, 0.0));
+  }
+
   @Test
   public void testRelateBoolean() throws ParseException {
     Geometry geom1 = geomFromEWKT("POINT(1 2)");

docs/api/flink/Geometry-Functions.md

@@ -166,6 +166,7 @@ These functions test spatial relationships between geometries, returning boolean
 | [ST_Disjoint](Predicates/ST_Disjoint.md) | Boolean | Return true if A and B are disjoint | v1.2.1 |
 | [ST_DWithin](Predicates/ST_DWithin.md) | Boolean | Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. | v1.5.1 |
 | [ST_Equals](Predicates/ST_Equals.md) | Boolean | Return true if A equals to B | v1.5.0 |
+| [ST_EqualsExact](Predicates/ST_EqualsExact.md) | Boolean | Return true if A and B have matching structures and corresponding coordinates within a tolerance | v1.9.1 |
 | [ST_Intersects](Predicates/ST_Intersects.md) | Boolean | Return true if A intersects B | v1.2.0 |
 | [ST_OrderingEquals](Predicates/ST_OrderingEquals.md) | Boolean | Returns true if the geometries are equal and the coordinates are in the same order | v1.2.1 |
 | [ST_Overlaps](Predicates/ST_Overlaps.md) | Boolean | Return true if A overlaps B | v1.5.0 |

docs/api/flink/Predicates/ST_EqualsExact.md

@@ -0,0 +1,62 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements.  See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership.  The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+
+# ST_EqualsExact
+
+Introduction: Return true if A and B have the same structure and their corresponding coordinates are equal within a tolerance.
+
+Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores z and m coordinates.
+
+Format: `ST_EqualsExact (A: Geometry, B: Geometry, tolerance: Double)`
+
+Return type: `Boolean`
+
+Since: `v1.9.1`
+
+Example:
+
+```sql
+SELECT ST_EqualsExact(
+    ST_GeomFromWKT('POINT (0 0)'),
+    ST_GeomFromWKT('POINT (0.03 0.04)'),
+    0.05
+)
+```
+
+Output:
+
+```
+true
+```
+
+The order of coordinates must match:
+
+```sql
+SELECT ST_EqualsExact(
+    ST_GeomFromWKT('LINESTRING (0 0, 1 1)'),
+    ST_GeomFromWKT('LINESTRING (1 1, 0 0)'),
+    0.0
+)
+```
+
+Output:
+
+```
+false
+```

docs/api/snowflake/vector-data/Geometry-Functions.md

@@ -158,6 +158,7 @@ These functions test spatial relationships between geometries, returning boolean
 | [ST_Disjoint](Predicates/ST_Disjoint.md) | Return true if A and B are disjoint |
 | [ST_DWithin](Predicates/ST_DWithin.md) | Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. This function essentially checks if the shortest distance between the envelope of the two geometries is <= the ... |
 | [ST_Equals](Predicates/ST_Equals.md) | Return true if A equals to B |
+| [ST_EqualsExact](Predicates/ST_EqualsExact.md) | Return true if A and B have matching structures and corresponding coordinates within a tolerance |
 | [ST_Intersects](Predicates/ST_Intersects.md) | Return true if A intersects B |
 | [ST_OrderingEquals](Predicates/ST_OrderingEquals.md) | Returns true if the geometries are equal and the coordinates are in the same order |
 | [ST_Overlaps](Predicates/ST_Overlaps.md) | Return true if A overlaps B |

docs/api/snowflake/vector-data/Predicates/ST_EqualsExact.md

@@ -0,0 +1,60 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements.  See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership.  The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+
+# ST_EqualsExact
+
+Introduction: Return true if A and B have the same structure and their corresponding coordinates are equal within a tolerance.
+
+Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores z and m coordinates.
+
+Format: `ST_EqualsExact (A: Geometry, B: Geometry, tolerance: Double)`
+
+Return type: `Boolean`
+
+SQL Example:
+
+```sql
+SELECT ST_EqualsExact(
+    ST_GeomFromWKT('POINT (0 0)'),
+    ST_GeomFromWKT('POINT (0.03 0.04)'),
+    0.05
+)
+```
+
+Output:
+
+```
+true
+```
+
+The order of coordinates must match:
+
+```sql
+SELECT ST_EqualsExact(
+    ST_GeomFromWKT('LINESTRING (0 0, 1 1)'),
+    ST_GeomFromWKT('LINESTRING (1 1, 0 0)'),
+    0.0
+)
+```
+
+Output:
+
+```
+false
+```

docs/api/sql/Geometry-Functions.md

@@ -168,6 +168,7 @@ These functions test spatial relationships between geometries, returning boolean
 | [ST_DWithin](Predicates/ST_DWithin.md) | Boolean | Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. | v1.5.1 |
 | [ST_3DDWithin](Predicates/ST_3DDWithin.md) | Boolean | Returns true if A and B are within a specified 3D Euclidean 'distance'. Accepts Geometry or Box3D inputs. | v1.9.1 |
 | [ST_Equals](Predicates/ST_Equals.md) | Boolean | Return true if A equals to B | v1.0.0 |
+| [ST_EqualsExact](Predicates/ST_EqualsExact.md) | Boolean | Return true if A and B have matching structures and corresponding coordinates within a tolerance | v1.9.1 |
 | [ST_Intersects](Predicates/ST_Intersects.md) | Boolean | Return true if A intersects B | v1.0.0 |
 | [ST_OrderingEquals](Predicates/ST_OrderingEquals.md) | Boolean | Returns true if the geometries are equal and the coordinates are in the same order | v1.2.1 |
 | [ST_Overlaps](Predicates/ST_Overlaps.md) | Boolean | Return true if A overlaps B | v1.0.0 |

docs/api/sql/Predicates/ST_EqualsExact.md

@@ -0,0 +1,62 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements.  See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership.  The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+
+# ST_EqualsExact
+
+Introduction: Returns true if A and B have the same structure and their corresponding coordinates are equal within a tolerance.
+
+Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores Z and M coordinates.
+
+Format: `ST_EqualsExact (A: Geometry, B: Geometry, tolerance: Double)`
+
+Return type: `Boolean`
+
+Since: `v1.9.1`
+
+SQL Example
+
+```sql
+SELECT ST_EqualsExact(
+    ST_GeomFromWKT('POINT (0 0)'),
+    ST_GeomFromWKT('POINT (0.03 0.04)'),
+    0.05
+)
+```
+
+Output:
+
+```
+true
+```
+
+The order of coordinates must match:
+
+```sql
+SELECT ST_EqualsExact(
+    ST_GeomFromWKT('LINESTRING (0 0, 1 1)'),
+    ST_GeomFromWKT('LINESTRING (1 1, 0 0)'),
+    0.0
+)
+```
+
+Output:
+
+```
+false
+```

flink/src/main/java/org/apache/sedona/flink/Catalog.java

@@ -269,6 +269,7 @@ public static UserDefinedFunction[] getPredicates() {
       new Predicates.ST_CoveredBy(),
       new Predicates.ST_Disjoint(),
       new Predicates.ST_Equals(),
+      new Predicates.ST_EqualsExact(),
       new Predicates.ST_OrderingEquals(),
       new Predicates.ST_Overlaps(),
       new Predicates.ST_Touches(),

flink/src/main/java/org/apache/sedona/flink/expressions/Predicates.java

@@ -347,6 +347,30 @@ public Boolean eval(
     }
   }
 
+  public static class ST_EqualsExact extends ScalarFunction {
+
+    public ST_EqualsExact() {}
+
+    @DataTypeHint("Boolean")
+    public Boolean eval(
+        @DataTypeHint(
+                value = "RAW",
+                rawSerializer = GeometryTypeSerializer.class,
+                bridgedTo = Geometry.class)
+            Object o1,
+        @DataTypeHint(
+                value = "RAW",
+                rawSerializer = GeometryTypeSerializer.class,
+                bridgedTo = Geometry.class)
+            Object o2,
+        @DataTypeHint("Double") Double tolerance) {
+      if (o1 == null || o2 == null || tolerance == null) return null;
+      Geometry geom1 = (Geometry) o1;
+      Geometry geom2 = (Geometry) o2;
+      return org.apache.sedona.common.Predicates.equalsExact(geom1, geom2, tolerance);
+    }
+  }
+
   public static class ST_OrderingEquals extends ScalarFunction {
 
     /** Constructor for relation checking without duplicate removal */

flink/src/test/java/org/apache/sedona/flink/PredicateTest.java

@@ -206,6 +206,29 @@ public void testEquals() {
     assertEquals(true, actual);
   }
 
+  @Test
+  public void testEqualsExact() {
+    Table table =
+        tableEnv.sqlQuery(
+            "SELECT"
+                + " ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.05)"
+                + " AS within_tolerance,"
+                + " ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.049)"
+                + " AS outside_tolerance,"
+                + " ST_EqualsExact(ST_GeomFromWKT(CAST(NULL AS STRING)),"
+                + " ST_Point(0.0, 0.0), 0.0) AS null_left,"
+                + " ST_EqualsExact(ST_Point(0.0, 0.0),"
+                + " ST_GeomFromWKT(CAST(NULL AS STRING)), 0.0) AS null_right,"
+                + " ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.0, 0.0),"
+                + " CAST(NULL AS DOUBLE)) AS null_tolerance");
+    org.apache.flink.types.Row row = first(table);
+    assertEquals(true, row.getField(0));
+    assertEquals(false, row.getField(1));
+    assertNull(row.getField(2));
+    assertNull(row.getField(3));
+    assertNull(row.getField(4));
+  }
+
   @Test
   public void testOrderingEquals() {
     Table lineStringTable = createLineStringTable(testDataSize);