API, Core: Add geometry and geography types support

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/transforms/Identity.java

@@ -19,16 +19,21 @@
 package org.apache.iceberg.transforms;
 
 import java.io.ObjectStreamException;
+import java.util.Set;
 import org.apache.iceberg.expressions.BoundPredicate;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.UnboundPredicate;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.util.SerializableFunction;
 
 class Identity<T> implements Transform<T, T> {
   private static final Identity<?> INSTANCE = new Identity<>();
 
+  private static final Set<Type.TypeID> UNSUPPORTED_TYPES =
+      ImmutableSet.of(Type.TypeID.VARIANT, Type.TypeID.GEOMETRY, Type.TypeID.GEOGRAPHY);
+
   private final Type type;
 
   /**
@@ -39,7 +44,7 @@ class Identity<T> implements Transform<T, T> {
   @Deprecated
   public static <I> Identity<I> get(Type type) {
     Preconditions.checkArgument(
-        type.typeId() != Type.TypeID.VARIANT, "Unsupported type for identity: %s", type);
+        !UNSUPPORTED_TYPES.contains(type.typeId()), "Unsupported type for identity: %s", type);
 
     return new Identity<>(type);
   }
@@ -93,6 +98,10 @@ public SerializableFunction<T, T> bind(Type type) {
 
   @Override
   public boolean canTransform(Type maybePrimitive) {
+    if (UNSUPPORTED_TYPES.contains(maybePrimitive.typeId())) {
+      return false;
+    }
+
     return maybePrimitive.isPrimitiveType();
   }
 

api/src/main/java/org/apache/iceberg/types/EdgeAlgorithm.java

@@ -0,0 +1,61 @@
+/*
+ * 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.types;
+
+import java.util.Locale;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/** The algorithm for interpolating edges. */
+public enum EdgeAlgorithm {
+  /** Edges are interpolated as geodesics on a sphere. */
+  SPHERICAL,
+  /** See <a href="https://en.wikipedia.org/wiki/Vincenty%27s_formulae">Vincenty's formulae</a> */
+  VINCENTY,
+  /**
+   * Thomas, Paul D. Spheroidal geodesics, reference systems, &amp; local geometry. US Naval
+   * Oceanographic Office, 1970.
+   */
+  THOMAS,
+  /**
+   * Thomas, Paul D. Mathematical models for navigation systems. US Naval Oceanographic Office,
+   * 1965.
+   */
+  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;
+
+  public static EdgeAlgorithm fromName(String algorithmName) {
+    Preconditions.checkNotNull(algorithmName, "Invalid edge interpolation algorithm: null");
+    try {
+      return EdgeAlgorithm.valueOf(algorithmName.toUpperCase(Locale.ENGLISH));
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException(
+          String.format("Invalid edge interpolation algorithm: %s", algorithmName), e);
+    }
+  }
+
+  @Override
+  public String toString() {
+    return name().toLowerCase(Locale.ENGLISH);
+  }
+}

api/src/main/java/org/apache/iceberg/types/Type.java

@@ -44,6 +44,8 @@ enum TypeID {
     FIXED(ByteBuffer.class),
     BINARY(ByteBuffer.class),
     DECIMAL(BigDecimal.class),
+    GEOMETRY(ByteBuffer.class),
+    GEOGRAPHY(ByteBuffer.class),
     STRUCT(StructLike.class),
     LIST(List.class),
     MAP(Map.class),

api/src/main/java/org/apache/iceberg/types/TypeUtil.java

@@ -540,6 +540,12 @@ private static int estimateSize(Type type) {
       case BINARY:
       case VARIANT:
         return 80;
+      case GEOMETRY:
+      case GEOGRAPHY:
+        // 80 bytes is an approximate size for a polygon or linestring with 4 to 5 coordinates.
+        // This is a reasonable estimate for the size of a geometry or geography object without
+        // additional details.
+        return 80;
       case UNKNOWN:
         // Consider Unknown as null
         return 0;

api/src/main/java/org/apache/iceberg/types/Types.java

@@ -58,9 +58,16 @@ private Types() {}
           .put(BinaryType.get().toString(), BinaryType.get())
           .put(UnknownType.get().toString(), UnknownType.get())
           .put(VariantType.get().toString(), VariantType.get())
+          .put(GeometryType.crs84().toString(), GeometryType.crs84())
+          .put(GeographyType.crs84().toString(), GeographyType.crs84())
           .buildOrThrow();
 
   private static final Pattern FIXED = Pattern.compile("fixed\\[\\s*(\\d+)\\s*\\]");
+  private static final Pattern GEOMETRY_PARAMETERS =
+      Pattern.compile("geometry\\s*(?:\\(\\s*([^,]+?)\\s*\\))?", Pattern.CASE_INSENSITIVE);
+  private static final Pattern GEOGRAPHY_PARAMETERS =
+      Pattern.compile(
+          "geography\\s*(?:\\(\\s*([^,]+)\\s*(?:,\\s*(\\w*)\\s*)?\\))?", Pattern.CASE_INSENSITIVE);
   private static final Pattern DECIMAL =
       Pattern.compile("decimal\\(\\s*(\\d+)\\s*,\\s*(\\d+)\\s*\\)");
 
@@ -70,6 +77,21 @@ public static Type fromTypeName(String typeString) {
       return TYPES.get(lowerTypeString);
     }
 
+    Matcher geometry = GEOMETRY_PARAMETERS.matcher(typeString);
+    if (geometry.matches()) {
+      String crs = geometry.group(1);
+      return GeometryType.of(crs != null ? crs.trim() : null);
+    }
+
+    Matcher geography = GEOGRAPHY_PARAMETERS.matcher(typeString);
+    if (geography.matches()) {
+      String crs = geography.group(1);
+      String algorithmName = geography.group(2);
+      EdgeAlgorithm algorithm =
+          algorithmName == null ? null : EdgeAlgorithm.fromName(algorithmName.trim());
+      return GeographyType.of(crs != null ? crs.trim() : null, algorithm);
+    }
+
     Matcher fixed = FIXED.matcher(lowerTypeString);
     if (fixed.matches()) {
       return FixedType.ofLength(Integer.parseInt(fixed.group(1)));
@@ -543,6 +565,148 @@ public int hashCode() {
     }
   }
 
+  public static class GeometryType extends PrimitiveType {
+
+    private final String crs;
+
+    private GeometryType(String crs) {
+      if (crs != null) {
+        Preconditions.checkArgument(!crs.isEmpty(), "Invalid CRS: (empty string)");
+        Preconditions.checkArgument(
+            crs.trim().equals(crs), "CRS must not have leading or trailing spaces: '%s'", crs);
+        this.crs = crs;
+      } else {
+        this.crs = null;
+      }
+    }
+
+    private GeometryType() {
+      crs = null;
+    }
+
+    public static GeometryType crs84() {
+      return new GeometryType();
+    }
+
+    public static GeometryType of(String crs) {
+      return new GeometryType(crs);
+    }
+
+    @Override
+    public TypeID typeId() {
+      return TypeID.GEOMETRY;
+    }
+
+    public String crs() {
+      return crs;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      } else if (!(o instanceof GeometryType)) {
+        return false;
+      }
+
+      GeometryType that = (GeometryType) o;
+      return Objects.equals(crs, that.crs);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(GeometryType.class, crs);
+    }
+
+    @Override
+    public String toString() {
+      if (crs == null) {
+        return "geometry";
+      }
+
+      return String.format("geometry(%s)", crs);
+    }
+  }
+
+  public static class GeographyType extends PrimitiveType {
+
+    public static final String DEFAULT_CRS = "OGC:CRS84";
+
+    private final String crs;
+    private final EdgeAlgorithm algorithm;
+
+    private GeographyType(String crs, EdgeAlgorithm algorithm) {
+      if (crs != null) {
+        Preconditions.checkArgument(!crs.isEmpty(), "Invalid CRS: (empty string)");
+        Preconditions.checkArgument(
+            crs.trim().equals(crs), "CRS must not have leading or trailing spaces: '%s'", crs);
+        this.crs = crs;
+      } else {
+        this.crs = null;
+      }
+
+      this.algorithm = algorithm;
+    }
+
+    private GeographyType() {
+      this.crs = null;
+      this.algorithm = null;
+    }
+
+    public static GeographyType crs84() {
+      return new GeographyType();
+    }
+
+    public static GeographyType forCRS(String crs) {
+      return new GeographyType(crs, null);
+    }
+
+    public static GeographyType of(String crs, EdgeAlgorithm algorithm) {
+      return new GeographyType(crs, algorithm);
+    }
+
+    @Override
+    public TypeID typeId() {
+      return TypeID.GEOGRAPHY;
+    }
+
+    public String crs() {
+      return crs;
+    }
+
+    public EdgeAlgorithm algorithm() {
+      return algorithm;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      } else if (!(o instanceof GeographyType)) {
+        return false;
+      }
+
+      GeographyType that = (GeographyType) o;
+      return Objects.equals(crs, that.crs) && Objects.equals(algorithm, that.algorithm);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(GeographyType.class, crs, algorithm);
+    }
+
+    @Override
+    public String toString() {
+      if (algorithm != null) {
+        return String.format("geography(%s, %s)", crs != null ? crs : DEFAULT_CRS, algorithm);
+      } else if (crs != null) {
+        return String.format("geography(%s)", crs);
+      } else {
+        return "geography";
+      }
+    }
+  }
+
   public static class NestedField implements Serializable {
     public static NestedField optional(int id, String name, Type type) {
       return new NestedField(true, id, name, type, null, null, null);

api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java

@@ -26,6 +26,8 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.NestedField;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 public class TestPartitionSpecValidation {
   private static final Schema SCHEMA =
@@ -37,7 +39,9 @@ public class TestPartitionSpecValidation {
           NestedField.required(5, "another_d", Types.TimestampType.withZone()),
           NestedField.required(6, "s", Types.StringType.get()),
           NestedField.required(7, "v", Types.VariantType.get()),
-          NestedField.optional(8, "u", Types.UnknownType.get()));
+          NestedField.required(8, "geom", Types.GeometryType.crs84()),
+          NestedField.required(9, "geog", Types.GeographyType.crs84()),
+          NestedField.optional(10, "u", Types.UnknownType.get()));
 
   @Test
   public void testMultipleTimestampPartitions() {
@@ -316,25 +320,24 @@ public void testAddPartitionFieldsWithAndWithoutFieldIds() {
     assertThat(spec.lastAssignedFieldId()).isEqualTo(1006);
   }
 
-  @Test
-  public void testVariantUnsupported() {
+  @ParameterizedTest
+  @MethodSource("unsupportedFieldsProvider")
+  public void testUnsupported(int fieldId, String partitionName, String expectedErrorMessage) {
     assertThatThrownBy(
             () ->
                 PartitionSpec.builderFor(SCHEMA)
-                    .add(7, 1005, "variant_partition1", Transforms.bucket(5))
+                    .add(fieldId, 1005, partitionName, Transforms.bucket(5))
                     .build())
         .isInstanceOf(ValidationException.class)
-        .hasMessage("Cannot partition by non-primitive source field: variant");
+        .hasMessage(expectedErrorMessage);
   }
 
-  @Test
-  public void testUnknownUnsupported() {
-    assertThatThrownBy(
-            () ->
-                PartitionSpec.builderFor(SCHEMA)
-                    .add(8, 1005, "unknown_partition1", Transforms.bucket(5))
-                    .build())
-        .isInstanceOf(ValidationException.class)
-        .hasMessage("Invalid source type unknown for transform: bucket[5]");
+  private static Object[][] unsupportedFieldsProvider() {
+    return new Object[][] {
+      {7, "variant_partition1", "Cannot partition by non-primitive source field: variant"},
+      {8, "geom_partition1", "Invalid source type geometry for transform: bucket[5]"},
+      {9, "geog_partition1", "Invalid source type geography for transform: bucket[5]"},
+      {10, "unknown_partition1", "Invalid source type unknown for transform: bucket[5]"}
+    };
   }
 }