diff --git a/README.md b/README.md index 7e03d07..f63cbc0 100644 --- a/README.md +++ b/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)): diff --git a/project.clj b/project.clj index 48fce74..b61db20 100644 --- a/project.clj +++ b/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" diff --git a/src/main/clojure/clojask/classes/ColInfo.clj b/src/main/clojure/clojask/classes/ColInfo.clj index c5db282..4aa4420 100644 --- a/src/main/clojure/clojask/classes/ColInfo.clj +++ b/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] diff --git a/src/main/clojure/clojask/classes/DataFrame.clj b/src/main/clojure/clojask/classes/DataFrame.clj index 2931d72..62ca8fb 100644 --- a/src/main/clojure/clojask/classes/DataFrame.clj +++ b/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] diff --git a/src/main/clojure/clojask/classes/JoinedDataFrame.clj b/src/main/clojure/clojask/classes/JoinedDataFrame.clj index ec8cc84..e10fe70 100644 --- a/src/main/clojure/clojask/classes/JoinedDataFrame.clj +++ b/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))) diff --git a/src/main/clojure/clojask/classes/RowInfo.clj b/src/main/clojure/clojask/classes/RowInfo.clj index 84fe16e..85a66bb 100644 --- a/src/main/clojure/clojask/classes/RowInfo.clj +++ b/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 diff --git a/src/main/clojure/clojask/dataframe.clj b/src/main/clojure/clojask/dataframe.clj index f70fc3d..d66e4e7 100644 --- a/src/main/clojure/clojask/dataframe.clj +++ b/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] diff --git a/src/main/clojure/clojask/preview.clj b/src/main/clojure/clojask/preview.clj index cb5a379..01b3c20 100644 --- a/src/main/clojure/clojask/preview.clj +++ b/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 diff --git a/test/clojask/regression_test.clj b/test/clojask/regression_test.clj index 8de3cd8..b17b229 100644 --- a/test/clojask/regression_test.clj +++ b/test/clojask/regression_test.clj @@ -1,11 +1,12 @@ (ns regression-test - "Computes that each exercised a bug fixed in 2.0.5." + "Computes that each exercised a bug fixed in 2.0.5 or later." (:require [clojure.test :refer :all] [clojure.java.io :as io] [clojure.string :as str] [clojask.dataframe :as ck] [clojask.api.aggregate :as agg] - [clojask.api.gb-aggregate :as gb])) + [clojask.api.gb-aggregate :as gb]) + (:import [com.clojask.exception TypeException OperationException])) (def input "test/clojask/Employees-example.csv") (def dir "test/clojask/test_outputs/regression/") @@ -130,3 +131,92 @@ "5,Evelyn,13,800,2020/12/03" "6,Ferdinand,21,700,2020/12/05"] (lines path))))) + +(deftest reorder-col-keeps-operations-filters-and-types + (let [df (ck/dataframe input) + path (out "reorder.csv") + order ["UpdateDate" "Employee" "EmployeeName" "Department" "Salary"]] + (ck/set-type df "Employee" "int") + (ck/operate df inc "Employee") + (ck/filter df "Salary" #(> (Integer/parseInt %) 800)) + (ck/set-formatter df "Department" #(str "D" %)) + (ck/reorder-col df order) + (is (= order (ck/get-col-names df))) + (ck/compute df 1 path) + (is (= ["UpdateDate,Employee,EmployeeName,Department,Salary" + "2020/12/03,4,Carla,D12,900" + "2020/12/05,5,Daniel,D12,1000" + "2020/11/26,8,Amy,D11,50000"] + (lines path))) + (is (= {"UpdateDate" "2020/12/03" "Employee" "4" "EmployeeName" "Carla" "Department" "D12" "Salary" "900"} + (update (first (ck/preview df 100 1 :format true)) "Employee" str))))) + +(deftest delete-col-and-select-col-choose-the-output-columns + (let [df (ck/dataframe input) + path (out "delete.csv")] + (ck/operate df #(str % "!") "EmployeeName") + ;; a filter on a column that is deleted afterwards still applies + (ck/filter df "Salary" #(> (Integer/parseInt %) 800)) + (ck/delete-col df ["Salary" "UpdateDate"]) + (is (= ["Employee" "EmployeeName" "Department"] (ck/get-col-names df))) + (ck/compute df 1 path) + (is (= ["Employee,EmployeeName,Department" "3,Carla!,12" "4,Daniel!,12" "7,Amy!,11"] (lines path))) + (ck/select-col df ["Department" "Employee"]) + (is (= ["Employee" "Department"] (ck/get-col-names df)) "select-col keeps the current order") + (ck/operate df str ["Employee" "Department"] "Both") + (is (= ["Employee" "Department" "Both"] (ck/get-col-names df)) "a new column is added to the output") + (is (= [["Employee" "Department" "Both"] ["3" "12" "312"] ["4" "12" "412"] ["7" "11" "711"]] + (mapv vec (ck/compute df 1 nil)))) + (is (= {"Employee" "3" "Department" "12" "Both" "312"} (first (ck/preview df 100 1)))))) + +(deftest aggregate-can-use-a-deleted-column + (let [df (ck/dataframe input) + path (out "delete-aggregate.csv")] + (ck/set-type df "Salary" "double") + (ck/delete-col df ["Salary" "EmployeeName"]) + (ck/group-by df ["Department"]) + (ck/aggregate df gb/max ["Salary"] ["Salary-max"]) + (is (= 4 (count (ck/preview df 100 10)))) + (ck/compute df 1 path) + (is (= 5 (count (lines path))) "header and four departments") + (is (= "Department,Salary-max" (first (lines path)))))) + +(deftest joins-use-the-output-columns-of-each-side + (doseq [in-memory [false true]] + (testing (str "in-memory " in-memory) + (let [a (ck/dataframe input) + b (ck/dataframe "test/clojask/Employees-info-example.csv") + path (out "join-view.csv") + swapped (out "join-view-swapped.csv") + selected (out "join-view-selected.csv")] + (ck/select-col a ["Employee" "Salary"]) + (ck/reorder-col a ["Salary" "Employee"]) + ;; the join key of b is deleted from b's output, not from the join + (ck/delete-col b ["Employee" "UpdateDate"]) + (let [j (ck/inner-join a b ["Employee"] ["Employee"])] + (is (= ["1_Salary" "1_Employee" "2_EmployeeName" "2_DayOff"] (ck/get-col-names j))) + (ck/compute j 1 path :in-memory in-memory) + (is (= "1_Salary,1_Employee,2_EmployeeName,2_DayOff" (first (lines path)))) + (is (contains? (set (lines path)) "300,1,Alice,20"))) + (testing "the larger side second, so the join swaps internally" + (let [j (ck/inner-join b a ["Employee"] ["Employee"])] + (ck/compute j 1 swapped :in-memory in-memory) + (is (= "1_EmployeeName,1_DayOff,2_Salary,2_Employee" (first (lines swapped)))) + (is (contains? (set (lines swapped)) "Alice,20,300,1")) + (ck/compute j 1 selected :select ["2_Employee" "1_DayOff"] :in-memory in-memory) + (is (contains? (set (lines selected)) "1,20")))))))) + +(deftest column-functions-reject-bad-input + (let [df (ck/dataframe input) + names (ck/get-col-names df)] + (is (thrown? TypeException (ck/delete-col df ["Nope"]))) + (is (thrown? TypeException (ck/select-col df ["Employee" "Nope"]))) + (is (thrown? OperationException (ck/delete-col df names))) + (is (thrown? TypeException (ck/reorder-col df ["Employee"]))) + (is (thrown? TypeException (ck/reorder-col df (assoc names 1 "Employee")))) + (is (= names (ck/get-col-names df)) "a rejected call changes nothing") + (ck/delete-col df ["Salary"]) + (is (thrown? TypeException (ck/reorder-col df names)) "a deleted column is not output any more") + (ck/group-by df ["Department"]) + (is (thrown? OperationException (ck/delete-col df ["Employee"]))) + (is (thrown? OperationException (ck/reorder-col df ["Department"])))))