fix NPE problem when convert flink object for Map

Author: hehuiyuan

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/functions/hive/conversion/HiveInspectors.java

@@ -332,6 +332,9 @@ public static Object toFlinkObject(ObjectInspector inspector, Object data, HiveS
         if (inspector instanceof MapObjectInspector) {
             MapObjectInspector mapInspector = (MapObjectInspector) inspector;
             Map<?, ?> map = mapInspector.getMap(data);
+            if (map == null) {
+                return new HashMap<>();
+            }
 
             Map<Object, Object> result = new HashMap<>(map.size());
             for (Map.Entry<?, ?> entry : map.entrySet()) {

flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/TableEnvHiveConnectorITCase.java

@@ -34,7 +34,6 @@
 import org.apache.flink.types.Row;
 import org.apache.flink.util.CollectionUtil;
 import org.apache.flink.util.IOUtils;
-
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
@@ -52,6 +51,7 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.flink.table.planner.runtime.utils.BatchAbstractTestBase.TEMPORARY_FOLDER;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -531,6 +531,52 @@ public void testLocationWithComma() throws Exception {
         }
     }
 
+    @Test
+    public void testReadHiveDataWithEmptyMapForHiveShim20X() throws Exception {
+        TableEnvironment tableEnv = getTableEnvWithHiveCatalog();
+
+        try {
+            // Flink to write parquet file
+            String folderURI = TEMPORARY_FOLDER.newFolder().toURI().toString();
+            tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+            String format = "parquet";
+            tableEnv.executeSql(
+                    String.format(
+                            "create table parquet_t (i int, j int) with ("
+                                    + "'connector'='filesystem','format'='%s','path'='%s')",
+                            format, folderURI));
+            tableEnv.executeSql("insert into parquet_t select 1, 2").await();
+            tableEnv.executeSql("drop table parquet_t");
+
+            // Hive to read parquet file and write to another table for empty map data
+            tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+            tableEnv.getConfig()
+                    .getConfiguration()
+                    .set(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, true);
+            tableEnv.executeSql(
+                    String.format(
+                            "create external table src_t (i int, j int) stored as %s location '%s'",
+                            format, folderURI));
+            tableEnv.executeSql(
+                    String.format(
+                            "create table target_t (a int, b map<string, string>) stored as parquet"));
+
+            tableEnv.executeSql(
+                    "insert into target_t select i, map[cast(null as string), cast(null as string)] from src_t");
+
+            // Hive to read parquet table with empty map value
+            List<Row> results =
+                    CollectionUtil.iteratorToList(
+                            tableEnv.sqlQuery("select * from target_t").execute().collect());
+        } finally {
+            tableEnv.getConfig()
+                    .getConfiguration()
+                    .set(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER, false);
+            tableEnv.executeSql("drop table if exists target_t");
+            tableEnv.executeSql("drop table if exists src_t");
+        }
+    }
+
     private TableEnvironment getTableEnvWithHiveCatalog() {
         TableEnvironment tableEnv =
                 HiveTestUtils.createTableEnvWithBlinkPlannerBatchMode(SqlDialect.HIVE);