[DX] spark sql support GeometryUDT type.

Author: freamdx

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

@@ -45,7 +45,8 @@ enum TypeID {
     DECIMAL(BigDecimal.class),
     STRUCT(StructLike.class),
     LIST(List.class),
-    MAP(Map.class);
+    MAP(Map.class),
+    GEOMETRY(Types.GeometryType.class);
 
     private final Class<?> javaClass;
 

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

@@ -615,6 +615,10 @@ public T map(Types.MapType map, T keyResult, T valueResult) {
     public T primitive(Type.PrimitiveType primitive) {
       return null;
     }
+
+    public T geometry(Type type) {
+      return null;
+    }
   }
 
   public static <T> T visit(Schema schema, SchemaVisitor<T> visitor) {
@@ -675,6 +679,9 @@ public static <T> T visit(Type type, SchemaVisitor<T> visitor) {
 
         return visitor.map(map, keyResult, valueResult);
 
+      case GEOMETRY:
+        return visitor.geometry(type);
+
       default:
         return visitor.primitive(type.asPrimitiveType());
     }

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

@@ -55,6 +55,7 @@ private Types() {}
           .put(StringType.get().toString(), StringType.get())
           .put(UUIDType.get().toString(), UUIDType.get())
           .put(BinaryType.get().toString(), BinaryType.get())
+          .put(GeometryType.get().toString().toLowerCase(Locale.ROOT), GeometryType.get())
           .buildOrThrow();
 
   private static final Pattern FIXED = Pattern.compile("fixed\\[\\s*(\\d+)\\s*\\]");
@@ -1053,4 +1054,22 @@ private List<NestedField> lazyFieldList() {
       return fields;
     }
   }
+
+  public static class GeometryType extends PrimitiveType {
+    private static final GeometryType INSTANCE = new GeometryType();
+
+    public static GeometryType get() {
+      return INSTANCE;
+    }
+
+    @Override
+    public TypeID typeId() {
+      return TypeID.GEOMETRY;
+    }
+
+    @Override
+    public String toString() {
+      return "Geometry";
+    }
+  }
 }

arrow/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java

@@ -66,6 +66,7 @@ public static Field convert(final NestedField field) {
 
     switch (field.type().typeId()) {
       case BINARY:
+      case GEOMETRY:
         arrowType = ArrowType.Binary.INSTANCE;
         break;
       case FIXED:

arrow/src/test/java/org/apache/iceberg/arrow/ArrowSchemaUtilTest.java

@@ -56,6 +56,7 @@ public class ArrowSchemaUtilTest {
   private static final String LIST_FIELD = "lt";
   private static final String MAP_FIELD = "mt";
   private static final String UUID_FIELD = "uu";
+  private static final String GEOM_FIELD = "geom";
 
   @Test
   public void convertPrimitive() {
@@ -79,7 +80,8 @@ public void convertPrimitive() {
                 MAP_FIELD,
                 Types.MapType.ofOptional(15, 16, StringType.get(), IntegerType.get())),
             Types.NestedField.optional(17, FIXED_WIDTH_BINARY_FIELD, Types.FixedType.ofLength(10)),
-            Types.NestedField.optional(18, UUID_FIELD, Types.UUIDType.get()));
+            Types.NestedField.optional(18, UUID_FIELD, Types.UUIDType.get()),
+            Types.NestedField.optional(19, GEOM_FIELD, Types.GeometryType.get()));
 
     org.apache.arrow.vector.types.pojo.Schema arrow = ArrowSchemaUtil.convert(iceberg);
 
@@ -179,6 +181,10 @@ private void validate(Type iceberg, Field field, boolean optional) {
         assertThat(field.getName()).isEqualTo(UUID_FIELD);
         assertThat(arrowType.getTypeID()).isEqualTo(ArrowType.FixedSizeBinary.TYPE_TYPE);
         break;
+      case GEOMETRY:
+        assertThat(field.getName()).isEqualTo(GEOM_FIELD);
+        assertThat(arrowType.getTypeID()).isEqualTo(ArrowType.Binary.TYPE_TYPE);
+        break;
       default:
         throw new UnsupportedOperationException("Check not implemented for type: " + iceberg);
     }

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveSchemaUtil.java

@@ -166,6 +166,7 @@ private static String convertToTypeString(Type type) {
         return "timestamp";
       case FIXED:
       case BINARY:
+      case GEOMETRY:
         return "binary";
       case DECIMAL:
         final Types.DecimalType decimalType = (Types.DecimalType) type;

orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java

@@ -58,7 +58,13 @@ static SearchArgument convert(Expression expr, TypeDescription readSchema) {
   // these Iceberg types
   private static final Set<TypeID> UNSUPPORTED_TYPES =
       ImmutableSet.of(
-          TypeID.BINARY, TypeID.FIXED, TypeID.UUID, TypeID.STRUCT, TypeID.MAP, TypeID.LIST);
+          TypeID.BINARY,
+          TypeID.FIXED,
+          TypeID.UUID,
+          TypeID.STRUCT,
+          TypeID.MAP,
+          TypeID.LIST,
+          TypeID.GEOMETRY);
 
   private final SearchArgument.Builder builder;
   private final Map<Integer, String> idToColumnName;

orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java

@@ -40,7 +40,8 @@ public final class ORCSchemaUtil {
   public enum BinaryType {
     UUID,
     FIXED,
-    BINARY
+    BINARY,
+    GEOMETRY
   }
 
   public enum LongType {
@@ -101,6 +102,7 @@ public TypeDescription type() {
           .put(Type.TypeID.FIXED, TypeDescription.Category.BINARY)
           .put(Type.TypeID.BINARY, TypeDescription.Category.BINARY)
           .put(Type.TypeID.DECIMAL, TypeDescription.Category.DECIMAL)
+          .put(Type.TypeID.GEOMETRY, TypeDescription.Category.BINARY)
           .build();
 
   private ORCSchemaUtil() {}
@@ -202,6 +204,10 @@ private static TypeDescription convert(Integer fieldId, Type type, boolean isReq
           orcType = TypeDescription.createMap(keyType, valueType);
           break;
         }
+      case GEOMETRY:
+        orcType = TypeDescription.createBinary();
+        orcType.setAttribute(ICEBERG_BINARY_TYPE_ATTRIBUTE, BinaryType.GEOMETRY.toString());
+        break;
       default:
         throw new IllegalArgumentException("Unhandled type " + type.typeId());
     }

orc/src/main/java/org/apache/iceberg/orc/OrcToIcebergVisitor.java

@@ -160,6 +160,10 @@ public Optional<Types.NestedField> primitive(TypeDescription primitive) {
           case BINARY:
             foundField = Types.NestedField.of(icebergID, isOptional, name, Types.BinaryType.get());
             break;
+          case GEOMETRY:
+            foundField =
+                Types.NestedField.of(icebergID, isOptional, name, Types.GeometryType.get());
+            break;
           default:
             throw new IllegalStateException("Invalid Binary type found in ORC type attribute");
         }

orc/src/test/java/org/apache/iceberg/orc/TestEstimateOrcAvgWidthVisitor.java

@@ -72,6 +72,8 @@ public class TestEstimateOrcAvgWidthVisitor {
               required(24, "booleanField", Types.BooleanType.get()),
               optional(25, "date", Types.DateType.get()),
               optional(27, "timestamp", Types.TimestampType.withZone())));
+  protected static final Types.NestedField GEOM_FIELD =
+      optional(24, "geometry", Types.GeometryType.get());
 
   @Test
   public void testEstimateIntegerWidth() {
@@ -225,6 +227,16 @@ public void testEstimateFullWidth() {
         .isEqualTo(611);
   }
 
+  @Test
+  public void testEstimateGeometryWidth() {
+    Schema geomSchema = new Schema(GEOM_FIELD);
+    TypeDescription geomOrcSchema = ORCSchemaUtil.convert(geomSchema);
+    long estimateLength = getEstimateLength(geomOrcSchema);
+    assertThat(estimateLength)
+        .as("Estimated average length of geometry must be 128.")
+        .isEqualTo(128);
+  }
+
   private Integer getEstimateLength(TypeDescription orcSchemaWithDate) {
     return OrcSchemaVisitor.visitSchema(orcSchemaWithDate, new EstimateOrcAvgWidthVisitor())
         .stream()