Apache hive 4.0.0 support.

arenadata · Aug 15, 2024 · f439792 · f439792
1 parent e4d17ad
commit f439792
Showing 1 changed file with 33 additions and 0 deletions.
diff --git a/...connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV400.java b/...connector-hive/src/main/java/org/apache/flink/table/catalog/hive/client/HiveShimV400.java
@@ -20,14 +20,20 @@
 
 import org.apache.flink.table.catalog.exceptions.CatalogException;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
 import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
 import org.apache.hadoop.hive.ql.io.HivePassThroughOutputFormat;
+import org.apache.hadoop.hive.serde2.Deserializer;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.Reporter;
 
+import java.lang.reflect.Method;
+import java.util.List;
 import java.util.Properties;
 
 /** Shim for Hive version 4.0.0. */
@@ -52,4 +58,31 @@ public FileSinkOperator.RecordWriter getHiveRecordWriter(
             throw new CatalogException("Failed to create Hive RecordWriter", e);
         }
     }
+
+    public List<FieldSchema> getFieldsFromDeserializer(
+            Configuration conf, Table table, boolean skipConfError) {
+        try {
+            Method utilMethod =
+                    getHiveMetaStoreUtilsClass()
+                            .getMethod(
+                                    "getDeserializer",
+                                    Configuration.class,
+                                    Table.class,
+                                    String.class,
+                                    boolean.class);
+            Deserializer deserializer =
+                    (Deserializer) utilMethod.invoke(null, conf, table, null, skipConfError);
+            utilMethod =
+                    getHiveMetaStoreUtilsClass()
+                            .getMethod(
+                                    "getFieldsFromDeserializer",
+                                    String.class,
+                                    Deserializer.class,
+                                    Configuration.class);
+            return (List<FieldSchema>)
+                    utilMethod.invoke(null, table.getTableName(), deserializer, conf);
+        } catch (Exception e) {
+            throw new CatalogException("Failed to get table schema from deserializer", e);
+        }
+    }
 }