[hive] Determine timestamp ltz type based on hive runtime version

paimon-hive/paimon-hive-common/src/main/java/org/apache/paimon/hive/HiveTypeUtils.java

@@ -185,7 +185,7 @@ public TypeInfo visit(TimestampType timestampType) {
 
         @Override
         public TypeInfo visit(LocalZonedTimestampType localZonedTimestampType) {
-            return LocalZonedTimestampTypeUtils.toHiveType(localZonedTimestampType);
+            return LocalZonedTimestampTypeUtils.hiveLocalZonedTimestampType();
         }
 
         @Override
@@ -254,7 +254,7 @@ static DataType visit(TypeInfo type, HiveToPaimonTypeVisitor visitor) {
         }
 
         public DataType atomic(TypeInfo atomic) {
-            if (LocalZonedTimestampTypeUtils.isLocalZonedTimestampType(atomic)) {
+            if (LocalZonedTimestampTypeUtils.isHiveLocalZonedTimestampType(atomic)) {
                 return DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE();
             }
 

paimon-hive/paimon-hive-common/src/main/java/org/apache/paimon/hive/LocalZonedTimestampTypeUtils.java

@@ -23,14 +23,27 @@
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
 
-/** To maintain compatibility with Hive 3. */
+import java.lang.reflect.Field;
+
+/**
+ * Utils to convert between Hive TimestampLocalTZTypeInfo and Paimon {@link
+ * LocalZonedTimestampType}, using reflection to solve compatibility between Hive 2 and Hive 3.
+ */
 public class LocalZonedTimestampTypeUtils {
 
-    public static boolean isLocalZonedTimestampType(TypeInfo hiveTypeInfo) {
-        return false;
+    public static boolean isHiveLocalZonedTimestampType(TypeInfo hiveTypeInfo) {
+        return "org.apache.hadoop.hive.serde2.typeinfo.TimestampLocalTZTypeInfo"
+                .equals(hiveTypeInfo.getClass().getName());
     }
 
-    public static TypeInfo toHiveType(LocalZonedTimestampType paimonType) {
-        return TypeInfoFactory.timestampTypeInfo;
+    public static TypeInfo hiveLocalZonedTimestampType() {
+        try {
+            Class<?> typeInfoFactoryClass =
+                    Class.forName("org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory");
+            Field field = typeInfoFactoryClass.getField("timestampLocalTZTypeInfo");
+            return (TypeInfo) field.get(null);
+        } catch (Exception e) {
+            return TypeInfoFactory.timestampTypeInfo;
+        }
     }
 }