API, Core: Add geometry and geography types support

api/src/main/java/org/apache/iceberg/Geography.java

@@ -0,0 +1,127 @@
+/*
+ * 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.iceberg;
+
+import java.io.Serializable;
+import java.util.Locale;
+import java.util.Objects;
+import org.locationtech.jts.geom.Geometry;
+
+/**
+ * Geospatial features from OGC – Simple feature access. The geometry is on a spherical or
+ * ellipsoidal surface. An edge-interpolation algorithm is used to evaluate spatial predicates.
+ */
+public class Geography implements Comparable<Geography>, Serializable {
+
+  /** The algorithm for interpolating edges. */
+  public enum EdgeInterpolationAlgorithm {
+    /** Edges are interpolated as geodesics on a sphere. */
+    SPHERICAL("spherical"),
+    /** See <a href="https://en.wikipedia.org/wiki/Vincenty%27s_formulae">Vincenty's formulae</a> */
+    VINCENTY("vincenty"),
+    /**
+     * Thomas, Paul D. Spheroidal geodesics, reference systems, &amp; local geometry. US Naval
+     * Oceanographic Office, 1970.
+     */
+    THOMAS("thomas"),
+    /**
+     * Thomas, Paul D. Mathematical models for navigation systems. US Naval Oceanographic Office,
+     * 1965.
+     */
+    ANDOYER("andoyer"),
+    /**
+     * <a href="https://link.springer.com/content/pdf/10.1007/s00190-012-0578-z.pdf">Karney, Charles
+     * FF. "Algorithms for geodesics." Journal of Geodesy 87 (2013): 43-55 </a>, and <a
+     * href="https://geographiclib.sourceforge.io/">GeographicLib</a>.
+     */
+    KARNEY("karney");
+
+    private final String value;
+
+    EdgeInterpolationAlgorithm(String value) {
+      this.value = value;
+    }
+
+    public String value() {
+      return value;
+    }
+
+    public static EdgeInterpolationAlgorithm fromName(String algorithmName) {
+      try {
+        return EdgeInterpolationAlgorithm.valueOf(algorithmName.toUpperCase(Locale.ENGLISH));
+      } catch (IllegalArgumentException e) {
+        throw new IllegalArgumentException(
+            String.format("Invalid edge interpolation algorithm name: %s", algorithmName), e);
+      }
+    }
+  }
+
+  private final Geometry geometry;
+
+  public Geography(Geometry geometry) {
+    this.geometry = geometry;
+  }
+
+  public Geometry geometry() {
+    return geometry;
+  }
+
+  @Override
+  public String toString() {
+    return "Geography(" + geometry + ")";
+  }
+
+  @Override
+  public int compareTo(Geography o) {
+    return geometry.compareTo(o.geometry);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (!(o instanceof Geography)) {
+      return false;
+    }
+    Geography geography = (Geography) o;
+    return Objects.equals(geometry, geography.geometry);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(geometry);
+  }
+
+  public boolean intersects(Geography other, EdgeInterpolationAlgorithm algorithm) {
+    if (algorithm != EdgeInterpolationAlgorithm.SPHERICAL) {
+      throw new UnsupportedOperationException(
+          "Interpolation algorithm other than spherical is not supported yet");
+    }
+
+    // TODO: implement a correct spherical intersection algorithm using S2
+    return geometry.intersects(other.geometry);
+  }
+
+  public boolean covers(Geography other, EdgeInterpolationAlgorithm algorithm) {
+    if (algorithm != EdgeInterpolationAlgorithm.SPHERICAL) {
+      throw new UnsupportedOperationException(
+          "Interpolation algorithm other than spherical is not supported yet");
+    }
+    // TODO: implement a correct spherical covers algorithm using S2
+    return geometry.covers(other.geometry);
+  }
+}

api/src/main/java/org/apache/iceberg/Schema.java

@@ -63,7 +63,9 @@ public class Schema implements Serializable {
       ImmutableMap.of(
           Type.TypeID.TIMESTAMP_NANO, 3,
           Type.TypeID.VARIANT, 3,
-          Type.TypeID.UNKNOWN, 3);
+          Type.TypeID.UNKNOWN, 3,
+          Type.TypeID.GEOMETRY, 3,
+          Type.TypeID.GEOGRAPHY, 3);
 
   private final StructType struct;
   private final int schemaId;

api/src/main/java/org/apache/iceberg/expressions/BoundLiteralPredicate.java

@@ -20,9 +20,12 @@
 
 import java.util.Comparator;
 import java.util.Set;
+import org.apache.iceberg.Geography;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.locationtech.jts.geom.Geometry;
 
 public class BoundLiteralPredicate<T> extends BoundPredicate<T> {
   private static final Set<Type.TypeID> INTEGRAL_TYPES =
@@ -88,11 +91,64 @@ public boolean test(T value) {
         return String.valueOf(value).startsWith((String) literal.value());
       case NOT_STARTS_WITH:
         return !String.valueOf(value).startsWith((String) literal.value());
+      case ST_INTERSECTS:
+      case ST_COVERS:
+      case ST_DISJOINT:
+      case ST_NOT_COVERS:
+        return testSpatial(value);
       default:
         throw new IllegalStateException("Invalid operation for BoundLiteralPredicate: " + op());
     }
   }
 
+  private boolean testSpatial(T value) {
+    Type type = term().type();
+    Type.TypeID typeId = type.typeId();
+    if (typeId == Type.TypeID.GEOMETRY) {
+      return testGeometry((Geometry) value);
+    } else if (typeId == Type.TypeID.GEOGRAPHY) {
+      Types.GeographyType geographyType = (Types.GeographyType) type;
+      Geography.EdgeInterpolationAlgorithm algorithm = geographyType.algorithm();
+      return testGeography((Geography) value, algorithm);
+    } else {
+      throw new IllegalStateException("Invalid term type for spatial predicate: " + type);
+    }
+  }
+
+  private boolean testGeometry(Geometry value) {
+    Geometry literalGeometry = (Geometry) literal.value();
+    switch (op()) {
+      case ST_INTERSECTS:
+        return value.intersects(literalGeometry);
+      case ST_COVERS:
+        return value.covers(literalGeometry);
+      case ST_DISJOINT:
+        return value.disjoint(literalGeometry);
+      case ST_NOT_COVERS:
+        return !value.covers(literalGeometry);
+      default:
+        throw new IllegalStateException(
+            "Invalid spatial operation for BoundLiteralPredicate: " + op());
+    }
+  }
+
+  private boolean testGeography(Geography value, Geography.EdgeInterpolationAlgorithm algorithm) {
+    Geography literalGeography = (Geography) literal.value();
+    switch (op()) {
+      case ST_INTERSECTS:
+        return value.intersects(literalGeography, algorithm);
+      case ST_COVERS:
+        return value.covers(literalGeography, algorithm);
+      case ST_DISJOINT:
+        return !value.intersects(literalGeography, algorithm);
+      case ST_NOT_COVERS:
+        return !value.covers(literalGeography, algorithm);
+      default:
+        throw new IllegalStateException(
+            "Invalid spatial operation for BoundLiteralPredicate: " + op());
+    }
+  }
+
   @Override
   @SuppressWarnings("unchecked")
   public boolean isEquivalentTo(Expression expr) {
@@ -159,6 +215,14 @@ public String toString() {
         return term() + " startsWith \"" + literal + "\"";
       case NOT_STARTS_WITH:
         return term() + " notStartsWith \"" + literal + "\"";
+      case ST_INTERSECTS:
+        return term() + " intersects " + literal;
+      case ST_COVERS:
+        return term() + " covers " + literal;
+      case ST_DISJOINT:
+        return term() + " disjoint " + literal;
+      case ST_NOT_COVERS:
+        return term() + " notCovers " + literal;
       case IN:
         return term() + " in { " + literal + " }";
       case NOT_IN:

api/src/main/java/org/apache/iceberg/expressions/Evaluator.java

@@ -21,10 +21,14 @@
 import java.io.Serializable;
 import java.util.Comparator;
 import java.util.Set;
+import org.apache.iceberg.Geography;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.expressions.ExpressionVisitors.BoundVisitor;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.NaNUtil;
+import org.locationtech.jts.geom.Geometry;
 
 /**
  * Evaluates an {@link Expression} for data described by a {@link StructType}.
@@ -156,5 +160,49 @@ public <T> Boolean startsWith(Bound<T> valueExpr, Literal<T> lit) {
     public <T> Boolean notStartsWith(Bound<T> valueExpr, Literal<T> lit) {
       return !startsWith(valueExpr, lit);
     }
+
+    @Override
+    public <T> Boolean stIntersects(Bound<T> valueExpr, Literal<T> lit) {
+      T evalRes = valueExpr.eval(struct);
+      if (evalRes == null) {
+        return false;
+      }
+      Type type = valueExpr.ref().type();
+      if (type.typeId() == Type.TypeID.GEOMETRY) {
+        return ((Geometry) evalRes).intersects((Geometry) lit.value());
+      } else if (type.typeId() == Type.TypeID.GEOGRAPHY) {
+        Types.GeographyType geographyType = (Types.GeographyType) type;
+        return ((Geography) evalRes).intersects((Geography) lit.value(), geographyType.algorithm());
+      } else {
+        throw new IllegalArgumentException("Invalid type for spatial predicate: " + type);
+      }
+    }
+
+    @Override
+    public <T> Boolean stCovers(Bound<T> valueExpr, Literal<T> lit) {
+      T evalRes = valueExpr.eval(struct);
+      if (evalRes == null) {
+        return false;
+      }
+      Type type = valueExpr.ref().type();
+      if (type.typeId() == Type.TypeID.GEOMETRY) {
+        return ((Geometry) evalRes).covers((Geometry) lit.value());
+      } else if (type.typeId() == Type.TypeID.GEOGRAPHY) {
+        Types.GeographyType geographyType = (Types.GeographyType) type;
+        return ((Geography) evalRes).covers((Geography) lit.value(), geographyType.algorithm());
+      } else {
+        throw new IllegalArgumentException("Invalid type for spatial predicate: " + type);
+      }
+    }
+
+    @Override
+    public <T> Boolean stDisjoint(Bound<T> valueExpr, Literal<T> lit) {
+      return !stIntersects(valueExpr, lit);
+    }
+
+    @Override
+    public <T> Boolean stNotCovers(Bound<T> valueExpr, Literal<T> lit) {
+      return !stCovers(valueExpr, lit);
+    }
   }
 }

api/src/main/java/org/apache/iceberg/expressions/Expression.java

@@ -44,6 +44,10 @@ enum Operation {
     OR,
     STARTS_WITH,
     NOT_STARTS_WITH,
+    ST_INTERSECTS,
+    ST_COVERS,
+    ST_DISJOINT,
+    ST_NOT_COVERS,
     COUNT,
     COUNT_STAR,
     MAX,
@@ -90,6 +94,14 @@ public Operation negate() {
           return Operation.NOT_STARTS_WITH;
         case NOT_STARTS_WITH:
           return Operation.STARTS_WITH;
+        case ST_INTERSECTS:
+          return Operation.ST_DISJOINT;
+        case ST_COVERS:
+          return Operation.ST_NOT_COVERS;
+        case ST_DISJOINT:
+          return Operation.ST_INTERSECTS;
+        case ST_NOT_COVERS:
+          return Operation.ST_COVERS;
         default:
           throw new IllegalArgumentException("No negation for operation: " + this);
       }

api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java

@@ -333,6 +333,10 @@ public <T> Expression predicate(UnboundPredicate<T> pred) {
         case NOT_EQ:
         case STARTS_WITH:
         case NOT_STARTS_WITH:
+        case ST_INTERSECTS:
+        case ST_COVERS:
+        case ST_DISJOINT:
+        case ST_NOT_COVERS:
           return new UnboundPredicate<>(
               pred.op(), pred.term(), (T) sanitize(pred.literal(), now, today));
         case IN:
@@ -485,6 +489,14 @@ public <T> String predicate(UnboundPredicate<T> pred) {
           return term + " STARTS WITH " + sanitize(pred.literal(), nowMicros, today);
         case NOT_STARTS_WITH:
           return term + " NOT STARTS WITH " + sanitize(pred.literal(), nowMicros, today);
+        case ST_INTERSECTS:
+          return term + " ST_INTERSECTS " + sanitize(pred.literal(), nowMicros, today);
+        case ST_COVERS:
+          return term + " ST_COVERS " + sanitize(pred.literal(), nowMicros, today);
+        case ST_DISJOINT:
+          return term + " ST_DISJOINT " + sanitize(pred.literal(), nowMicros, today);
+        case ST_NOT_COVERS:
+          return term + " ST_NOT_COVERS " + sanitize(pred.literal(), nowMicros, today);
         default:
           throw new UnsupportedOperationException(
               "Cannot sanitize unsupported predicate type: " + pred.op());
@@ -568,6 +580,10 @@ private static String sanitize(Literal<?> literal, long now, int today) {
       return sanitizeNumber(((Literals.FloatLiteral) literal).value(), "float");
     } else if (literal instanceof Literals.DoubleLiteral) {
       return sanitizeNumber(((Literals.DoubleLiteral) literal).value(), "float");
+    } else if (literal instanceof Literals.GeometryLiteral) {
+      return "(geometry)";
+    } else if (literal instanceof Literals.GeographyLiteral) {
+      return "(geography)";
     } else {
       // for uuid, decimal, fixed, variant, and binary, match the string result
       return sanitizeSimpleString(literal.value().toString());