Add VARIANT type to ColumnInfoTypeName for correct SEA metadata

NEXT_CHANGELOG.md

@@ -11,6 +11,7 @@
 - Fixed `PARSE_SYNTAX_ERROR` for column names containing special characters (e.g., dots) when `EnableBatchedInserts` is enabled, by re-quoting column names with backticks in reconstructed multi-row INSERT statements.
 - Fixed Volume ingestion for SEA mode, which was broken due to statement being closed prematurely.
 - Fixed escaped pattern characters in catalogName for `getSchemas`, as returned catalogName should be unescaped.
+- Fixed `getColumnClassName()` returning null for VARIANT columns in SEA mode by adding VARIANT to the type system.
 - Fixed `getColumns()` returning `DATA_TYPE=0` (NULL) for GEOMETRY/GEOGRAPHY columns in Thrift mode. Now returns `Types.VARCHAR` (12) when geospatial is disabled and `Types.OTHER` (1111) when enabled, consistent with SEA mode.
 
 ---

src/main/java/com/databricks/jdbc/common/util/DatabricksTypeUtil.java

@@ -116,6 +116,8 @@ public static ColumnInfoTypeName getColumnInfoType(String typeName) {
         return ColumnInfoTypeName.MAP;
       case DatabricksTypeUtil.INTERVAL:
         return ColumnInfoTypeName.INTERVAL;
+      case DatabricksTypeUtil.VARIANT:
+        return ColumnInfoTypeName.VARIANT;
     }
     return ColumnInfoTypeName.USER_DEFINED_TYPE;
   }
@@ -163,6 +165,7 @@ public static int getColumnType(ColumnInfoTypeName typeName) {
         return Types.STRUCT;
       case ARRAY:
         return Types.ARRAY;
+      case VARIANT:
       case GEOMETRY:
       case GEOGRAPHY:
       case USER_DEFINED_TYPE:
@@ -203,6 +206,7 @@ public static String getColumnTypeClassName(ColumnInfoTypeName typeName) {
       case CHAR:
       case STRING:
       case INTERVAL:
+      case VARIANT:
       case USER_DEFINED_TYPE:
         return "java.lang.String";
       case TIMESTAMP:

src/main/java/com/databricks/jdbc/model/core/ColumnInfoTypeName.java

@@ -30,5 +30,6 @@ public enum ColumnInfoTypeName {
   STRING,
   STRUCT,
   TIMESTAMP,
-  USER_DEFINED_TYPE;
+  USER_DEFINED_TYPE,
+  VARIANT;
 }

src/test/java/com/databricks/jdbc/common/util/DatabricksTypeUtilTest.java

@@ -91,7 +91,8 @@ void testGetColumnType() {
             Map.entry(ColumnInfoTypeName.ARRAY, Types.ARRAY),
             Map.entry(ColumnInfoTypeName.GEOMETRY, Types.OTHER),
             Map.entry(ColumnInfoTypeName.GEOGRAPHY, Types.OTHER),
-            Map.entry(ColumnInfoTypeName.USER_DEFINED_TYPE, Types.OTHER));
+            Map.entry(ColumnInfoTypeName.USER_DEFINED_TYPE, Types.OTHER),
+            Map.entry(ColumnInfoTypeName.VARIANT, Types.OTHER));
 
     expectedMappings.forEach(
         (typeName, expectedSqlType) ->
@@ -132,7 +133,8 @@ void testGetColumnTypeClassName() {
             Map.entry(ColumnInfoTypeName.GEOGRAPHY, GEOGRAPHY_CLASS_NAME),
             Map.entry(ColumnInfoTypeName.MAP, "java.util.Map"),
             Map.entry(ColumnInfoTypeName.NULL, "null"),
-            Map.entry(ColumnInfoTypeName.VOID, "null"));
+            Map.entry(ColumnInfoTypeName.VOID, "null"),
+            Map.entry(ColumnInfoTypeName.VARIANT, "java.lang.String"));
 
     expectedMappings.forEach(
         (columnType, expectedClassName) ->
@@ -191,6 +193,7 @@ void testGetMetadataColPrecision() {
   void testIsSigned() {
     assertTrue(DatabricksTypeUtil.isSigned(ColumnInfoTypeName.INT));
     assertFalse(DatabricksTypeUtil.isSigned(ColumnInfoTypeName.BOOLEAN));
+    assertFalse(DatabricksTypeUtil.isSigned(ColumnInfoTypeName.VARIANT));
   }
 
   @Test
@@ -285,6 +288,7 @@ void testInferDatabricksType() {
     "MAP, MAP",
     "CHAR, STRING",
     "INTERVAL, INTERVAL",
+    "VARIANT, VARIANT",
     "UNKNOWN, USER_DEFINED_TYPE"
   })
   public void testGetColumnInfoType(String inputTypeName, String expectedTypeName) {