[iceberg] branch master updated: MR: Use HadoopFileIO instead of HadoopInputFile (#1526)

[bl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/master by this push:
     new ab0b5a6  MR: Use HadoopFileIO instead of HadoopInputFile (#1526)
ab0b5a6 is described below

commit ab0b5a6b7fd37e42306bf398775b09119b27b92e
Author: Holden Karau <ho...@pigscanfly.ca>
AuthorDate: Mon Sep 28 17:18:05 2020 -0700

    MR: Use HadoopFileIO instead of HadoopInputFile (#1526)
---
 .../java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java  | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

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 ad49205..82d2283 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
@@ -53,9 +53,10 @@ import org.apache.iceberg.data.parquet.GenericParquetReaders;
 import org.apache.iceberg.expressions.Evaluator;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.hadoop.HadoopInputFile;
+import org.apache.iceberg.hadoop.HadoopFileIO;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.mr.Catalogs;
 import org.apache.iceberg.mr.InputFormatConfig;
@@ -165,6 +166,7 @@ public class IcebergInputFormat<T> extends InputFormat<Void, T> {
     private Iterator<FileScanTask> tasks;
     private T currentRow;
     private CloseableIterator<T> currentIterator;
+    private FileIO fileIO;
 
     @Override
     public void initialize(InputSplit split, TaskAttemptContext newContext) {
@@ -172,6 +174,7 @@ public class IcebergInputFormat<T> extends InputFormat<Void, T> {
       // For now IcebergInputFormat does its own split planning and does not accept FileSplit instances
       CombinedScanTask task = ((IcebergSplit) split).task();
       this.context = newContext;
+      this.fileIO = new HadoopFileIO(this.context.getConfiguration());
       this.tasks = task.files().iterator();
       this.tableSchema = SchemaParser.fromJson(conf.get(InputFormatConfig.TABLE_SCHEMA));
       String readSchemaStr = conf.get(InputFormatConfig.READ_SCHEMA);
@@ -227,8 +230,7 @@ public class IcebergInputFormat<T> extends InputFormat<Void, T> {
 
     private CloseableIterable<T> open(FileScanTask currentTask, Schema readSchema) {
       DataFile file = currentTask.file();
-      // TODO we should make use of FileIO to create inputFile
-      InputFile inputFile = HadoopInputFile.fromLocation(file.path(), context.getConfiguration());
+      InputFile inputFile = fileIO.newInputFile(file.path().toString());
       CloseableIterable<T> iterable;
       switch (file.format()) {
         case AVRO: