[FLINK-22294][hive] Hive reading fail when getting file numbers on different filesystem nameservices

Author: zuston

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceFileEnumerator.java

@@ -110,13 +110,10 @@ public static InputSplit[] createMRSplits(
     public static int getNumFiles(List<HiveTablePartition> partitions, JobConf jobConf)
             throws IOException {
         int numFiles = 0;
-        FileSystem fs = null;
         for (HiveTablePartition partition : partitions) {
             StorageDescriptor sd = partition.getStorageDescriptor();
             org.apache.hadoop.fs.Path inputPath = new org.apache.hadoop.fs.Path(sd.getLocation());
-            if (fs == null) {
-                fs = inputPath.getFileSystem(jobConf);
-            }
+            FileSystem fs = inputPath.getFileSystem(jobConf);
             // it's possible a partition exists in metastore but the data has been removed
             if (!fs.exists(inputPath)) {
                 continue;

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java

@@ -342,22 +342,4 @@ public BaseStatistics getStatistics(BaseStatistics cachedStats) {
     public InputSplitAssigner getInputSplitAssigner(HiveTableInputSplit[] inputSplits) {
         return new LocatableInputSplitAssigner(inputSplits);
     }
-
-    public int getNumFiles() throws IOException {
-        int numFiles = 0;
-        FileSystem fs = null;
-        for (HiveTablePartition partition : partitions) {
-            StorageDescriptor sd = partition.getStorageDescriptor();
-            Path inputPath = new Path(sd.getLocation());
-            if (fs == null) {
-                fs = inputPath.getFileSystem(jobConf.conf());
-            }
-            // it's possible a partition exists in metastore but the data has been removed
-            if (!fs.exists(inputPath)) {
-                continue;
-            }
-            numFiles += fs.listStatus(inputPath).length;
-        }
-        return numFiles;
-    }
 }