Repository: apache/iceberg (language: python)

Issue:
Use the same shared FileSystem instance across calls in HadoopTableOperations
Closes https://github.com/Netflix/iceberg/issues/92.


Discussion / hints:
@mccheah I'm not clear on the value of this change.  The file system instance will actually get reused because it's cached at the FileSystem level (https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java#L156).

I really don't have a strong opinion for either implementation.  @rdblue thoughts?


The rationale is explained in the issue that was in the previous repository. See https://github.com/Netflix/iceberg/issues/92 and https://github.com/Netflix/iceberg/issues/91.
I'm also very much averse to JVM-wide caching, as `FileSystem.get` does.
Sorry, I reviewed the copy of this in the Netflix repo: https://github.com/Netflix/iceberg/pull/108

Overall, the change is okay. I think it is fine to expose the factory method as long as it is package-private and only used by HadoopTableOperations. But that really limits the utility of it, while going against best practices for using the Hadoop FS API.

I think the best option for the use case is to subclass HadoopTables & HadoopTableOperations and replace the current behavior if you need to.
A compromise could be to have `HadoopTableOperations` have an overridable method like so:

```
FileSystem getFileSystem(Path path, Configuration conf);
```

The default implementation can just call `Util.getFs`, but overriding this can run their own caching layer.
Suggested the above in https://github.com/apache/incubator-iceberg/issues/13. If we prefer that, we can close this.
Closing in favor of #15 .

Fix patch:
diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
index a95454b8b0ee..a222080d71db 100644
--- a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
+++ b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
@@ -26,6 +26,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.function.BiFunction;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.InputFormat;
@@ -44,6 +45,7 @@
 import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.SerializableTable;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.SystemConfigs;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.TableScan;
@@ -78,6 +80,7 @@
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.util.PartitionUtil;
 import org.apache.iceberg.util.SerializationUtil;
+import org.apache.iceberg.util.ThreadPools;
 
 /**
  * Generic Mrv2 InputFormat API for Iceberg.
@@ -104,7 +107,21 @@ public List<InputSplit> getSplits(JobContext context) {
         Optional.ofNullable(
                 HiveIcebergStorageHandler.table(conf, conf.get(InputFormatConfig.TABLE_IDENTIFIER)))
             .orElseGet(() -> Catalogs.loadTable(conf));
+    final ExecutorService workerPool =
+        ThreadPools.newWorkerPool(
+            "iceberg-plan-worker-pool",
+            conf.getInt(
+                SystemConfigs.WORKER_THREAD_POOL_SIZE.propertyKey(),
+                ThreadPools.WORKER_THREAD_POOL_SIZE));
+    try {
+      return planInputSplits(table, conf, workerPool);
+    } finally {
+      workerPool.shutdown();
+    }
+  }
 
+  private List<InputSplit> planInputSplits(
+      Table table, Configuration conf, ExecutorService workerPool) {
     TableScan scan =
         table
             .newScan()
@@ -144,6 +161,7 @@ public List<InputSplit> getSplits(JobContext context) {
     InputFormatConfig.InMemoryDataModel model =
         conf.getEnum(
             InputFormatConfig.IN_MEMORY_DATA_MODEL, InputFormatConfig.InMemoryDataModel.GENERIC);
+    scan = scan.planWith(workerPool);
     try (CloseableIterable<CombinedScanTask> tasksIterable = scan.planTasks()) {
       Table serializableTable = SerializableTable.copyOf(table);
       tasksIterable.forEach(


Test patch:
diff --git a/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
index c9d5d487de38..86d390ca9ffe 100644
--- a/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
+++ b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
@@ -25,11 +25,14 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.lang.reflect.Method;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.security.PrivilegedAction;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobConf;
@@ -39,6 +42,7 @@
 import org.apache.hadoop.mapreduce.RecordReader;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
@@ -67,6 +71,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ThreadPools;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestTemplate;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -381,6 +386,45 @@ public void testCustomCatalog() throws IOException {
     testInputFormat.create(builder.conf()).validate(expectedRecords);
   }
 
+  @TestTemplate
+  public void testWorkerPool() throws Exception {
+    Table table = helper.createUnpartitionedTable();
+    UserGroupInformation user1 =
+        UserGroupInformation.createUserForTesting("user1", new String[] {});
+    UserGroupInformation user2 =
+        UserGroupInformation.createUserForTesting("user2", new String[] {});
+    final ExecutorService workerPool1 = ThreadPools.newWorkerPool("iceberg-plan-worker-pool", 1);
+    final ExecutorService workerPool2 = ThreadPools.newWorkerPool("iceberg-plan-worker-pool", 1);
+    try {
+      assertThat(getUserFromWorkerPool(user1, table, workerPool1)).isEqualTo("user1");
+      assertThat(getUserFromWorkerPool(user2, table, workerPool1)).isEqualTo("user1");
+      assertThat(getUserFromWorkerPool(user2, table, workerPool2)).isEqualTo("user2");
+    } finally {
+      workerPool1.shutdown();
+      workerPool2.shutdown();
+    }
+  }
+
+  private String getUserFromWorkerPool(
+      UserGroupInformation user, Table table, ExecutorService workerpool) throws Exception {
+    Method method =
+        IcebergInputFormat.class.getDeclaredMethod(
+            "planInputSplits", Table.class, Configuration.class, ExecutorService.class);
+    method.setAccessible(true);
+    return user.doAs(
+        (PrivilegedAction<String>)
+            () -> {
+              try {
+                method.invoke(new IcebergInputFormat<>(), table, conf, workerpool);
+                return workerpool
+                    .submit(() -> UserGroupInformation.getCurrentUser().getUserName())
+                    .get();
+              } catch (Exception e) {
+                throw new RuntimeException("Failed to get user from worker pool", e);
+              }
+            });
+  }
+
   // TODO - Capture template type T in toString method:
   // https://github.com/apache/iceberg/issues/1542
   public abstract static class TestInputFormat<T> {