fix(geo): Fix correctness issue with spatial joins on GeometryCollections (facebookincubator#15778)

Summary:
Pull Request resolved: facebookincubator#15778

The envelope for the spatial index was being calculated incorrectly, because
the Shapefile format does not add an envelope to the beginning of a GeometryCollection.
This was causing GeometryCollections to miss hits on the Spatial Index, causing rows
to be dropped.

This uses the newly moved GeometrySerde to get the Envelope in the standard way.
However, this requires us to include GEOS for spatial joins.

Reviewed By: Sullivan-Patrick

Differential Revision: D89239585

Author: jagill

velox/exec/SpatialJoinBuild.cpp

@@ -15,8 +15,11 @@
  */
 
 #include "velox/exec/SpatialJoinBuild.h"
-#include "velox/common/base/IOUtils.h"
+#include <limits>
 #include "velox/common/geospatial/GeometryConstants.h"
+#ifdef VELOX_ENABLE_GEO
+#include "velox/common/geospatial/GeometrySerde.h"
+#endif
 #include "velox/exec/Task.h"
 
 namespace facebook::velox::exec {
@@ -127,51 +130,29 @@ std::vector<RowVectorPtr> SpatialJoinBuild::mergeDataVectors() const {
 Envelope SpatialJoinBuild::readEnvelope(
     const StringView& geometryBytes,
     double radius) {
-  VELOX_CHECK_GE(
-      geometryBytes.size(),
-      sizeof(GeometrySerializationType) + 2 * sizeof(double));
-  double minX;
-  double minY;
-  double maxX;
-  double maxY;
-
-  velox::common::InputByteStream inputStream(geometryBytes.data());
-  // Geometry Serde makes it easy to get the envelope.
-  // The first byte is the GeometrySerializationType.
-  // All coordinates are doubles (8 bytes)
-  // Depending on the type, the next bytes are:
-  // 1. POINT: x, y
-  // 2. ENVELOPE: minX, minY, maxX, maxY
-  // 3. Else: EsriShapeType (4 bytes), minX, minY, maxX, maxY, GeometryBytes
-  auto geometryType = inputStream.read<GeometrySerializationType>();
-  if (geometryType == GeometrySerializationType::POINT) {
-    double x = inputStream.read<double>();
-    double y = inputStream.read<double>();
-    minX = x - radius;
-    minY = y - radius;
-    maxX = x + radius;
-    maxY = y + radius;
+#ifdef VELOX_ENABLE_GEO
+  radius = std::max(radius, 0.0);
+  auto geosEnvelope =
+      common::geospatial::GeometryDeserializer::deserializeEnvelope(
+          geometryBytes);
+  if (geosEnvelope->isNull()) {
+    return Envelope::empty();
   } else {
-    if (geometryType != GeometrySerializationType::ENVELOPE) {
-      // Unused esriType
-      inputStream.read<velox::common::geospatial::EsriShapeType>();
-    }
-    minX = inputStream.read<double>() - radius;
-    minY = inputStream.read<double>() - radius;
-    maxX = inputStream.read<double>() + radius;
-    maxY = inputStream.read<double>() + radius;
+    return Envelope::from(
+        geosEnvelope->getMinX() - radius,
+        geosEnvelope->getMinY() - radius,
+        geosEnvelope->getMaxX() + radius,
+        geosEnvelope->getMaxY() + radius);
   }
-
-  Envelope envelope;
-
-  // This also catches NaNs
-  if (minX <= maxX && minY <= maxY) {
-    envelope = Envelope::from(minX, minY, maxX, maxY);
-  } else {
-    envelope = Envelope::empty();
-  }
-
-  return envelope;
+#else
+  // When VELOX_ENABLE_GEO is not set, return an envelope of infinite area
+  // to ensure all geometries are considered for spatial join
+  return Envelope::from(
+      -std::numeric_limits<double>::infinity(),
+      -std::numeric_limits<double>::infinity(),
+      std::numeric_limits<double>::infinity(),
+      std::numeric_limits<double>::infinity());
+#endif
 }
 
 SpatialIndex SpatialJoinBuild::buildSpatialIndex(

velox/exec/tests/SpatialJoinTest.cpp

@@ -383,6 +383,46 @@ TEST_F(SpatialJoinTest, testSimpleNullRowsJoin) {
       {"POINT (1 1)"});
 }
 
+TEST_F(SpatialJoinTest, testGeometryCollection) {
+  runTest(
+      {"GEOMETRYCOLLECTION (POINT (1 1))",
+       "GEOMETRYCOLLECTION EMPTY",
+       "POINT (1 1)"},
+      {"GEOMETRYCOLLECTION (POINT (1 1))",
+       "GEOMETRYCOLLECTION EMPTY",
+       "POINT (1 1)"},
+      std::nullopt,
+      "ST_Intersects(left_g, right_g)",
+      core::JoinType::kInner,
+      {"GEOMETRYCOLLECTION (POINT (1 1))",
+       "GEOMETRYCOLLECTION (POINT (1 1))",
+       "POINT (1 1)",
+       "POINT (1 1)"},
+      {"GEOMETRYCOLLECTION (POINT (1 1))",
+       "POINT (1 1)",
+       "GEOMETRYCOLLECTION (POINT (1 1))",
+       "POINT (1 1)"});
+
+  runTest(
+      {"GEOMETRYCOLLECTION (POINT (1 1))",
+       "GEOMETRYCOLLECTION EMPTY",
+       "POINT (1 1)"},
+      {"GEOMETRYCOLLECTION (POINT (1 2))",
+       "GEOMETRYCOLLECTION EMPTY",
+       "POINT (1 2)"},
+      std::vector<std::optional<double>>{1.0, 1.0, 1.0},
+      "ST_Distance(left_g, right_g) <= radius",
+      core::JoinType::kInner,
+      {"GEOMETRYCOLLECTION (POINT (1 1))",
+       "GEOMETRYCOLLECTION (POINT (1 1))",
+       "POINT (1 1)",
+       "POINT (1 1)"},
+      {"GEOMETRYCOLLECTION (POINT (1 2))",
+       "POINT (1 2)",
+       "GEOMETRYCOLLECTION (POINT (1 2))",
+       "POINT (1 2)"});
+}
+
 TEST_F(SpatialJoinTest, testDistanceJoin) {
   runTest(
       {"POINT (1 2)", "POLYGON ((1 2, 2 2, 2 3, 1 3, 1 2))", std::nullopt},