MR: Add InputFormat (#843)

Author: rdsr

.gitignore

@@ -35,3 +35,6 @@ spark/tmp/
 .project
 .settings
 bin/
+
+# Hive/metastore files
+metastore_db/

build.gradle

@@ -220,6 +220,24 @@ project(':iceberg-hive') {
   }
 }
 
+project(':iceberg-mr') {
+  dependencies {
+    compile project(':iceberg-api')
+    compile project(':iceberg-core')
+    compile project(':iceberg-orc')
+    compile project(':iceberg-parquet')
+    compile project(':iceberg-data')
+
+    compileOnly("org.apache.hadoop:hadoop-client") {
+      exclude group: 'org.apache.avro', module: 'avro'
+    }
+
+    testCompile project(path: ':iceberg-data', configuration: 'testArtifacts')
+    testCompile project(path: ':iceberg-api', configuration: 'testArtifacts')
+    testCompile project(path: ':iceberg-core', configuration: 'testArtifacts')
+  }
+}
+
 project(':iceberg-orc') {
   dependencies {
     compile project(':iceberg-api')

core/src/main/java/org/apache/iceberg/hadoop/Util.java

@@ -19,13 +19,23 @@
 
 package org.apache.iceberg.hadoop;
 
+import com.google.common.collect.Sets;
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.Set;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Util {
+  private static final Logger LOG = LoggerFactory.getLogger(Util.class);
 
-class Util {
   private Util() {
   }
 
@@ -36,4 +46,21 @@ public static FileSystem getFs(Path path, Configuration conf) {
       throw new RuntimeIOException(e, "Failed to get file system for path: %s", path);
     }
   }
+
+  public static String[] blockLocations(CombinedScanTask task, Configuration conf) {
+    Set<String> locationSets = Sets.newHashSet();
+    for (FileScanTask f : task.files()) {
+      Path path = new Path(f.file().path().toString());
+      try {
+        FileSystem fs = path.getFileSystem(conf);
+        for (BlockLocation b : fs.getFileBlockLocations(path, f.start(), f.length())) {
+          locationSets.addAll(Arrays.asList(b.getHosts()));
+        }
+      } catch (IOException ioe) {
+        LOG.warn("Failed to get block locations for path {}", path, ioe);
+      }
+    }
+
+    return locationSets.toArray(new String[0]);
+  }
 }

mr/src/main/java/org/apache/iceberg/mr/SerializationUtil.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.mr;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+
+public class SerializationUtil {
+
+  private SerializationUtil() {
+  }
+
+  public static byte[] serializeToBytes(Object obj) {
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+      oos.writeObject(obj);
+      return baos.toByteArray();
+    } catch (IOException e) {
+      throw new RuntimeIOException("Failed to serialize object", e);
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  public static <T> T deserializeFromBytes(byte[] bytes) {
+    if (bytes == null) {
+      return null;
+    }
+
+    try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+        ObjectInputStream ois = new ObjectInputStream(bais)) {
+      return (T) ois.readObject();
+    } catch (IOException e) {
+      throw new RuntimeIOException("Failed to deserialize object", e);
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException("Could not read object ", e);
+    }
+  }
+
+  public static String serializeToBase64(Object obj) {
+    byte[] bytes = serializeToBytes(obj);
+    return new String(Base64.getMimeEncoder().encode(bytes), StandardCharsets.UTF_8);
+  }
+
+  public static <T> T deserializeFromBase64(String base64) {
+    if (base64 == null) {
+      return null;
+    }
+    byte[] bytes = Base64.getMimeDecoder().decode(base64.getBytes(StandardCharsets.UTF_8));
+    return deserializeFromBytes(bytes);
+  }
+}