From bccf2d8a20ec920f1a8451637e34318776737d7a Mon Sep 17 00:00:00 2001
From: dan mcweeney <mcweeney@adobe.com>
Date: Thu, 3 Aug 2017 16:33:14 -0400
Subject: [PATCH] Fixes #24 - Teach Table Generator how to write to a HDFS
 file.  * Make session & scaling serializable so it can be passed around  *
 Teach session factory pattern to take in a hadoop config object  * Make a
 serializable hadoop config object to ensure Session is    serializable

---
 pom.xml                                       |  13 ++
 src/main/java/com/teradata/tpcds/Scaling.java |   3 +-
 src/main/java/com/teradata/tpcds/Session.java | 116 ++++++++++++++++--
 .../com/teradata/tpcds/TableGenerator.java    |  43 +++++--
 4 files changed, 151 insertions(+), 24 deletions(-)

diff --git a/pom.xml b/pom.xml
index 894e8ab..24add69 100644
--- a/pom.xml
+++ b/pom.xml
@@ -237,6 +237,19 @@
             <groupId>javax.inject</groupId>
             <artifactId>javax.inject</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>2.7.1</version>
+            <scope>provided</scope>
+            <optional>true</optional>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.code.findbugs</groupId>
+                    <artifactId>jsr305</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 
     <profiles>
diff --git a/src/main/java/com/teradata/tpcds/Scaling.java b/src/main/java/com/teradata/tpcds/Scaling.java
index 57933ab..ac45542 100644
--- a/src/main/java/com/teradata/tpcds/Scaling.java
+++ b/src/main/java/com/teradata/tpcds/Scaling.java
@@ -17,6 +17,7 @@
 import com.teradata.tpcds.distribution.CalendarDistribution;
 import com.teradata.tpcds.type.Date;
 
+import java.io.Serializable;
 import java.util.EnumMap;
 import java.util.Map;
 
@@ -39,7 +40,7 @@
 import static com.teradata.tpcds.type.Date.fromJulianDays;
 import static com.teradata.tpcds.type.Date.isLeapYear;
 
-public class Scaling
+public class Scaling implements Serializable
 {
     private final double scale;
     private final Map<Table, Long> tableToRowCountMap = new EnumMap<>(Table.class);
diff --git a/src/main/java/com/teradata/tpcds/Session.java b/src/main/java/com/teradata/tpcds/Session.java
index a1477dd..5bbbc9b 100644
--- a/src/main/java/com/teradata/tpcds/Session.java
+++ b/src/main/java/com/teradata/tpcds/Session.java
@@ -14,6 +14,10 @@
 
 package com.teradata.tpcds;
 
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.io.Serializable;
 import java.util.Optional;
 
 import static com.teradata.tpcds.Options.DEFAULT_DIRECTORY;
@@ -26,12 +30,12 @@
 import static com.teradata.tpcds.Options.DEFAULT_SEPARATOR;
 import static com.teradata.tpcds.Options.DEFAULT_SUFFIX;
 
-public class Session
+public class Session implements Serializable
 {
     private final Scaling scaling;
     private final String targetDirectory;
     private final String suffix;
-    private final Optional<Table> table;
+    private final Table table;
     private final String nullString;
     private final char separator;
     private final boolean doNotTerminate;
@@ -39,6 +43,7 @@ public class Session
     private final int parallelism;
     private final int chunkNumber;
     private final boolean overwrite;
+    private final SerializableHadoopConfiguration hadoopConfig;
 
     public Session(double scale, String targetDirectory, String suffix, Optional<Table> table, String nullString, char separator, boolean doNotTerminate, boolean noSexism, int parallelism, boolean overwrite)
     {
@@ -46,11 +51,19 @@ public Session(double scale, String targetDirectory, String suffix, Optional<Tab
     }
 
     public Session(double scale, String targetDirectory, String suffix, Optional<Table> table, String nullString, char separator, boolean doNotTerminate, boolean noSexism, int parallelism, int chunkNumber, boolean overwrite)
+    {
+        this(scale, targetDirectory, suffix, table, nullString, separator, doNotTerminate, noSexism, parallelism,
+                chunkNumber, overwrite, null);
+    }
+
+    public Session(double scale, String targetDirectory, String suffix, Optional<Table> table, String nullString,
+                   char separator, boolean doNotTerminate, boolean noSexism, int parallelism, int chunkNumber,
+                   boolean overwrite, Configuration hadoopConfig)
     {
         this.scaling = new Scaling(scale);
         this.targetDirectory = targetDirectory;
         this.suffix = suffix;
-        this.table = table;
+        this.table = table.isPresent() ? table.get() : null;
         this.nullString = nullString;
         this.separator = separator;
         this.doNotTerminate = doNotTerminate;
@@ -58,6 +71,7 @@ public Session(double scale, String targetDirectory, String suffix, Optional<Tab
         this.parallelism = parallelism;
         this.chunkNumber = chunkNumber;
         this.overwrite = overwrite;
+        this.hadoopConfig = new SerializableHadoopConfiguration(hadoopConfig);
     }
 
     public static Session getDefaultSession()
@@ -71,7 +85,7 @@ public Session withTable(Table table)
                 this.scaling.getScale(),
                 this.targetDirectory,
                 this.suffix,
-                Optional.of(table),
+                Optional.ofNullable(table),
                 this.nullString,
                 this.separator,
                 this.doNotTerminate,
@@ -88,7 +102,7 @@ public Session withScale(double scale)
                 scale,
                 this.targetDirectory,
                 this.suffix,
-                this.table,
+                Optional.ofNullable(this.table),
                 this.nullString,
                 this.separator,
                 this.doNotTerminate,
@@ -105,7 +119,7 @@ public Session withParallelism(int parallelism)
                 this.scaling.getScale(),
                 this.targetDirectory,
                 this.suffix,
-                this.table,
+                Optional.ofNullable(this.table),
                 this.nullString,
                 this.separator,
                 this.doNotTerminate,
@@ -122,7 +136,7 @@ public Session withChunkNumber(int chunkNumber)
                 this.scaling.getScale(),
                 this.targetDirectory,
                 this.suffix,
-                this.table,
+                Optional.ofNullable(this.table),
                 this.nullString,
                 this.separator,
                 this.doNotTerminate,
@@ -139,7 +153,7 @@ public Session withNoSexism(boolean noSexism)
                 this.scaling.getScale(),
                 this.targetDirectory,
                 this.suffix,
-                this.table,
+                Optional.ofNullable(this.table),
                 this.nullString,
                 this.separator,
                 this.doNotTerminate,
@@ -150,6 +164,42 @@ public Session withNoSexism(boolean noSexism)
         );
     }
 
+    public Session withHadoop(Configuration hadoopConfig)
+    {
+        return new Session(
+                this.scaling.getScale(),
+                this.targetDirectory,
+                this.suffix,
+                Optional.ofNullable(this.table),
+                this.nullString,
+                this.separator,
+                this.doNotTerminate,
+                noSexism,
+                this.parallelism,
+                this.chunkNumber,
+                this.overwrite,
+                hadoopConfig
+        );
+    }
+
+    public Session withTargetDirectory(String targetDirectory)
+    {
+        return new Session(
+                this.scaling.getScale(),
+                targetDirectory,
+                this.suffix,
+                Optional.ofNullable(this.table),
+                this.nullString,
+                this.separator,
+                this.doNotTerminate,
+                this.noSexism,
+                this.parallelism,
+                this.chunkNumber,
+                this.overwrite,
+                this.hadoopConfig.get()
+        );
+    }
+
     public Scaling getScaling()
     {
         return scaling;
@@ -167,15 +217,15 @@ public String getSuffix()
 
     public boolean generateOnlyOneTable()
     {
-        return table.isPresent();
+        return Optional.ofNullable(table).isPresent();
     }
 
     public Table getOnlyTableToGenerate()
     {
-        if (!table.isPresent()) {
+        if (!Optional.ofNullable(table).isPresent()) {
             throw new TpcdsException("table not present");
         }
-        return table.get();
+        return Optional.ofNullable(table).get();
     }
 
     public String getNullString()
@@ -213,6 +263,11 @@ public boolean shouldOverwrite()
         return overwrite;
     }
 
+    public Configuration getHadoopConfig()
+    {
+        return this.hadoopConfig.get();
+    }
+
     public String getCommandLineArguments()
     {
         StringBuilder output = new StringBuilder();
@@ -225,8 +280,8 @@ public String getCommandLineArguments()
         if (!suffix.equals(DEFAULT_SUFFIX)) {
             output.append("--suffix ").append(suffix).append(" ");
         }
-        if (table.isPresent()) {
-            output.append("--table ").append(table.get().getName()).append(" ");
+        if (Optional.ofNullable(table).isPresent()) {
+            output.append("--table ").append(Optional.ofNullable(table).get().getName()).append(" ");
         }
         if (!nullString.equals(DEFAULT_NULL_STRING)) {
             output.append("--null ").append(nullString).append(" ");
@@ -255,3 +310,38 @@ public String getCommandLineArguments()
         return output.toString();
     }
 }
+
+class SerializableHadoopConfiguration implements Serializable
+{
+    Configuration conf;
+
+    public SerializableHadoopConfiguration(Configuration hadoopConf)
+    {
+        this.conf = hadoopConf;
+
+        if (this.conf == null) {
+            this.conf = new Configuration();
+        }
+    }
+
+    public SerializableHadoopConfiguration()
+    {
+        this.conf = new Configuration();
+    }
+
+    public Configuration get()
+    {
+        return this.conf;
+    }
+
+    private void writeObject(java.io.ObjectOutputStream out) throws IOException
+    {
+        this.conf.write(out);
+    }
+
+    private void readObject(java.io.ObjectInputStream in) throws IOException
+    {
+        this.conf = new Configuration();
+        this.conf.readFields(in);
+    }
+}
diff --git a/src/main/java/com/teradata/tpcds/TableGenerator.java b/src/main/java/com/teradata/tpcds/TableGenerator.java
index d42971e..c8f4886 100644
--- a/src/main/java/com/teradata/tpcds/TableGenerator.java
+++ b/src/main/java/com/teradata/tpcds/TableGenerator.java
@@ -14,11 +14,16 @@
 
 package com.teradata.tpcds;
 
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.io.Writer;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -65,19 +70,37 @@ private OutputStreamWriter addFileWriterForTable(Table table)
             throws IOException
     {
         String path = getPath(table);
-        File file = new File(path);
-        boolean newFileCreated = file.createNewFile();
-        if (!newFileCreated) {
-            if (session.shouldOverwrite()) {
-                // truncate the file
-                new FileOutputStream(path).close();
-            }
-            else {
-                throw new TpcdsException(format("File %s exists.  Remove it or run with the '--overwrite' option", path));
+        if (session.getHadoopConfig() == null) {
+            File file = new File(path);
+            boolean newFileCreated = file.createNewFile();
+            if (!newFileCreated) {
+                if (session.shouldOverwrite()) {
+                    // truncate the file
+                    new FileOutputStream(path).close();
+                }
+                else {
+                    throw new TpcdsException(format("File %s exists.  Remove it or run with the '--overwrite' option", path));
+                }
             }
+            return new OutputStreamWriter(new FileOutputStream(path, true), StandardCharsets.ISO_8859_1);
         }
+        try {
+            FileSystem fs = FileSystem.get(new URI(session.getTargetDirectory()), session.getHadoopConfig());
 
-        return new OutputStreamWriter(new FileOutputStream(path, true), StandardCharsets.ISO_8859_1);
+            Path hadoopPath = new Path(path);
+            if (fs.exists(hadoopPath)) {
+                if (session.shouldOverwrite()) {
+                    fs.delete(hadoopPath, true);
+                }
+                else {
+                    throw new TpcdsException(format("File %s exists.  Remove it or run with the '--overwrite' option", path));
+                }
+            }
+            return new OutputStreamWriter(fs.create(hadoopPath, true, 4096), StandardCharsets.ISO_8859_1);
+        }
+        catch (final URISyntaxException uriEx) {
+            throw new TpcdsException("Could not create URI for connecting to Hadoop:" + uriEx.getMessage());
+        }
     }
 
     private String getPath(Table table)