PARQUET-2417: Add support for geometry logical type

parquet-column/pom.xml

@@ -76,6 +76,18 @@
       <version>${slf4j.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>org.locationtech.jts</groupId>
+      <artifactId>jts-core</artifactId>
+      <version>${jts.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>net.sf.geographiclib</groupId>
+      <artifactId>GeographicLib-Java</artifactId>
+      <version>1.50</version>
+    </dependency>
+
     <dependency>
       <groupId>com.carrotsearch</groupId>
       <artifactId>junit-benchmarks</artifactId>

parquet-column/src/main/java/org/apache/parquet/column/statistics/BinaryStatistics.java

@@ -18,7 +18,9 @@
  */
 package org.apache.parquet.column.statistics;
 
+import org.apache.parquet.column.statistics.geometry.GeometryStatistics;
 import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
 import org.apache.parquet.schema.PrimitiveType;
 import org.apache.parquet.schema.Types;
 
@@ -30,6 +32,7 @@ public class BinaryStatistics extends Statistics<Binary> {
 
   private Binary max;
   private Binary min;
+  private GeometryStatistics geometryStatistics = null;
 
   /**
    * @deprecated will be removed in 2.0.0. Use {@link Statistics#createStats(org.apache.parquet.schema.Type)} instead
@@ -41,6 +44,13 @@ public BinaryStatistics() {
 
   BinaryStatistics(PrimitiveType type) {
     super(type);
+    LogicalTypeAnnotation logicalType = type.getLogicalTypeAnnotation();
+    if (logicalType instanceof LogicalTypeAnnotation.GeometryLogicalTypeAnnotation) {
+      LogicalTypeAnnotation.GeometryLogicalTypeAnnotation geometryLogicalType =
+          (LogicalTypeAnnotation.GeometryLogicalTypeAnnotation) logicalType;
+      geometryStatistics = new GeometryStatistics(
+          geometryLogicalType.getEdges(), geometryLogicalType.getCrs(), geometryLogicalType.getMetadata());
+    }
   }
 
   private BinaryStatistics(BinaryStatistics other) {
@@ -49,6 +59,9 @@ private BinaryStatistics(BinaryStatistics other) {
       initializeStats(other.min, other.max);
     }
     setNumNulls(other.getNumNulls());
+    if (other.geometryStatistics != null) {
+      geometryStatistics = other.geometryStatistics.copy();
+    }
   }
 
   @Override
@@ -62,6 +75,9 @@ public void updateStats(Binary value) {
     } else if (comparator().compare(max, value) < 0) {
       max = value.copy();
     }
+    if (geometryStatistics != null) {
+      geometryStatistics.update(value);
+    }
   }
 
   @Override
@@ -72,6 +88,9 @@ public void mergeStatisticsMinMax(Statistics stats) {
     } else {
       updateStats(binaryStats.getMin(), binaryStats.getMax());
     }
+    if (geometryStatistics != null) {
+      geometryStatistics.merge(binaryStats.geometryStatistics);
+    }
   }
 
   /**
@@ -190,4 +209,12 @@ public void setMinMax(Binary min, Binary max) {
   public BinaryStatistics copy() {
     return new BinaryStatistics(this);
   }
+
+  public void setGeometryStatistics(GeometryStatistics geometryStatistics) {
+    this.geometryStatistics = geometryStatistics;
+  }
+
+  public GeometryStatistics getGeometryStatistics() {
+    return geometryStatistics;
+  }
 }

parquet-column/src/main/java/org/apache/parquet/column/statistics/Statistics.java

@@ -20,6 +20,7 @@
 
 import java.util.Arrays;
 import org.apache.parquet.column.UnknownColumnTypeException;
+import org.apache.parquet.column.statistics.geometry.GeometryStatistics;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.schema.Float16;
 import org.apache.parquet.schema.LogicalTypeAnnotation;
@@ -64,6 +65,10 @@ public Builder withNumNulls(long numNulls) {
       return this;
     }
 
+    public Builder withGeometryStatistics(GeometryStatistics geometryStatistics) {
+      throw new UnsupportedOperationException("Please use the GeometryBuilder");
+    }
+
     public Statistics<?> build() {
       Statistics<?> stats = createStats(type);
       if (min != null && max != null) {
@@ -178,6 +183,30 @@ public Statistics<?> build() {
     }
   }
 
+  // Builder for GEOMETRY type to handle GeometryStatistics
+  private static class GeometryBuilder extends Builder {
+
+    private GeometryStatistics geometryStatistics;
+
+    public GeometryBuilder(PrimitiveType type) {
+      super(type);
+      assert type.getPrimitiveTypeName() == PrimitiveTypeName.BINARY;
+    }
+
+    @Override
+    public Builder withGeometryStatistics(GeometryStatistics geometryStatistics) {
+      this.geometryStatistics = geometryStatistics;
+      return this;
+    }
+
+    @Override
+    public Statistics<?> build() {
+      BinaryStatistics stats = (BinaryStatistics) super.build();
+      stats.setGeometryStatistics(geometryStatistics);
+      return stats;
+    }
+  }
+
   private final PrimitiveType type;
   private final PrimitiveComparator<T> comparator;
   private boolean hasNonNullValue;
@@ -269,6 +298,11 @@ public static Builder getBuilderForReading(PrimitiveType type) {
         if (logicalTypeAnnotation instanceof LogicalTypeAnnotation.Float16LogicalTypeAnnotation) {
           return new Float16Builder(type);
         }
+        return new Builder(type);
+      case BINARY:
+        if (type.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.GeometryLogicalTypeAnnotation) {
+          return new GeometryBuilder(type);
+        }
       default:
         return new Builder(type);
     }

parquet-column/src/main/java/org/apache/parquet/column/statistics/geometry/BoundingBox.java

@@ -0,0 +1,175 @@
+/*
+ * 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.
+ */
+package org.apache.parquet.column.statistics.geometry;
+
+import org.apache.parquet.Preconditions;
+import org.locationtech.jts.geom.Coordinate;
+import org.locationtech.jts.geom.Envelope;
+import org.locationtech.jts.geom.Geometry;
+
+public class BoundingBox {
+
+  private double xMin = Double.POSITIVE_INFINITY;
+  private double xMax = Double.NEGATIVE_INFINITY;
+  private double yMin = Double.POSITIVE_INFINITY;
+  private double yMax = Double.NEGATIVE_INFINITY;
+  private double zMin = Double.POSITIVE_INFINITY;
+  private double zMax = Double.NEGATIVE_INFINITY;
+  private double mMin = Double.POSITIVE_INFINITY;
+  private double mMax = Double.NEGATIVE_INFINITY;
+
+  public BoundingBox(
+      double xMin, double xMax, double yMin, double yMax, double zMin, double zMax, double mMin, double mMax) {
+    this.xMin = xMin;
+    this.xMax = xMax;
+    this.yMin = yMin;
+    this.yMax = yMax;
+    this.zMin = zMin;
+    this.zMax = zMax;
+    this.mMin = mMin;
+    this.mMax = mMax;
+  }
+
+  public BoundingBox() {}
+
+  public double getXMin() {
+    return xMin;
+  }
+
+  public double getXMax() {
+    return xMax;
+  }
+
+  public double getYMin() {
+    return yMin;
+  }
+
+  public double getYMax() {
+    return yMax;
+  }
+
+  public double getZMin() {
+    return zMin;
+  }
+
+  public double getZMax() {
+    return zMax;
+  }
+
+  public double getMMin() {
+    return mMin;
+  }
+
+  public double getMMax() {
+    return mMax;
+  }
+
+  void update(double minX, double maxX, double minY, double maxY, double minZ, double maxZ) {
+    xMin = Math.min(xMin, minX);
+    yMin = Math.min(yMin, minY);
+    xMax = Math.max(xMax, maxX);
+    yMax = Math.max(yMax, maxY);
+    zMin = Math.min(zMin, minZ);
+    zMax = Math.max(zMax, maxZ);
+  }
+
+  void update(Geometry geometry) {
+    GeometryUtils.normalizeLongitude(geometry);
+    Envelope envelope = geometry.getEnvelopeInternal();
+    double minX = envelope.getMinX();
+    double minY = envelope.getMinY();
+    double maxX = envelope.getMaxX();
+    double maxY = envelope.getMaxY();
+
+    // JTS (Java Topology Suite) does not handle Z-coordinates directly in the Envelope class
+    // because it's primarily used for 2D geometries. However, we can iterate through the
+    // coordinates of the geometry to find the minimum and maximum Z values.
+    double minZ = Double.POSITIVE_INFINITY;
+    double maxZ = Double.NEGATIVE_INFINITY;
+
+    Coordinate[] coordinates = geometry.getCoordinates();
+    for (Coordinate coord : coordinates) {
+      if (!Double.isNaN(coord.getZ())) {
+        // Update zMin and zMax by iterating through the coordinates.
+        minZ = Math.min(minZ, coord.getZ());
+        maxZ = Math.max(maxZ, coord.getZ());
+      }
+    }
+
+    update(minX, maxX, minY, maxY, minZ, maxZ);
+  }
+
+  // Method to merge a Geometry object into this bounding box
+  public void merge(Geometry geometry) {
+    Preconditions.checkArgument(geometry != null, "Cannot merge with null geometry");
+    GeometryUtils.normalizeLongitude(geometry);
+    update(geometry);
+  }
+
+  public void merge(BoundingBox other) {
+    Preconditions.checkArgument(other != null, "Cannot merge with null bounding box");
+    xMin = Math.min(xMin, other.xMin);
+    xMax = Math.max(xMax, other.xMax);
+    yMin = Math.min(yMin, other.yMin);
+    yMax = Math.max(yMax, other.yMax);
+    zMin = Math.min(zMin, other.zMin);
+    zMax = Math.max(zMax, other.zMax);
+    mMin = Math.min(mMin, other.mMin);
+    mMax = Math.max(mMax, other.mMax);
+  }
+
+  public void reset() {
+    xMin = Double.POSITIVE_INFINITY;
+    xMax = Double.NEGATIVE_INFINITY;
+    yMin = Double.POSITIVE_INFINITY;
+    yMax = Double.NEGATIVE_INFINITY;
+    zMin = Double.POSITIVE_INFINITY;
+    zMax = Double.NEGATIVE_INFINITY;
+    mMin = Double.POSITIVE_INFINITY;
+    mMax = Double.NEGATIVE_INFINITY;
+  }
+
+  public void abort() {
+    xMin = Double.NaN;
+    xMax = Double.NaN;
+    yMin = Double.NaN;
+    yMax = Double.NaN;
+    zMin = Double.NaN;
+    zMax = Double.NaN;
+    mMin = Double.NaN;
+    mMax = Double.NaN;
+  }
+
+  public BoundingBox copy() {
+    return new BoundingBox(xMin, xMax, yMin, yMax, zMin, zMax, mMin, mMax);
+  }
+
+  @Override
+  public String toString() {
+    return "BoundingBox{" + "xMin="
+        + xMin + ", xMax="
+        + xMax + ", yMin="
+        + yMin + ", yMax="
+        + yMax + ", zMin="
+        + zMin + ", zMax="
+        + zMax + ", mMin="
+        + mMin + ", mMax="
+        + mMax + '}';
+  }
+}