Fix select-col, delete-col and reorder-col; release 2.1.1

README.md

@@ -20,12 +20,12 @@ A Clojure dataframe library for **larger-than-memory** datasets. Process million
 
 **Add to project.clj:**
 ```clojure
-[com.github.clojure-finance/clojask "2.1.0"]
+[com.github.clojure-finance/clojask "2.1.1"]
 ```
 
 **Or deps.edn:**
 ```clojure
-com.github.clojure-finance/clojask {:mvn/version "2.1.0"}
+com.github.clojure-finance/clojask {:mvn/version "2.1.1"}
 ```
 
 **Basic example** (using [Employees-example.csv](test/clojask/Employees-example.csv)):

project.clj

@@ -1,4 +1,4 @@
-(defproject com.github.clojure-finance/clojask "2.1.0"
+(defproject com.github.clojure-finance/clojask "2.1.1"
   :description "Data analysis and manipulation library with parallel computing for larger-than-memory datasets"
   :url "https://github.com/clojure-finance/clojask"
   :license {:name "MIT"

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

@@ -14,12 +14,11 @@
   (getKeys [] "get collection of keys")
   (getKeyIndex [] "get map with key = column name, value = index")
   (getIndexKey [] "get map with key = index, value = column name")
-  (getDeletedCol [] "get indices of deleted columns")
+  (getView [] "get the indices of the output columns, in order")
   (setFormatter [format col])
   (setTypeFormatter [parser format col] "set parser and formatter as one change for rollback")
   (getFormatter [])
-  (delCol [col-to-del])
-  (setColInfo [new-col-set])
+  (setView [indices] "set the output columns and their order")
   (renameColInfo [old-col new-col])
   (copy [] "copy all the information for rollback purpose")
   (rollback [] "undo the change making use of the copied")
@@ -34,7 +33,7 @@
           ^:unsynchronized-mutable col-dsp
           ^:unsynchronized-mutable col-type
           ^:unsynchronized-mutable col-format
-          ^:unsynchronized-mutable col-deleted
+          ^:unsynchronized-mutable col-view ;; nil, or the indices of the output columns in order
           ^:unsynchronized-mutable hist]
 
   ;; method
@@ -46,7 +45,7 @@
     (set! key-index (zipmap col-keys (iterate inc 0)))
     (set! index-key (zipmap (iterate inc 0) col-keys))
     (set! col-dsp (zipmap (take (count colNames) (iterate inc 0)) (map vector (map vector (iterate inc 0)))))
-    (set! col-deleted (set nil)))
+    (set! col-view nil))
 
   (getFormatter
     [this]
@@ -73,9 +72,10 @@
     [this]
     index-key)
 
-  (getDeletedCol
+  (getView
     [this]
-    col-deleted)
+    ;; nil until a column is deleted or moved: every column, in index order
+    (or col-view (vec (range (count index-key)))))
 
   (operate
     [this operation col]
@@ -100,6 +100,7 @@
             (set! key-index (assoc key-index newCol (count key-index)))
             (set! index-key (assoc index-key (count index-key) newCol))
             (set! col-dsp (assoc col-dsp (get key-index newCol) (conj [(vec (map (fn [_] (get key-index _)) col))] operation)))
+            (when col-view (set! col-view (conj col-view (get key-index newCol))))
             nil))
         (do
           (throw (OperationException. (str external " are not original column names")))))))
@@ -124,27 +125,10 @@
     (.setType this parser col)
     (set! col-format (assoc col-format (get key-index col) format)))
 
-  (delCol
-    [this col-to-delete]
+  (setView
+    [this indices]
     (.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 (map original-key-index new-col-set)
-          original-type (.getType this)
-          original-format (.getFormatter this)]
-      (set! col-keys (vec new-col-set))
-      (set! key-index (zipmap new-col-set (iterate inc 0)))
-      (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))))
-      (if (not (empty? (.getFormatter this)))
-        (set! col-format (zipmap (map #(first (first (get col-dsp (first %)))) original-format) (map last original-format))))))
+    (set! col-view (vec indices)))
 
   (renameColInfo
     [this old-col new-col]
@@ -162,7 +146,7 @@
                 :col-dsp col-dsp
                 :col-type col-type
                 :col-format col-format
-                :col-deleted col-deleted}))
+                :col-view col-view}))
 
   (rollback
    [this]
@@ -173,7 +157,7 @@
          (set! col-type (:col-type hist))
          (set! col-format (:col-format hist))
          (set! col-dsp (:col-dsp hist))
-         (set! col-deleted (:col-deleted hist)))))
+         (set! col-view (:col-view hist)))))
 
   (commit
    [this]

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

@@ -20,6 +20,15 @@
 
 (def debug (atom false))
 
+(defn- view-names
+  "The column names that delete-col, select-col and reorder-col work on.
+   They choose the output columns of a plain compute, so they cannot follow
+   group-by or aggregate."
+  [df fn-name]
+  (when (or (not= [] (.getGroupbyKeys (:row-info df))) (not= [] (.getAggreFunc (:row-info df))))
+    (throw (OperationException. (str fn-name " cannot follow group-by or aggregate; use compute's :select instead."))))
+  (.getColNames df))
+
 (definterface GenDFIntf
   (checkInputPathClash [path] "check if paths clashes with dataframes a/b input path")
   (getColNames [] "get the names of all the columns")
@@ -38,7 +47,7 @@
   (setParser [parser col] "add the parser for a col which acts like setType")
   (colDesc [] "get column description in ColInfo")
   (colTypes [] "get column type in ColInfo")
-  (getColIndex [] "get column indices, excluding deleted columns")
+  (getColIndex [] "get the indices of the output columns, in order (deleted ones left out)")
   (getAggreColNames [] "get column names if there is aggregate")
   (getStat [] "get the statistics of the dataframe")
   (delCol [col-to-del] "delete one or more columns in the dataframe")
@@ -119,10 +128,12 @@
   (compute
     [this ^int num-worker ^String output-dir ^boolean exception ^boolean order select melt ifheader out]
     (let [key-index (.getKeyIndex (:col-info this))
+          view (vec (.getColIndex this))
           select (if (coll? select) select [select])
-          index (if (= select [nil]) (take (count key-index) (iterate inc 0)) (keep key-index select))]
+          index (if (= select [nil]) view (keep key-index select))]
       (assert (or (= (count select) (count index)) (= select [nil])) "Must select existing columns; get-col-names lists them.")
-      (if (= ifheader true) (.printCol this output-dir index out))
+      ;; printCol takes positions in getColNames, which lists the view
+      (if (= ifheader true) (.printCol this output-dir (mapv #(.indexOf view %) index) out))
       (let [res (engine/start-compute num-worker batch-size this output-dir exception index melt out)]
         (if (= res "success")
           "success"
@@ -203,8 +214,7 @@
 
   (getColIndex
     [this]
-    (let [index-key (.getIndexKey (:col-info this))]
-      (take (count index-key) (iterate inc 0))))
+    (.getView (:col-info this)))
 
   (getStat
     [this]
@@ -228,21 +238,29 @@
           aggre-new-keys (.getAggreNewKeys (:row-info this))]
       (concat groupby-keys-value aggre-new-keys)))
 
-  ;; deprecated
+  ;; delCol and reorderCol only change which columns compute writes and in
+  ;; which order; operations, filters and types stay on their columns.
   (delCol
     [this col-to-del]
-    (cond (not (= 0 (count (u/are-in col-to-del this))))
-          (throw (TypeException. "Input includes non-existent column name(s).")))
-    (.delCol (.col-info this) col-to-del)
-    this)
-  ;; deprecated
+    (let [names (view-names this "delete-col")
+          col-to-del (if (coll? col-to-del) col-to-del [col-to-del])
+          kept (vec (remove (set col-to-del) names))]
+      (cond (not= [] (remove (set names) col-to-del))
+            (throw (TypeException. "Input includes non-existent column name(s)."))
+            (= [] kept)
+            (throw (OperationException. "Cannot delete every column.")))
+      (.setView col-info (mapv (.getKeyIndex col-info) kept))
+      this))
+
   (reorderCol
     [this new-col-order]
-    (cond (not (= (set (.getKeys (.col-info this))) (set new-col-order)))
-          (throw (TypeException. "Set of input in reorder-col contains column(s) that do not exist in dataframe.")))
-    (.setColInfo (.col-info this) new-col-order)
-    (.setRowInfo (.row-info this) (.getDesc (.col-info this)) new-col-order)
-    this)
+    (let [names (view-names this "reorder-col")]
+      (cond (not (and (coll? new-col-order)
+                      (= (count names) (count new-col-order))
+                      (= (set names) (set new-col-order))))
+            (throw (TypeException. "reorder-col needs every column exactly once; get-col-names lists them.")))
+      (.setView col-info (mapv (.getKeyIndex col-info) new-col-order))
+      this))
 
   (renameCol
     [this old-col new-col]

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

@@ -84,15 +84,16 @@
     [this ^int num-worker ^String output-dir ^boolean exception ^boolean order select ifheader out inmemory]
     (let [select (if (coll? select) select [select])
           na (count (.getKeyIndex (.col-info a)))
-          nb (count (.getKeyIndex (.col-info b)))
+          ;; the internal index of each column in getColNames order: internal
+          ;; rows are always record-a columns then record-b columns, and each
+          ;; side lists only its output columns (get-col-names)
+          a-cols (vec (.getColIndex a))
+          b-cols (mapv #(+ na %) (.getColIndex b))
+          joined-index (vec (if swapped (concat b-cols a-cols) (concat a-cols b-cols)))
           select (if (= select [nil])
-                   (vec (take (+ na nb) (iterate inc 0)))
+                   (vec (range (count joined-index)))
                    (mapv (fn [key] (.indexOf (.getColNames this) key)) select))
-          ;; select follows the user-facing getColNames order; internal rows
-          ;; are always record-a columns then record-b columns
-          internal (if swapped
-                     (mapv (fn [i] (if (< i nb) (+ i na) (- i nb))) select)
-                     select)
+          internal (mapv joined-index select)
           a-index (vec (apply sorted-set (remove (fn [num] (>= num na)) internal)))
           ;; a-write 
           b-index (mapv #(- % na) (apply sorted-set (remove (fn [num] (< num na)) internal)))

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

@@ -8,7 +8,6 @@
   (filter [cols predicate])
   (groupby [a])
   (aggregate [func old-key new-key])
-  (setRowInfo [new-col-desc new-col-set])
   (copy [])
   (rollback [])
   (commit []))
@@ -57,26 +56,6 @@
       (set! aggre-new-key (conj aggre-new-key new-key)))
     nil)
 
-  (setRowInfo
-    [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 this)))
-        (set! filters (vec (map vector new-filter-fns new-filter-cols))))
-      (if (not (empty? (.getGroupbyKeys this)))
-        (set! groupby-key (vec (map vector new-groupby-fns new-groupby-cols)))
-        )
-      (if (not (empty? (.getAggreFunc this)))
-        (set! aggre-func (vec (map vector new-aggre-fns new-aggre-cols))))))
-
   (copy
     [this]
     (set! hist {:filters filters

src/main/clojure/clojask/dataframe.clj

@@ -9,7 +9,6 @@
             [clojask.classes.JoinedDataFrame]
             [clojask.utils :as u]
             [clojure.pprint :as pprint]
-            [clojure.set :as set]
             [clojure.string :as str]
             [clojure.java.io :as io])
   (:import [clojask.classes.ColInfo ColInfo]
@@ -378,12 +377,16 @@
       (dataframe output-dir :if-header (boolean (or custom-header header)))
       (persistent! (deref ret)))))
 
-;; ============== Below functions are deprecated ==============
-
 (defn select-col
+  "Keep only the columns in col-to-keep, in their current order."
   [this col-to-keep]
-  (let [col-to-del (set/difference (set (.getColNames this)) (set col-to-keep))]
-    (.delCol this (vec col-to-del))))
+  (let [col-to-keep (if (coll? col-to-keep) col-to-keep [col-to-keep])
+        names (.getColNames this)]
+    (cond (not= [] (remove (set names) col-to-keep))
+          (throw (TypeException. "Input includes non-existent column name(s).")))
+    (let [result (.delCol this (vec (remove (set col-to-keep) names)))]
+      (.errorPredetect this "invalid arguments passed to select-col function")
+      result)))
 
 (defn delete-col
   [this col-to-del]

src/main/clojure/clojask/preview.clj

@@ -18,9 +18,10 @@
         operations (.getDesc (:col-info  dataframe))
         types (.getType (:col-info  dataframe))
         filters (.getFilters (:row-info dataframe))
-        indices index
         no-aggre (= (.getAggreFunc (:row-info dataframe)) []) ;; if need to groupby & aggregate
         no-groupby (= (.getGroupbyKeys (:row-info dataframe)) [])
+        ;; group-by and aggregate address columns by index, so they get every column
+        indices (if (and no-aggre no-groupby) index (range (count index-key)))
         ;;
         preview-work-func (fn [seg]
                             (let [data (:d seg)]
@@ -31,9 +32,9 @@
                               (fn [row]
                                 ;; a row removed by a filter has no :d
                                 (when-let [d (:d row)]
-                                  (mapv (fn [_] (if-let [formatter (get formatters _)]
-                                                  (formatter (nth d _))
-                                                  (nth d _))) index)))
+                                  (mapv (fn [v i] (if-let [formatter (get formatters i)]
+                                                    (formatter v)
+                                                    v)) d index)))
                               (fn [row]
                                 (:d row))) ;; the function body of output operation (take over the work in output node) without formatting