rollback when operation was wrong

Author: hkulyc

src/main/clojure/clojask/classes/ColInfo.clj

@@ -20,7 +20,11 @@
   (getFormatter [])
   (delCol [col-to-del])
   (setColInfo [new-col-set])
-  (renameColInfo [old-col new-col]))
+  (renameColInfo [old-col new-col])
+  (copy [] "copy all the information for rollback purpose")
+  (rollback [] "undo the change making use of the copied")
+  (commit [])
+  )
 
 
 (deftype ColInfo
@@ -31,7 +35,8 @@
           ^:unsynchronized-mutable col-dsp
           ^:unsynchronized-mutable col-type
           ^:unsynchronized-mutable col-format
-          ^:unsynchronized-mutable col-deleted]
+          ^:unsynchronized-mutable col-deleted
+          ^:unsynchronized-mutable hist]
 
   ;; method
   ColIntf
@@ -44,17 +49,48 @@
     (set! col-dsp (zipmap (take (count colNames) (iterate inc 0)) (map vector (map vector (iterate inc 0)))))
     (set! col-deleted (set nil)))
 
+  (getFormatter
+    [this]
+    col-format)
+
+  (getDesc
+    [this]
+    col-dsp)
+
+  (getType
+    [this]
+    col-type)
+
+  (getKeys
+    [this]
+    (mapv (fn [index] (get index-key index))
+          (take (count index-key) (iterate inc 0))))
+
+  (getKeyIndex
+    [this]
+    key-index)
+
+  (getIndexKey
+    [this]
+    index-key)
+
+  (getDeletedCol
+    [this]
+    col-deleted)
+
   (operate
     [this operation col]
+    (.copy this)
     (if (contains? key-index col)
       (do
         (set! col-dsp (assoc col-dsp (get key-index col) (conj (get col-dsp (get key-index col)) operation)))
           ;; "success"
-          nil)
-          (throw (OperationException. "Column name passed to operate not found"))))
+        nil)
+      (throw (OperationException. "Column name passed to operate not found"))))
 
   (operate
     [this operation col newCol]
+    (.copy this)
     (let [col (if (coll? col)
                 col
                 [col])
@@ -70,11 +106,11 @@
             ;; "success"
             nil))
         (do
-          (throw (OperationException. (str external " are not original column names")))
-          ))))
+          (throw (OperationException. (str external " are not original column names")))))))
 
   (setType
     [this operation col]
+    (.copy this)
     (if (.contains col-keys col)
       ;; if this column has been assigned a type
       (do
@@ -86,44 +122,18 @@
 
   (setFormatter
     [this format col]
+    (.copy this)
     (set! col-format (assoc col-format (get key-index col) format)))
 
-  (getFormatter
-    [this]
-    col-format)
-
-  (getDesc
-    [this]
-    col-dsp)
-
-  (getType
-    [this]
-    col-type)
-
-  (getKeys
-    [this]
-    (mapv (fn [index] (get index-key index)) 
-          (take (count index-key) (iterate inc 0))))
-
-  (getKeyIndex
-   [this]
-   key-index)
-
-  (getIndexKey
-   [this]
-   index-key)
-
-  (getDeletedCol
-    [this]
-    col-deleted)
-
   (delCol
     [this col-to-delete]
+    (.copy this)
     (let [col-indices (set (map key-index col-to-delete))]
       (set! col-deleted (set/union col-deleted col-indices))))
 
   (setColInfo
     [this new-col-set]
+    (.copy this)
     (let [original-key-index (.getKeyIndex this)
           new-col-dsp-vals (vals (select-keys original-key-index new-col-set))
           original-type (.getType this)
@@ -133,13 +143,39 @@
       (set! index-key (zipmap (iterate inc 0) new-col-set))
       (set! col-dsp (zipmap (take (count col-keys) (iterate inc 0)) (map vector (map vector new-col-dsp-vals))))
       (if (not (empty? (.getType this)))
-          (set! col-type (zipmap (map #(first (first (get col-dsp (first %)))) original-type) (map last original-type))))
+        (set! col-type (zipmap (map #(first (first (get col-dsp (first %)))) original-type) (map last original-type))))
       (if (not (empty? (.getFormatter this)))
-          (set! col-format (zipmap (map #(first (first (get col-dsp (first %)))) original-format) (map last original-format))))))
-  
+        (set! col-format (zipmap (map #(first (first (get col-dsp (first %)))) original-format) (map last original-format))))))
+
   (renameColInfo
     [this old-col new-col]
+    (.copy this)
     (set! col-keys (mapv (fn [_] (if (= _ old-col) new-col _)) col-keys))
     (let [index (get key-index old-col)]
-     (set! key-index (set/rename-keys key-index {old-col new-col}))
-     (set! index-key (update index-key index (fn [_] new-col))))))
+      (set! key-index (set/rename-keys key-index {old-col new-col}))
+      (set! index-key (update index-key index (fn [_] new-col)))))
+
+  (copy
+    [this]
+    (set! hist {:col-keys col-keys
+                :key-index key-index
+                :index-key index-key
+                :col-dsp col-dsp
+                :col-type col-type
+                :col-format col-format
+                :col-deleted col-deleted}))
+
+  (rollback
+   [this]
+   (if (not= hist {})
+     (do (set! col-keys (:col-keys hist))  ;; contains only the original keys
+         (set! key-index (:key-index hist))
+         (set! index-key (:index-key hist))
+         (set! col-type (:col-type hist))
+         (set! col-format (:col-format hist))
+         (set! col-dsp (:col-dsp hist))
+         (set! col-deleted (:col-deleted hist)))))
+  
+  (commit
+   [this]
+   (set! hist {})))

src/main/clojure/clojask/classes/DataFrame.clj

@@ -61,7 +61,9 @@
   (sort [a b] "sort the dataframe based on columns")
   (addFormatter [a b] "format the column as the last step of the computation")
   (previewDF [] "preview function used for error predetection")
-  (errorPredetect [msg] "prints exception with msg if error is detected in preview"))
+  (errorPredetect [msg] "prints exception with msg if error is detected in preview")
+  (rollback [] "rollback the history of ColInfo and RowInfo")
+  (commit [] "commit the change"))
 
 ;; each dataframe can have a delayed object
 (defrecord DataFrame
@@ -82,7 +84,7 @@
             path1 (.getAbsolutePath (io/file path1))
             path2 (.getAbsolutePath (io/file path))]
         (if (= path1 path2)
-              (throw (OperationException. "Output path should be different from input path of dataframe argument."))))
+          (throw (OperationException. "Output path should be different from input path of dataframe argument."))))
       (catch Exception e nil)))
 
   (getOutput
@@ -190,7 +192,7 @@
                  cols
                  (vector cols))
           indices (map (fn [_] (get (.getKeyIndex (:col-info this)) _)) cols)]
-      (cond (not (u/are-in cols this))
+      (cond (not= [] (u/are-in cols this))
             (throw (TypeException. "Input includes non-existent column name(s).")))
       (if (nil? (.filter row-info indices predicate))
         this
@@ -406,6 +408,18 @@
         (.previewDF this)
         (catch Exception e
           (do
-            (throw (OperationException. (format  (str msg " (original error: %s)") (str (.getMessage e))))))))))
+            (.rollback this)
+            (throw (OperationException. (format  (str msg " (original error: %s)") (str (.getMessage e)))))))))
+    (.commit this))
+  
+  (rollback
+    [this]
+    (.rollback col-info)
+    (.rollback row-info))
+
+  (commit
+    [this]
+    (.commit col-info)
+    (.commit row-info))
 
   Object)

src/main/clojure/clojask/classes/RowInfo.clj

@@ -15,71 +15,103 @@
   (groupby [a])
   (aggregate [func old-key new-key])
   (setRowInfo [new-col-desc new-col-set])
-  (renameRowInfo [new-col-names]))
+  (renameRowInfo [new-col-names])
+  (copy [])
+  (rollback [])
+  (commit []))
 
 (deftype RowInfo
          [^:unsynchronized-mutable filters
           ^:unsynchronized-mutable groupby-key
           ^:unsynchronized-mutable aggre-func
           ;; ^:unsynchronized-mutable aggre-old-key
-          ^:unsynchronized-mutable aggre-new-key]
+          ^:unsynchronized-mutable aggre-new-key
+          ^:unsynchronized-mutable hist]
   RowIntf
   (getFilters
-    [self]
+    [this]
     filters)
+
+  (getGroupbyKeys
+    [this]
+    groupby-key)
+
+  ;; (getAggreOldKeys
+  ;;  [this]
+  ;;  aggre-old-key)
+  (getAggreNewKeys
+    [this]
+    aggre-new-key)
+
+  (getAggreFunc
+    [this]
+    aggre-func)
+
   (filter
-   [self cols predicate]
-   (set! filters (conj filters [predicate cols]))
+    [this cols predicate]
+    (.copy this)
+    (set! filters (conj filters [predicate cols]))
    ;; "success"
-   nil)
+    nil)
+
   (groupby
-    [self key]
+    [this key]
+    (.copy this)
     (set! groupby-key key)
     ;; "success"
     nil)
-  (getGroupbyKeys
-   [self]
-   groupby-key)
-  ;; (getAggreOldKeys
-  ;;  [self]
-  ;;  aggre-old-key)
-  (getAggreNewKeys
-   [self]
-   aggre-new-key)
-  (getAggreFunc
-   [self]
-   aggre-func)
+
   (aggregate
-    [self func old-keys new-keys]
+    [this func old-keys new-keys]
+    (.copy this)
     (if true
     ;;  (not= groupby-key [])
       (do
         (doseq [old-key old-keys]
           (set! aggre-func (conj aggre-func [func old-key])))
         ;; (set! aggre-old-key old-key)
         (doseq [new-key new-keys]
-         (set! aggre-new-key (conj aggre-new-key new-key)))
+          (set! aggre-new-key (conj aggre-new-key new-key)))
         ; "success"
         nil)
-        (throw (OperationException. "you must first group the dataframe by some keys then aggregate"))))
+      (throw (OperationException. "you must first group the dataframe by some keys then aggregate"))))
+
   (setRowInfo
-    [self new-col-desc new-col-set]
-    (let [original-filter (.getFilters self)
-          original-groupby-keys (.getGroupbyKeys self)
-          original-aggre-func (.getAggreFunc self)
+    [this new-col-desc new-col-set]
+    (.copy this)
+    (let [original-filter (.getFilters this)
+          original-groupby-keys (.getGroupbyKeys this)
+          original-aggre-func (.getAggreFunc this)
           new-filter-fns (map #(first %) original-filter)
           new-filter-cols (map (fn [fcols] (map #(first (first (get new-col-desc %))) fcols)) (doall (map #(last %) original-filter)))
           new-groupby-fns (map #(first %) original-groupby-keys)
           new-groupby-cols (map #(first (first (get new-col-desc (last %)))) original-groupby-keys)
           new-aggre-fns (map #(first %) original-aggre-func)
           new-aggre-cols (map #(first (first (get new-col-desc (last %)))) original-aggre-func)]
-      (if (not (empty? (.getFilters self)))
+      (if (not (empty? (.getFilters this)))
         (set! filters (vec (map vector new-filter-fns new-filter-cols))))
-      (if (not (empty? (.getGroupbyKeys self)))
-        (set! groupby-key (vec (map vector new-groupby-fns new-groupby-cols)))        
+      (if (not (empty? (.getGroupbyKeys this)))
+        (set! groupby-key (vec (map vector new-groupby-fns new-groupby-cols)))
         ;(set! groupby-key (vec (map #(first (first (get new-col-desc %))) original-groupby-keys)))
         )
-      (if (not (empty? (.getAggreFunc self)))
-        (set! aggre-func (vec (map vector new-aggre-fns new-aggre-cols))))
-      ))
-  )
+      (if (not (empty? (.getAggreFunc this)))
+        (set! aggre-func (vec (map vector new-aggre-fns new-aggre-cols))))))
+
+  (copy
+    [this]
+    (set! hist {:filters filters
+                :groupby-key groupby-key
+                :aggre-func aggre-func
+                :aggre-new-key aggre-new-key}))
+
+  (rollback
+    [this]
+    (if (not= hist {})
+      (do (set! filters (:filters hist))
+          (set! groupby-key (:groupby-key hist))
+          (set! aggre-func (:aggre-func hist))
+          (set! aggre-new-key (:aggre-new-key hist)))))
+  
+  (commit
+   [this]
+   (set! hist {})))