HIVE-30240: Iceberg: Add support for GEOMETRY data type

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveSchemaConverter.java

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.GeometryTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
@@ -148,6 +149,8 @@ Type convertType(TypeInfo typeInfo, String defaultValue, boolean shouldAddInitia
           case DECIMAL:
             DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) typeInfo;
             return Types.DecimalType.of(decimalTypeInfo.precision(), decimalTypeInfo.scale());
+          case GEOMETRY:
+            return Types.GeometryType.of(((GeometryTypeInfo) typeInfo).getCrs());
           case INTERVAL_YEAR_MONTH:
           case INTERVAL_DAY_TIME:
           default:

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

@@ -29,6 +29,7 @@
 import java.util.stream.Collectors;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
 import org.apache.hive.iceberg.com.fasterxml.jackson.core.type.TypeReference;
 import org.apache.hive.iceberg.com.fasterxml.jackson.databind.ObjectMapper;
@@ -413,6 +414,8 @@ public static String convertToTypeString(Type type) {
         return "variant";
       case UNKNOWN:
         return "unknown";
+      case GEOMETRY:
+        return TypeInfoFactory.getGeometryTypeInfo(((Types.GeometryType) type).crs()).getQualifiedName();
       default:
         throw new UnsupportedOperationException(type + " is not supported");
     }
@@ -623,6 +626,8 @@ public static Object getDefaultValue(String defaultValue, Type type) {
     return switch (type.typeId()) {
       case DATE, TIME, TIMESTAMP, TIMESTAMP_NANO ->
           Literal.of(stripQuotes(defaultValue)).to(type);
+      case GEOMETRY -> throw new UnsupportedOperationException(
+          "Default values for type " + type + " are not supported");
       default -> Conversions.fromPartitionString(type, stripQuotes(defaultValue));
     };
   }

iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveSchemaUtil.java

@@ -201,6 +201,52 @@ public void testUnknownTypeAndTypeInfoConvert() {
     checkConvert(TypeInfoFactory.getUnknownTypeInfo(), Types.UnknownType.get());
   }
 
+  @Test
+  public void testGeometryTypeConvertToIcebergSchema() {
+    List<FieldSchema> hiveSchema = ImmutableList.of(
+        new FieldSchema("id", serdeConstants.INT_TYPE_NAME, null),
+        new FieldSchema("geom", serdeConstants.GEOMETRY_TYPE_NAME, "geometry comment"),
+        new FieldSchema("geom_3857", "geometry('EPSG:3857')", null));
+    Schema expected = new Schema(
+        optional(1, "id", Types.IntegerType.get()),
+        optional(2, "geom", Types.GeometryType.crs84(), "geometry comment"),
+        optional(3, "geom_3857", Types.GeometryType.of("EPSG:3857")));
+
+    assertThat(HiveSchemaUtil.convert(hiveSchema).asStruct()).isEqualTo(expected.asStruct());
+  }
+
+  @Test
+  public void testGeometryTypeConvertToHiveSchema() {
+    Schema icebergSchema = new Schema(
+        optional(1, "id", Types.IntegerType.get()),
+        optional(2, "geom", Types.GeometryType.crs84(), "geometry comment"),
+        optional(3, "geom_3857", Types.GeometryType.of("EPSG:3857")));
+    List<FieldSchema> expected = ImmutableList.of(
+        new FieldSchema("id", serdeConstants.INT_TYPE_NAME, null),
+        new FieldSchema("geom", serdeConstants.GEOMETRY_TYPE_NAME, "geometry comment"),
+        new FieldSchema("geom_3857", "geometry('EPSG:3857')", null));
+
+    assertThat(HiveSchemaUtil.convert(icebergSchema)).isEqualTo(expected);
+  }
+
+  @Test
+  public void testGeometryTypeAndTypeInfoConvert() {
+    checkConvert(TypeInfoFactory.geometryTypeInfo, Types.GeometryType.crs84());
+    checkConvert(TypeInfoFactory.getGeometryTypeInfo("OGC:CRS84"), Types.GeometryType.crs84());
+    checkConvert(TypeInfoFactory.getGeometryTypeInfo("EPSG:3857"), Types.GeometryType.of("EPSG:3857"));
+    checkConvert(TypeInfoFactory.getListTypeInfo(TypeInfoFactory.getGeometryTypeInfo("EPSG:4326")),
+        Types.ListType.ofOptional(1, Types.GeometryType.of("EPSG:4326")));
+  }
+
+  @Test
+  public void testGeometryTypeDefaultValueNotSupported() {
+    assertThat(HiveSchemaUtil.getDefaultValue(null, Types.GeometryType.crs84())).isNull();
+    assertThat(HiveSchemaUtil.getDefaultValue("NULL", Types.GeometryType.crs84())).isNull();
+    assertThatThrownBy(() -> HiveSchemaUtil.getDefaultValue("'0101000000'", Types.GeometryType.crs84()))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageContaining("Default values for type geometry");
+  }
+
   protected List<FieldSchema> getSupportedFieldSchemas() {
     List<FieldSchema> fields = Lists.newArrayList();
     fields.add(new FieldSchema("c_float", serdeConstants.FLOAT_TYPE_NAME, "float comment"));

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/BaseHiveIcebergMetaHook.java

@@ -29,6 +29,7 @@
 import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.commons.lang3.StringUtils;
@@ -74,6 +75,8 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -201,6 +204,7 @@ public void preCreateTable(CreateTableRequest request) {
     }
 
     assertFileFormat(tableProperties.getProperty(TableProperties.DEFAULT_FILE_FORMAT));
+    assertFileFormatSupportsSchema(tableProperties.getProperty(TableProperties.DEFAULT_FILE_FORMAT), () -> schema);
 
     // Remove hive primary key columns from table request, as iceberg doesn't support hive primary key.
     request.setPrimaryKeys(null);
@@ -390,6 +394,30 @@ private static void assertFileFormat(String format) {
         String.format("Unsupported fileformat %s", format));
   }
 
+  /**
+   * Iceberg cannot write geospatial columns to ORC files, so reject the combination before any data is written.
+   */
+  static void assertFileFormatSupportsSchema(String format, Supplier<Schema> schemaSupplier) {
+    if (format == null || !format.toLowerCase().contains(FileFormat.ORC.getLabel())) {
+      return;
+    }
+    Schema schema = schemaSupplier.get();
+    TypeUtil.indexByName(schema.asStruct()).entrySet().stream()
+        .filter(entry -> isGeospatial(schema.findType(entry.getValue())))
+        .map(Map.Entry::getKey)
+        .sorted()
+        .findFirst()
+        .ifPresent(column -> {
+          throw new IllegalArgumentException(String.format(
+              "Column %s has type %s, which is not supported by the ORC file format. Use Parquet or Avro instead",
+              column, schema.findType(column)));
+        });
+  }
+
+  private static boolean isGeospatial(Type type) {
+    return type.typeId() == Type.TypeID.GEOMETRY || type.typeId() == Type.TypeID.GEOGRAPHY;
+  }
+
   protected void setCommonHmsTablePropertiesForIceberg(org.apache.hadoop.hive.metastore.api.Table hmsTable) {
     if (IcebergCatalogProperties.isHadoopTable(conf, tableProperties)) {
       String location = (hmsTable.getSd() != null) ? hmsTable.getSd().getLocation() : null;

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java

@@ -388,6 +388,10 @@ private void doPreAlterTable(org.apache.hadoop.hive.metastore.api.Table hmsTable
       hmsTable.getParameters().put(TableProperties.DEFAULT_NAME_MAPPING, NameMappingParser.toJson(nameMapping));
     }
 
+    if (!isTableMigration) {
+      assertFileFormatSupportsSchema(alteredFileFormat(hmsTable), () -> alteredSchema(hmsTable));
+    }
+
     if (AlterTableType.ADDCOLS.equals(currentAlterTableOp)) {
       handleAddColumns(hmsTable);
     } else if (AlterTableType.REPLACE_COLUMNS.equals(currentAlterTableOp)) {
@@ -775,6 +779,19 @@ private void setupAlterOperationType(org.apache.hadoop.hive.metastore.api.Table
     }
   }
 
+  private String alteredFileFormat(org.apache.hadoop.hive.metastore.api.Table hmsTable) {
+    return hmsTable.getParameters().getOrDefault(TableProperties.DEFAULT_FILE_FORMAT,
+        icebergTable.properties().get(TableProperties.DEFAULT_FILE_FORMAT));
+  }
+
+  private Schema alteredSchema(org.apache.hadoop.hive.metastore.api.Table hmsTable) {
+    if (AlterTableType.ADDCOLS.equals(currentAlterTableOp) ||
+        AlterTableType.REPLACE_COLUMNS.equals(currentAlterTableOp)) {
+      return HiveSchemaUtil.convert(hmsTable.getSd().getCols(), Collections.emptyMap(), true);
+    }
+    return icebergTable.schema();
+  }
+
   private void setFileFormat(String format) {
     if (format == null) {
       return;

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/serde/objectinspector/IcebergBinaryObjectInspector.java

@@ -20,31 +20,69 @@
 package org.apache.iceberg.mr.hive.serde.objectinspector;
 
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.AbstractPrimitiveJavaObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.BinaryObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
 import org.apache.hadoop.io.BytesWritable;
+import org.apache.iceberg.geospatial.GeospatialBound;
 import org.apache.iceberg.util.ByteBuffers;
 
 public class IcebergBinaryObjectInspector extends AbstractPrimitiveJavaObjectInspector
     implements BinaryObjectInspector, WriteObjectInspector {
 
-  private static final IcebergBinaryObjectInspector INSTANCE = new IcebergBinaryObjectInspector();
+  private static final IcebergBinaryObjectInspector INSTANCE =
+      new IcebergBinaryObjectInspector(TypeInfoFactory.binaryTypeInfo);
+
+  private static final Map<PrimitiveTypeInfo, IcebergBinaryObjectInspector> INSTANCES = new ConcurrentHashMap<>();
 
   public static IcebergBinaryObjectInspector get() {
     return INSTANCE;
   }
 
-  private IcebergBinaryObjectInspector() {
-    super(TypeInfoFactory.binaryTypeInfo);
+  /**
+   * Returns an inspector for a Hive type whose values are binary, for example GEOMETRY.
+   */
+  public static IcebergBinaryObjectInspector get(PrimitiveTypeInfo typeInfo) {
+    if (TypeInfoFactory.binaryTypeInfo.equals(typeInfo)) {
+      return INSTANCE;
+    }
+    return INSTANCES.computeIfAbsent(typeInfo, IcebergBinaryObjectInspector::new);
+  }
+
+  private IcebergBinaryObjectInspector(PrimitiveTypeInfo typeInfo) {
+    super(typeInfo);
   }
 
   @Override
   public byte[] getPrimitiveJavaObject(Object o) {
+    if (o instanceof GeospatialBound bound) {
+      return toWkbPoint(bound);
+    }
     return ByteBuffers.toByteArray((ByteBuffer) o);
   }
 
+  /**
+   * Metadata tables report the bounds of geospatial columns as {@link GeospatialBound} points rather than WKB.
+   */
+  private static byte[] toWkbPoint(GeospatialBound bound) {
+    int dimensionCode = (bound.hasZ() ? 1 : 0) + (bound.hasM() ? 2 : 0);
+    int dimensions = 2 + (bound.hasZ() ? 1 : 0) + (bound.hasM() ? 1 : 0);
+    ByteBuffer wkb = ByteBuffer.allocate(1 + 4 + dimensions * Double.BYTES).order(ByteOrder.LITTLE_ENDIAN);
+    wkb.put((byte) 1).putInt(1 + 1000 * dimensionCode).putDouble(bound.x()).putDouble(bound.y());
+    if (bound.hasZ()) {
+      wkb.putDouble(bound.z());
+    }
+    if (bound.hasM()) {
+      wkb.putDouble(bound.m());
+    }
+    return wkb.array();
+  }
+
   @Override
   public BytesWritable getPrimitiveWritableObject(Object o) {
     return o == null ? null : new BytesWritable(getPrimitiveJavaObject(o));

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/serde/objectinspector/IcebergObjectInspector.java

@@ -109,6 +109,9 @@ public ObjectInspector primitive(Type.PrimitiveType primitiveType) {
         return IcebergTimeObjectInspector.get();
       case UNKNOWN:
         return UnknownObjectInspector.get();
+      case GEOMETRY:
+        return IcebergBinaryObjectInspector.get(
+            TypeInfoFactory.getGeometryTypeInfo(((Types.GeometryType) primitiveType).crs()));
       default:
         throw new IllegalArgumentException(primitiveType.typeId() + " type is not supported");
     }

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveValueConverter.java

@@ -88,6 +88,7 @@ public static Object convert(Type type, Object object) {
           return LocalDateTime.ofInstant(((Timestamp) object).toSqlTimestamp().toInstant(), ZoneId.of("UTC"));
         }
       case BINARY:
+      case GEOMETRY:
         return ByteBuffer.wrap((byte[]) object);
       case FIXED:
         return (byte[]) object;