fixing and renaming CSVFragmentTupleSource

uwescience · Jun 7, 2017 · 406be71 · 406be71
1 parent 625b0e7
commit 406be71
Show file tree

Hide file tree

Showing 5 changed files with 41 additions and 31 deletions.
diff --git a/src/edu/washington/escience/myria/CsvTupleReader.java b/src/edu/washington/escience/myria/CsvTupleReader.java
@@ -7,10 +7,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
-import java.nio.ByteBuffer;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.Iterator;
 
 import javax.annotation.Nullable;
@@ -27,7 +23,6 @@
 
 import edu.washington.escience.myria.storage.TupleBatch;
 import edu.washington.escience.myria.storage.TupleBatchBuffer;
-import edu.washington.escience.myria.storage.TupleUtils;
 import edu.washington.escience.myria.util.DateTimeUtils;
 
 /**
@@ -192,6 +187,22 @@ public Schema getSchema() {
     return schema;
   }
 
+  public char getDelimiter() {
+    return delimiter;
+  }
+
+  public char getQuote() {
+    return quote;
+  }
+
+  public char getEscape() {
+    return escape;
+  }
+
+  public Integer getSkip() {
+    return numberOfSkippedLines;
+  }
+
   @Override
   public void close() throws IOException {
     parser = null;

diff --git a/...encoding/CSVFileScanFragmentEncoding.java → ...oding/CSVFragmentTupleSourceEncoding.java b/...encoding/CSVFileScanFragmentEncoding.java → ...oding/CSVFragmentTupleSourceEncoding.java
@@ -5,22 +5,17 @@
 import edu.washington.escience.myria.CsvTupleReader;
 import edu.washington.escience.myria.api.encoding.QueryConstruct.ConstructArgs;
 import edu.washington.escience.myria.io.AmazonS3Source;
-import edu.washington.escience.myria.operator.CSVFileScanFragment;
+import edu.washington.escience.myria.operator.CSVFragmentTupleSource;
 
-public class CSVFileScanFragmentEncoding extends LeafOperatorEncoding<CSVFileScanFragment> {
+public class CSVFragmentTupleSourceEncoding extends LeafOperatorEncoding<CSVFragmentTupleSource> {
 
   @Required public CsvTupleReader reader;
   @Required public AmazonS3Source source;
 
-  public Character delimiter;
-  public Character quote;
-  public Character escape;
-  public Integer skip;
-
   public Set<Integer> workers;
 
   @Override
-  public CSVFileScanFragment construct(ConstructArgs args) {
+  public CSVFragmentTupleSource construct(ConstructArgs args) {
     /* Attempt to use all the workers if not specified */
     if (workers == null) {
       workers = args.getServer().getAliveWorkers();
@@ -30,7 +25,13 @@ public CSVFileScanFragment construct(ConstructArgs args) {
     int[] workersArray =
         args.getServer().parallelIngestComputeNumWorkers(source.getFileSize(), workers);
 
-    return new CSVFileScanFragment(
-        source, reader.getSchema(), workersArray, delimiter, quote, escape, skip);
+    return new CSVFragmentTupleSource(
+        source,
+        reader.getSchema(),
+        workersArray,
+        reader.getDelimiter(),
+        reader.getQuote(),
+        reader.getEscape(),
+        reader.getSkip());
   }
 }
diff --git a/src/edu/washington/escience/myria/api/encoding/OperatorEncoding.java b/src/edu/washington/escience/myria/api/encoding/OperatorEncoding.java
@@ -10,6 +10,7 @@
 
 import edu.washington.escience.myria.api.MyriaApiException;
 import edu.washington.escience.myria.api.encoding.QueryConstruct.ConstructArgs;
+import edu.washington.escience.myria.operator.CSVFragmentTupleSource;
 import edu.washington.escience.myria.operator.Operator;
 
 /**
@@ -29,7 +30,7 @@
   @Type(name = "Consumer", value = ConsumerEncoding.class),
   @Type(name = "Counter", value = CounterEncoding.class),
   @Type(name = "CrossWithSingleton", value = CrossWithSingletonEncoding.class),
-  @Type(name = "CSVFileScanFragment", value = CSVFileScanFragmentEncoding.class),
+  @Type(name = "CSVFileScanFragment", value = CSVFragmentTupleSource.class),
   @Type(name = "DbInsert", value = DbInsertEncoding.class),
   @Type(name = "DbQueryScan", value = QueryScanEncoding.class),
   @Type(name = "DbCreateIndex", value = CreateIndexEncoding.class),

diff --git a/...e/myria/operator/CSVFileScanFragment.java → ...yria/operator/CSVFragmentTupleSource.java b/...e/myria/operator/CSVFileScanFragment.java → ...yria/operator/CSVFragmentTupleSource.java
@@ -7,15 +7,13 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
-import java.nio.ByteBuffer;
 import java.util.Iterator;
 
 import javax.annotation.Nullable;
 
 import org.apache.commons.csv.CSVFormat;
 import org.apache.commons.csv.CSVParser;
 import org.apache.commons.csv.CSVRecord;
-import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.BooleanUtils;
 
 import com.google.common.base.MoreObjects;
@@ -31,13 +29,12 @@
 import edu.washington.escience.myria.io.FileSource;
 import edu.washington.escience.myria.storage.TupleBatch;
 import edu.washington.escience.myria.storage.TupleBatchBuffer;
-import edu.washington.escience.myria.storage.TupleUtils;
 import edu.washington.escience.myria.util.DateTimeUtils;
 
 /**
  *
  */
-public class CSVFileScanFragment extends LeafOperator {
+public class CSVFragmentTupleSource extends LeafOperator {
 
   /** The Schema of the relation stored in this file. */
   private final Schema schema;
@@ -85,9 +82,9 @@ public class CSVFileScanFragment extends LeafOperator {
    * The logger for debug, trace, etc. messages in this class.
    */
   private static final org.slf4j.Logger LOGGER =
-      org.slf4j.LoggerFactory.getLogger(CSVFileScanFragment.class);
+      org.slf4j.LoggerFactory.getLogger(CSVFragmentTupleSource.class);
 
-  public CSVFileScanFragment(
+  public CSVFragmentTupleSource(
       final String filename,
       final Schema schema,
       final long startByteRange,
@@ -96,7 +93,7 @@ public CSVFileScanFragment(
     this(filename, schema, startByteRange, endByteRange, isLastWorker, null, null, null, null);
   }
 
-  public CSVFileScanFragment(
+  public CSVFragmentTupleSource(
       final DataSource source,
       final Schema schema,
       final long startByteRange,
@@ -105,7 +102,7 @@ public CSVFileScanFragment(
     this(source, schema, startByteRange, endByteRange, isLastWorker, null, null, null, null);
   }
 
-  public CSVFileScanFragment(
+  public CSVFragmentTupleSource(
       final String filename,
       final Schema schema,
       final long startByteRange,
@@ -124,7 +121,7 @@ public CSVFileScanFragment(
         null);
   }
 
-  public CSVFileScanFragment(
+  public CSVFragmentTupleSource(
       final DataSource source,
       final Schema schema,
       final long startByteRange,
@@ -134,7 +131,7 @@ public CSVFileScanFragment(
     this(source, schema, startByteRange, endByteRange, isLastWorker, delimiter, null, null, null);
   }
 
-  public CSVFileScanFragment(
+  public CSVFragmentTupleSource(
       final String filename,
       final Schema schema,
       final long startByteRange,
@@ -156,7 +153,7 @@ public CSVFileScanFragment(
         numberOfSkippedLines);
   }
 
-  public CSVFileScanFragment(
+  public CSVFragmentTupleSource(
       final DataSource source,
       final Schema schema,
       final long partitionStartByteRange,
@@ -185,7 +182,7 @@ public CSVFileScanFragment(
     flagAsRangeSelected = true;
   }
 
-  public CSVFileScanFragment(
+  public CSVFragmentTupleSource(
       final AmazonS3Source source,
       final Schema schema,
       final int[] workerIds,

diff --git a/src/edu/washington/escience/myria/parallel/Server.java b/src/edu/washington/escience/myria/parallel/Server.java
@@ -95,7 +95,7 @@
 import edu.washington.escience.myria.io.DataSink;
 import edu.washington.escience.myria.io.UriSink;
 import edu.washington.escience.myria.operator.Apply;
-import edu.washington.escience.myria.operator.CSVFileScanFragment;
+import edu.washington.escience.myria.operator.CSVFragmentTupleSource;
 import edu.washington.escience.myria.operator.DbCreateFunction;
 import edu.washington.escience.myria.operator.DbCreateIndex;
 import edu.washington.escience.myria.operator.DbCreateView;
@@ -899,8 +899,8 @@ public DatasetStatus parallelIngestDataset(
 
     Map<Integer, SubQueryPlan> workerPlans = new HashMap<>();
     for (int workerID = 1; workerID <= workersArray.length; workerID++) {
-      CSVFileScanFragment scanFragment =
-          new CSVFileScanFragment(
+      CSVFragmentTupleSource scanFragment =
+          new CSVFragmentTupleSource(
               s3Source, schema, workersArray, delimiter, quote, escape, numberOfSkippedLines);
       workerPlans.put(
           workersArray[workerID - 1],