HIVE-29699: Iceberg: Add support for the Unknown data type

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

@@ -186,6 +186,8 @@ Type convertType(TypeInfo typeInfo, String defaultValue, boolean shouldAddInitia
         return Types.ListType.ofOptional(listId, listType);
       case VARIANT:
         return Types.VariantType.get();
+      case UNKNOWN:
+        return Types.UnknownType.get();
       default:
         throw new IllegalArgumentException("Unknown type " + typeInfo.getCategory());
     }

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

@@ -411,6 +411,8 @@ public static String convertToTypeString(Type type) {
         return String.format("map<%s,%s>", convert(mapType.keyType()), convert(mapType.valueType()));
       case VARIANT:
         return "variant";
+      case UNKNOWN:
+        return "unknown";
       default:
         throw new UnsupportedOperationException(type + " is not supported");
     }

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

@@ -26,6 +26,7 @@
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.serde.serdeConstants;
 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.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -171,6 +172,35 @@ public void testConversionWithoutLastComment() {
     assertThat(schema.asStruct()).isEqualTo(expected.asStruct());
   }
 
+  @Test
+  public void testUnknownTypeConvertToIcebergSchema() {
+    List<FieldSchema> hiveSchema = ImmutableList.of(
+        new FieldSchema("id", serdeConstants.INT_TYPE_NAME, null),
+        new FieldSchema("placeholder", serdeConstants.UNKNOWN_TYPE_NAME, "placeholder comment"));
+    Schema expected = new Schema(
+        optional(1, "id", Types.IntegerType.get()),
+        optional(2, "placeholder", Types.UnknownType.get(), "placeholder comment"));
+
+    assertThat(HiveSchemaUtil.convert(hiveSchema).asStruct()).isEqualTo(expected.asStruct());
+  }
+
+  @Test
+  public void testUnknownTypeConvertToHiveSchema() {
+    Schema icebergSchema = new Schema(
+        optional(1, "id", Types.IntegerType.get()),
+        optional(2, "placeholder", Types.UnknownType.get(), "placeholder comment"));
+    List<FieldSchema> expected = ImmutableList.of(
+        new FieldSchema("id", serdeConstants.INT_TYPE_NAME, null),
+        new FieldSchema("placeholder", serdeConstants.UNKNOWN_TYPE_NAME, "placeholder comment"));
+
+    assertThat(HiveSchemaUtil.convert(icebergSchema)).isEqualTo(expected);
+  }
+
+  @Test
+  public void testUnknownTypeAndTypeInfoConvert() {
+    checkConvert(TypeInfoFactory.getUnknownTypeInfo(), Types.UnknownType.get());
+  }
+
   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/Deserializer.java

@@ -37,6 +37,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.schema.SchemaWithPartnerVisitor;
+import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Type.PrimitiveType;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.ListType;
@@ -110,6 +111,15 @@ public FieldDeserializer field(NestedField field, ObjectInspectorPair pair, Fiel
 
     @Override
     public FieldDeserializer primitive(PrimitiveType type, ObjectInspectorPair pair) {
+      if (type.typeId() == Type.TypeID.UNKNOWN) {
+        return o -> {
+          if (o != null) {
+            throw new IllegalArgumentException("UNKNOWN type columns only accept NULL values");
+          }
+          return null;
+        };
+      }
+
       return o -> {
         if (o == null) {
           return null;

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

@@ -23,6 +23,7 @@
 import javax.annotation.Nullable;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.UnknownObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
 import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
@@ -106,6 +107,8 @@ public ObjectInspector primitive(Type.PrimitiveType primitiveType) {
             IcebergTimestampObjectInspectorHive3.get(9);
       case TIME:
         return IcebergTimeObjectInspector.get();
+      case UNKNOWN:
+        return UnknownObjectInspector.get();
       default:
         throw new IllegalArgumentException(primitiveType.typeId() + " type is not supported");
     }

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

@@ -104,6 +104,8 @@ public static Object convert(Type type, Object object) {
       case DOUBLE:
       case STRING:
         return object;
+      case UNKNOWN:
+        return null;
       default:
         throw new UnsupportedOperationException("Not a supported type: " + type);
     }

iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergUnknownType.java

@@ -0,0 +1,89 @@
+/*
+ * 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.mr.hive;
+
+import java.util.Collection;
+import java.util.List;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.mr.hive.test.TestTables.TestTableType;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runners.Parameterized.Parameters;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+
+public class TestHiveIcebergUnknownType extends HiveIcebergStorageHandlerWithEngineBase {
+
+  @Parameters(name = "fileFormat={0}, catalog={1}, isVectorized={2}, formatVersion={3}")
+  public static Collection<Object[]> parameters() {
+    return HiveIcebergStorageHandlerWithEngineBase.getParameters(p ->
+        p.fileFormat() == FileFormat.PARQUET &&
+        p.testTableType() == TestTableType.HIVE_CATALOG &&
+        p.formatVersion() == 3);
+  }
+
+  @Test
+  public void testUnknownTypeReadWrite() {
+    Schema schema = new Schema(
+        optional(1, "id", Types.IntegerType.get()),
+        optional(2, "placeholder", Types.UnknownType.get()));
+
+    TableIdentifier table = TableIdentifier.of("default", "unknown_type_basic");
+
+    testTables.createTable(shell, table.name(),
+        schema, org.apache.iceberg.PartitionSpec.unpartitioned(), fileFormat, ImmutableList.of(), formatVersion,
+        ImmutableMap.of());
+
+    shell.executeStatement(
+        String.format("INSERT INTO %s VALUES (1, NULL), (2, NULL)", table));
+
+    List<Object[]> rows = shell.executeStatement(
+        String.format("SELECT id, placeholder FROM %s ORDER BY id", table));
+
+    Assert.assertEquals(2, rows.size());
+    Assert.assertEquals(1, rows.get(0)[0]);
+    Assert.assertNull(rows.get(0)[1]);
+    Assert.assertEquals(2, rows.get(1)[0]);
+    Assert.assertNull(rows.get(1)[1]);
+  }
+
+  @Test
+  public void testUnknownTypeDdl() {
+    TableIdentifier table = TableIdentifier.of("default", "unknown_type_ddl");
+
+    shell.executeStatement(String.format(
+        "CREATE EXTERNAL TABLE %s (id INT, placeholder UNKNOWN) STORED BY ICEBERG " +
+            "TBLPROPERTIES ('format-version'='3')", table));
+
+    shell.executeStatement(String.format("INSERT INTO %s VALUES (1, NULL)", table));
+
+    List<Object[]> rows = shell.executeStatement(
+        String.format("SELECT id, placeholder FROM %s", table));
+
+    Assert.assertEquals(1, rows.size());
+    Assert.assertEquals(1, rows.get(0)[0]);
+    Assert.assertNull(rows.get(0)[1]);
+  }
+}

iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/serde/objectinspector/TestIcebergObjectInspector.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
 import org.apache.hadoop.hive.serde2.objectinspector.StructField;
 import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.UnknownObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
 import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
@@ -32,6 +33,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.apache.iceberg.types.Types.NestedField.required;
 
 
@@ -60,7 +62,8 @@ public class TestIcebergObjectInspector {
                   Types.NestedField.required(20, "nested_field", Types.StringType.get(), "nested field comment")),
                   "struct comment"
           ),
-          required(21, "time_field", Types.TimeType.get(), "time comment")
+          required(21, "time_field", Types.TimeType.get(), "time comment"),
+          optional(22, "unknown_field", Types.UnknownType.get(), "unknown comment")
   );
 
   @Test
@@ -205,6 +208,13 @@ public void testIcebergObjectInspector() {
     Assertions.assertEquals("time_field", timeField.getFieldName());
     Assertions.assertEquals("time comment", timeField.getFieldComment());
     Assertions.assertEquals(IcebergTimeObjectInspector.get(), timeField.getFieldObjectInspector());
+
+    // unknown
+    StructField unknownField = soi.getStructFieldRef("unknown_field");
+    Assertions.assertEquals(22, unknownField.getFieldID());
+    Assertions.assertEquals("unknown_field", unknownField.getFieldName());
+    Assertions.assertEquals("unknown comment", unknownField.getFieldComment());
+    Assertions.assertEquals(UnknownObjectInspector.get(), unknownField.getFieldObjectInspector());
   }
 
   private static ObjectInspector getPrimitiveObjectInspector(Class<?> clazz) {

iceberg/iceberg-handler/src/test/queries/positive/unknown_type.q

@@ -0,0 +1,22 @@
+-- Create test table with unknown placeholder column
+CREATE EXTERNAL TABLE unknown_test_basic (
+    id INT,
+    placeholder UNKNOWN
+) STORED BY ICEBERG tblproperties('format-version'='3');
+
+-- Unknown columns are not stored; only NULL values are accepted
+INSERT INTO unknown_test_basic VALUES
+(1, NULL),
+(2, NULL);
+
+SELECT id, placeholder FROM unknown_test_basic ORDER BY id;
+
+-- Add another unknown column to an existing table
+ALTER TABLE unknown_test_basic ADD COLUMNS (extra UNKNOWN);
+
+INSERT INTO unknown_test_basic VALUES
+(3, NULL, NULL);
+
+SELECT id, placeholder, extra FROM unknown_test_basic ORDER BY id;
+
+DESC FORMATTED unknown_test_basic;

iceberg/iceberg-handler/src/test/results/positive/unknown_type.q.out

@@ -0,0 +1,115 @@
+PREHOOK: query: CREATE EXTERNAL TABLE unknown_test_basic (
+    id INT,
+    placeholder UNKNOWN
+) STORED BY ICEBERG tblproperties('format-version'='3')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@unknown_test_basic
+POSTHOOK: query: CREATE EXTERNAL TABLE unknown_test_basic (
+    id INT,
+    placeholder UNKNOWN
+) STORED BY ICEBERG tblproperties('format-version'='3')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@unknown_test_basic
+PREHOOK: query: INSERT INTO unknown_test_basic VALUES
+(1, NULL),
+(2, NULL)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@unknown_test_basic
+POSTHOOK: query: INSERT INTO unknown_test_basic VALUES
+(1, NULL),
+(2, NULL)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@unknown_test_basic
+PREHOOK: query: SELECT id, placeholder FROM unknown_test_basic ORDER BY id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@unknown_test_basic
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: SELECT id, placeholder FROM unknown_test_basic ORDER BY id
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@unknown_test_basic
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+1	NULL
+2	NULL
+PREHOOK: query: ALTER TABLE unknown_test_basic ADD COLUMNS (extra UNKNOWN)
+PREHOOK: type: ALTERTABLE_ADDCOLS
+PREHOOK: Input: default@unknown_test_basic
+PREHOOK: Output: default@unknown_test_basic
+POSTHOOK: query: ALTER TABLE unknown_test_basic ADD COLUMNS (extra UNKNOWN)
+POSTHOOK: type: ALTERTABLE_ADDCOLS
+POSTHOOK: Input: default@unknown_test_basic
+POSTHOOK: Output: default@unknown_test_basic
+PREHOOK: query: INSERT INTO unknown_test_basic VALUES
+(3, NULL, NULL)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@unknown_test_basic
+POSTHOOK: query: INSERT INTO unknown_test_basic VALUES
+(3, NULL, NULL)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@unknown_test_basic
+PREHOOK: query: SELECT id, placeholder, extra FROM unknown_test_basic ORDER BY id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@unknown_test_basic
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: SELECT id, placeholder, extra FROM unknown_test_basic ORDER BY id
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@unknown_test_basic
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+1	NULL	NULL
+2	NULL	NULL
+3	NULL	NULL
+PREHOOK: query: DESC FORMATTED unknown_test_basic
+PREHOOK: type: DESCTABLE
+PREHOOK: Input: default@unknown_test_basic
+POSTHOOK: query: DESC FORMATTED unknown_test_basic
+POSTHOOK: type: DESCTABLE
+POSTHOOK: Input: default@unknown_test_basic
+# col_name            	data_type           	comment             
+id                  	int                 	                    
+placeholder         	unknown             	                    
+extra               	unknown             	                    
+
+# Detailed Table Information	 	 
+Database:           	default             	 
+#### A masked pattern was here ####
+Retention:          	0                   	 
+#### A masked pattern was here ####
+Table Type:         	EXTERNAL_TABLE      	 
+Table Parameters:	 	 
+	COLUMN_STATS_ACCURATE	{\"BASIC_STATS\":\"true\"}
+	EXTERNAL            	TRUE                
+	bucketing_version   	2                   
+	current-schema      	{\"type\":\"struct\",\"schema-id\":1,\"fields\":[{\"id\":1,\"name\":\"id\",\"required\":false,\"type\":\"int\"},{\"id\":2,\"name\":\"placeholder\",\"required\":false,\"type\":\"unknown\"},{\"id\":3,\"name\":\"extra\",\"required\":false,\"type\":\"unknown\"}]}
+	current-snapshot-id 	#Masked# 
+	current-snapshot-summary	{\"added-data-files\":\"1\",\"added-records\":\"1\",\"added-files-size\":\"526\",\"changed-partition-count\":\"1\",\"total-records\":\"3\",\"total-files-size\":\"997\",\"total-data-files\":\"2\",\"total-delete-files\":\"0\",\"total-position-deletes\":\"0\",\"total-equality-deletes\":\"0\",\"iceberg-version\":\"Apache Iceberg 1.10.1 (commit ccb8bc435062171e64bc8b7e5f56e6aed9c5b934)\"}
+	current-snapshot-timestamp-ms	#Masked#       
+	format-version      	3                   
+	metadata_location   	hdfs://### HDFS PATH ###
+	numFiles            	2                   
+	numRows             	3                   
+	parquet.compression 	zstd                
+	previous_metadata_location	hdfs://### HDFS PATH ###
+	rawDataSize         	0                   
+	serialization.format	1                   
+	snapshot-count      	2                   
+	storage_handler     	org.apache.iceberg.mr.hive.HiveIcebergStorageHandler
+	table_type          	ICEBERG             
+	totalSize           	#Masked#
+#### A masked pattern was here ####
+	uuid                	#Masked#
+	write.delete.mode   	merge-on-read       
+	write.merge.mode    	merge-on-read       
+	write.metadata.delete-after-commit.enabled	true                
+	write.update.mode   	merge-on-read       
+
+# Storage Information	 	 
+SerDe Library:      	org.apache.iceberg.mr.hive.HiveIcebergSerDe	 
+InputFormat:        	org.apache.iceberg.mr.hive.HiveIcebergInputFormat	 
+OutputFormat:       	org.apache.iceberg.mr.hive.HiveIcebergOutputFormat	 
+Compressed:         	No                  	 
+Sort Columns:       	[]                  	 

jdbc/src/java/org/apache/hive/jdbc/JdbcColumn.java

@@ -168,6 +168,8 @@ static Type typeStringToHiveType(String type) throws SQLException {
       /* fall through */
     case "null":
       return Type.NULL_TYPE;
+    case serdeConstants.UNKNOWN_TYPE_NAME:
+      return Type.UNKNOWN_TYPE;
     default:
       throw new SQLException("Unrecognized column type: " + type);
     }
@@ -224,6 +226,8 @@ static String getColumnTypeName(String type) throws SQLException {
         /* fall through */
       case "null":
         return serdeConstants.VOID_TYPE_NAME;
+      case serdeConstants.UNKNOWN_TYPE_NAME:
+        return serdeConstants.UNKNOWN_TYPE_NAME;
       case "map":
         return serdeConstants.MAP_TYPE_NAME;
       case "array":