working on example

Author: JohnMount

Examples/WindowFunctions/Arrow.ipynb

@@ -37,6 +37,8 @@
     "\n",
     "table_description = describe_table(d)\n",
     "\n",
+    "d['irrelevant_column'] = 1\n",
+    "\n",
     "id_ops_a = table_description. \\\n",
     "    project(group_by=['g']). \\\n",
     "    extend({\n",
@@ -57,7 +59,7 @@
     {
      "name": "stdout",
      "text": [
-      "[{'x', 'v', 'g'} -> ['g', 'x', 'v', 'ngroup']]\n"
+      "[{'g', 'v', 'x'} -> ['g', 'x', 'v', 'ngroup']]\n"
      ],
      "output_type": "stream"
     }
@@ -112,7 +114,7 @@
     {
      "name": "stdout",
      "text": [
-      "[{'v', 'x', 'g', 'ngroup'} -> ['g', 'x', 'v', 'ngroup', 'row_number', 'shift_v']]\n"
+      "[{'g', 'v', 'ngroup', 'x'} -> ['g', 'x', 'v', 'ngroup', 'row_number', 'shift_v']]\n"
      ],
      "output_type": "stream"
     }
@@ -172,7 +174,7 @@
     {
      "name": "stdout",
      "text": [
-      "[{'shift_v', 'g', 'v', 'row_number', 'ngroup', 'x'} -> ['g', 'x', 'v', 'ngroup', 'row_number', 'shift_v', 'size', 'max_v', 'min_v', 'sum_v', 'mean_v', 'count_v', 'size_v']]\n"
+      "[{'g', 'x', 'row_number', 'v', 'shift_v', 'ngroup'} -> ['g', 'x', 'v', 'ngroup', 'row_number', 'shift_v', 'size', 'max_v', 'min_v', 'sum_v', 'mean_v', 'count_v', 'size_v']]\n"
      ],
      "output_type": "stream"
     }

data_algebra/arrow.py

@@ -32,21 +32,35 @@ def _r_copy_replace(self, ops):
     def transform(self, other):
         """replace self input table with other"""
         if isinstance(other, pandas.DataFrame):
+            cols = set(other.columns)
+            missing = set(self.incoming_columns) - cols
+            if len(missing) > 0:
+                raise ValueError("missing required columns: " + str(missing))
+            if len(cols - set(self.incoming_columns)):
+                other = other[self.incoming_columns]
             return self.v.transform(other)
         if isinstance(other, data_algebra.data_ops.ViewRepresentation):
             other = DataOpArrow(other)
         if not isinstance(other, DataOpArrow):
             raise TypeError("other must be a DataOpArrow")
+        missing = set(self.incoming_columns) - set(other.outgoing_columns)
+        if len(missing) > 0:
+            raise ValueError("missing required columns: " + str(missing))
+        if len(set(other.outgoing_columns) - set(self.incoming_columns)):
+            # extra columns, in a strict categorical formulation we would
+            # reject this. instead insert a select columns node to get the match
+            other = DataOpArrow(other.v.select_columns([c for c in self.incoming_columns]))
+        # check categorical arrow composition conditions
         if set(self.incoming_columns) != set(other.outgoing_columns):
-            raise TypeError("arrow composition conditions not met (incoming columsn don't match outgoing)")
+            raise ValueError("arrow composition conditions not met (incoming column set doesn't match outgoing)")
         return DataOpArrow(other._r_copy_replace(self.v))
 
-    def __rrshift__(self, other):  # override other >> self
-        return self.transform(other)
-
     def __rshift__(self, other):  # override self >> other
         return other.transform(self)
 
+    def __rrshift__(self, other):  # override other >> self
+        return self.transform(other)
+
     def __repr__(self):
         return "DataOpArrow(" + self.v.__repr__() + ")"