Merge pull request #413 from uwescience/load_options

Adding optional args to load() function.

examples/load_opts.myl

@@ -0,0 +1,2 @@
+t = load("https://s3-us-west-2.amazonaws.com/myria/public-adhoc-TwitterK.csv", csv(schema(column0:int, column1:int), skip=1));
+store(t, TwitterK2);

examples/standalone.myl

@@ -1,5 +1,5 @@
-Emp = load("./examples/emp.csv", id:int, dept_id:int, name:string, salary:int);
-Dept = load("./examples/dept.csv", id:int, name:string, manager:int);
+Emp = load("./examples/emp.csv", csv(schema(id:int, dept_id:int, name:string, salary:int)));
+Dept = load("./examples/dept.csv", csv(schema(id:int, name:string, manager:int)));
 
 out = [from Emp, Dept
        where Emp.dept_id == Dept.id AND Emp.salary > 5000

raco/algebra.py

@@ -1266,15 +1266,17 @@ class FileScan(ZeroaryOperator):
 
     """Load table data from a file."""
 
-    def __init__(self, path=None, _scheme=None):
+    def __init__(self, path=None, _scheme=None, options={}):
         self.path = path
         self._scheme = _scheme
+        self.options = options
         ZeroaryOperator.__init__(self)
 
     def __eq__(self, other):
         return (ZeroaryOperator.__eq__(self, other)
                 and self.path == other.path
-                and self.scheme() == other.scheme())
+                and self.scheme() == other.scheme()
+                and self.options == other.options)
 
     def __hash__(self):
         return ("%s-%s" % (self.opname(), self.path)).__hash__()
@@ -1283,9 +1285,11 @@ def shortStr(self):
         return "%s(%s)" % (self.opname(), self.path)
 
     def __repr__(self):
-        return "{op}({path!r}, {sch!r})".format(op=self.opname(),
-                                                path=self.path,
-                                                sch=self._scheme)
+        return "{op}({path!r}, {sch!r}, {opt!r})".format(
+            op=self.opname(),
+            path=self.path,
+            sch=self._scheme,
+            opt=self.options)
 
     def num_tuples(self):
         raise NotImplementedError("{op}.num_tuples".format(op=type(self)))
@@ -1294,7 +1298,7 @@ def copy(self, other):
         """deep copy"""
         self.path = other.path
         self._scheme = other._scheme
-
+        self.options = other.options
         ZeroaryOperator.copy(self, other)
 
     def scheme(self):

raco/fakedb.py

@@ -354,6 +354,9 @@ def scantemp(self, op):
     def myriascan(self, op):
         return self.scan(op)
 
+    def myriafilescan(self, op):
+        return self.filescan(op)
+
     def myriasink(self, op):
         return self.sink(op)
 

raco/language/myrialang.py

@@ -174,6 +174,18 @@ def compileme(self):
         }
 
 
+class MyriaFileScan(algebra.FileScan, MyriaOperator):
+    def compileme(self):
+        return dict({
+            "opType": "FileScan",
+            "source": {
+                "dataType": "URI",
+                "uri": self.path,
+            },
+            "schema": scheme_to_schema(self.scheme())
+        }, **self.options)
+
+
 class MyriaLimit(algebra.Limit, MyriaOperator):
     def compileme(self, inputid):
         return {
@@ -1482,6 +1494,7 @@ def fire(self, expr):
     rules.OneToOne(algebra.NaryJoin, MyriaLeapFrogJoin),
     rules.OneToOne(algebra.Scan, MyriaScan),
     rules.OneToOne(algebra.ScanTemp, MyriaScanTemp),
+    rules.OneToOne(algebra.FileScan, MyriaFileScan),
     rules.OneToOne(algebra.SingletonRelation, MyriaSingleton),
     rules.OneToOne(algebra.EmptyRelation, MyriaEmptyRelation),
     rules.OneToOne(algebra.UnionAll, MyriaUnionAll),
@@ -1513,6 +1526,7 @@ class MyriaAlgebra(Algebra):
         MyriaHyperShuffleConsumer,
         MyriaScan,
         MyriaScanTemp,
+        MyriaFileScan,
         MyriaEmptyRelation,
         MyriaSingleton
     )
@@ -1766,7 +1780,6 @@ def compile_to_json(raw_query, logical_plan, physical_plan,
     # raw_query must be a string
     if not isinstance(raw_query, basestring):
         raise ValueError("raw query must be a string")
-
     return {"rawQuery": raw_query,
             "logicalRa": str(logical_plan),
             "language": language,

raco/myrial/interpreter.py

@@ -97,8 +97,8 @@ def scan(self, rel_key):
         return raco.algebra.Scan(rel_key, scheme,
                                  self.catalog.num_tuples(rel_key))
 
-    def load(self, path, scheme):
-        return raco.algebra.FileScan(path, scheme)
+    def load(self, path, scheme, options):
+        return raco.algebra.FileScan(path, scheme, options)
 
     def table(self, emit_clause):
         """Emit a single-row table literal."""

raco/myrial/parser.py

@@ -555,8 +555,9 @@ def p_expression_scan(p):
 
     @staticmethod
     def p_expression_load(p):
-        'expression :  LOAD LPAREN STRING_LITERAL COMMA column_def_list RPAREN'
-        p[0] = ('LOAD', p[3], scheme.Scheme(p[5]))
+        'expression : LOAD LPAREN STRING_LITERAL COMMA file_parser_fun RPAREN'
+        schema, options = p[5]
+        p[0] = ('LOAD', p[3], scheme.Scheme(schema), options)
 
     @staticmethod
     def p_relation_key(p):
@@ -581,6 +582,51 @@ def p_column_def(p):
         'column_def : unreserved_id COLON type_name'
         p[0] = (p[1], p[3])
 
+    @staticmethod
+    def p_schema_fun(p):
+        'schema_fun : SCHEMA LPAREN column_def_list RPAREN'
+        p[0] = p[3]
+
+    @staticmethod
+    def p_file_parser_fun(p):
+        """file_parser_fun : file_parser_type LPAREN \
+   schema_fun COMMA option_list RPAREN
+ | file_parser_type LPAREN schema_fun RPAREN"""
+        if len(p) == 7:
+            schema, options = (p[3], p[5])
+        else:
+            schema, options = (p[3], {})
+        p[0] = (schema, options)
+
+    @staticmethod
+    def p_file_parser_type(p):
+        'file_parser_type : CSV'
+        p[0] = p[1]
+
+    @staticmethod
+    def p_option_list(p):
+        """option_list : option_list COMMA option
+                           | option"""
+        if len(p) == 4:
+            opts = p[1] + [p[3]]
+        else:
+            opts = [p[1]]
+        p[0] = dict(opts)
+
+    @staticmethod
+    def p_option(p):
+        'option : unreserved_id EQUALS literal_arg'
+        p[0] = (p[1], p[3])
+
+    @staticmethod
+    def p_literal_arg(p):
+        """literal_arg : STRING_LITERAL
+                       | INTEGER_LITERAL
+                       | FLOAT_LITERAL
+                       | TRUE
+                       | FALSE"""
+        p[0] = p[1]
+
     @staticmethod
     def p_type_name(p):
         """type_name : STRING

raco/myrial/scanner.py

@@ -6,7 +6,8 @@
 import raco.myrial.exceptions
 
 keywords = ['WHILE', 'DO', 'DEF', 'APPLY', 'CASE', 'WHEN', 'THEN',
-            'ELSE', 'END', 'CONST', 'LOAD', 'DUMP', 'UDA', 'TRUE', 'FALSE']
+            'ELSE', 'END', 'CONST', 'LOAD', 'DUMP', 'CSV', 'SCHEMA',
+            'UDA', 'TRUE', 'FALSE']
 
 types = ['INT', 'STRING', 'FLOAT', 'BOOLEAN']
 

raco/test_style.py

@@ -21,7 +21,8 @@ class StyleTest(unittest.TestCase):
 
     def test_style(self):
         "run flake8 with the right arguments and ensure all files pass"
-        check_output_and_print_stderr(['flake8', '--ignore=F', 'raco'])
+        check_output_and_print_stderr([
+            'flake8', '--ignore=F', '--exclude=parsetab.py', 'raco'])
 
     def test_pylint(self):
         "run pylint -E to catch obvious errors"