Backport geometry and geography types support

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

@@ -28,6 +28,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.BiMap;
@@ -54,9 +55,20 @@ public class Schema implements Serializable {
   private static final Joiner NEWLINE = Joiner.on('\n');
   private static final String ALL_COLUMNS = "*";
   private static final int DEFAULT_SCHEMA_ID = 0;
-  private static final int DEFAULT_VALUES_MIN_FORMAT_VERSION = 3;
-  private static final Map<Type.TypeID, Integer> MIN_FORMAT_VERSIONS =
-      ImmutableMap.of(Type.TypeID.TIMESTAMP_NANO, 3, Type.TypeID.VARIANT, 3);
+
+  @VisibleForTesting static final int DEFAULT_VALUES_MIN_FORMAT_VERSION = 3;
+
+  @VisibleForTesting
+  static final Map<Type.TypeID, Integer> MIN_FORMAT_VERSIONS =
+      ImmutableMap.of(
+          Type.TypeID.TIMESTAMP_NANO,
+          3,
+          Type.TypeID.VARIANT,
+          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

@@ -24,13 +24,14 @@
 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.types.Types;
 import org.apache.iceberg.util.SerializableFunction;
 
 class Identity<T> implements Transform<T, T> {
-  private static final Set<Type> UNSUPPORTED_TYPES = Set.of(Types.VariantType.get());
   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;
 
@@ -42,7 +43,7 @@ class Identity<T> implements Transform<T, T> {
   @Deprecated
   public static <I> Identity<I> get(Type type) {
     Preconditions.checkArgument(
-        !UNSUPPORTED_TYPES.contains(type), "Unsupported type for identity: %s", type);
+        !UNSUPPORTED_TYPES.contains(type.typeId()), "Unsupported type for identity: %s", type);
 
     return new Identity<>(type);
   }
@@ -88,6 +89,9 @@ 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/AssignFreshIds.java

@@ -125,8 +125,8 @@ public Type map(Types.MapType map, Supplier<Type> keyFuture, Supplier<Type> valu
   }
 
   @Override
-  public Type variant() {
-    return Types.VariantType.get();
+  public Type variant(Types.VariantType variant) {
+    return variant;
   }
 
   @Override

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

@@ -92,6 +92,11 @@ public Type map(Types.MapType map, Supplier<Type> keyFuture, Supplier<Type> valu
     }
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return variant;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

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, "Edge interpolation algorithm cannot be 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/ReassignDoc.java

@@ -96,6 +96,11 @@ public Type map(Types.MapType map, Supplier<Type> keyTypeFuture, Supplier<Type>
     }
   }
 
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return variant;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

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

@@ -162,8 +162,8 @@ public Type map(Types.MapType map, Supplier<Type> keyTypeFuture, Supplier<Type>
   }
 
   @Override
-  public Type variant() {
-    return Types.VariantType.get();
+  public Type variant(Types.VariantType variant) {
+    return variant;
   }
 
   @Override

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

@@ -43,6 +43,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),
@@ -81,6 +83,10 @@ default Types.MapType asMapType() {
     throw new IllegalArgumentException("Not a map type: " + this);
   }
 
+  default Types.VariantType asVariantType() {
+    throw new IllegalArgumentException("Not a variant type: " + this);
+  }
+
   default boolean isNestedType() {
     return false;
   }

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

@@ -395,6 +395,10 @@ public static Type find(Schema schema, Predicate<Type> predicate) {
     return visit(schema, new FindTypeVisitor(predicate));
   }
 
+  public static Type find(Type type, Predicate<Type> predicate) {
+    return visit(type, new FindTypeVisitor(predicate));
+  }
+
   public static boolean isPromotionAllowed(Type from, Type.PrimitiveType to) {
     // Warning! Before changing this function, make sure that the type change doesn't introduce
     // compatibility problems in partitioning.
@@ -536,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 DECIMAL:
         // 12 (header) + (12 + 12 + 4) (BigInteger) + 4 (scale) = 44 bytes
         return 44;
@@ -709,8 +719,8 @@ public T map(Types.MapType map, Supplier<T> keyResult, Supplier<T> valueResult)
       return null;
     }
 
-    public T variant() {
-      return null;
+    public T variant(Types.VariantType variant) {
+      throw new UnsupportedOperationException("Unsupported type: variant");
     }
 
     public T primitive(Type.PrimitiveType primitive) {
@@ -790,7 +800,7 @@ public static <T> T visit(Type type, CustomOrderSchemaVisitor<T> visitor) {
             new VisitFuture<>(map.valueType(), visitor));
 
       case VARIANT:
-        return visitor.variant();
+        return visitor.variant(type.asVariantType());
 
       default:
         return visitor.primitive(type.asPrimitiveType());