diff --git a/project.clj b/project.clj index 6bbe5c9..c620d54 100644 --- a/project.clj +++ b/project.clj @@ -10,7 +10,7 @@ [com.taoensso/timbre "5.2.1"] ;; [techascent/tech.ml.dataset "5.17" :exclusions [[ch.qos.logback/logback-classic][org.slf4j/slf4j-api]]] [com.google.code.externalsortinginjava/externalsortinginjava "0.6.0"] - [com.github.clojure-finance/clojask-io "1.0.6"] + [com.github.clojure-finance/clojask-io "1.0.7"] [com.github.clojure-finance/clojure-heap "1.0.3"]] ;; lein deploy clojars, credentials from CLOJARS_USERNAME / CLOJARS_PASSWORD :deploy-repositories [["clojars" {:url "https://clojars.org/repo" diff --git a/src/main/clojure/clojask/dataframe.clj b/src/main/clojure/clojask/dataframe.clj index 213466b..3de1cc4 100644 --- a/src/main/clojure/clojask/dataframe.clj +++ b/src/main/clojure/clojask/dataframe.clj @@ -83,37 +83,57 @@ [path & {:keys [if-header] :or {if-header true}}] (try (if (fn? path) - ;; if the input is the clojask-io.input - (if (:clojask-io (path)) - (let [io-func path - read-func (fn [] (:data (io-func))) - colNames (u/check-duplicate-col (if if-header (doall (first (read-func))) (generate-col (count (first (read-func)))))) - col-info (ColInfo. (doall (map keyword colNames)) {} {} {} {} {} {} {}) - row-info (RowInfo. [] [] [] [] {}) - stat (compute-stat path io-func) - func (if if-header (fn [] (rest (read-func))) read-func)] - (.init col-info colNames) - (DataFrame. (:path (path)) func 300 col-info row-info stat (atom (or (:output (io-func)) (fn [wtr rows] (doseq [msg rows](.write wtr (str (str/join "," msg) "\n")))))) if-header)) - ;; if the is the lazy seq function - (let [headers (first (path)) - colNames (u/check-duplicate-col (if if-header headers (generate-col (count headers)))) - col-info (ColInfo. (doall (map keyword colNames)) {} {} {} {} {} {} {}) - row-info (RowInfo. [] [] [] [] {}) - stat (compute-stat path) - func (if if-header (fn [] (rest (path))) path)] - (.init col-info colNames) - (DataFrame. nil func 300 col-info row-info stat (atom (fn [wtr rows] (doseq [msg rows] (.write wtr (str (str/join "," msg) "\n"))))) if-header))) + ;; probe is realized once: header row, :path, :output, then closed + (let [probe (path)] + ;; if the input is the clojask-io.input + (if (:clojask-io probe) + (let [io-func path + read-func (fn [] (:data (io-func))) + colNames (u/check-duplicate-col (if if-header (doall (first (:data probe))) (generate-col (count (first (:data probe)))))) + col-info (ColInfo. (doall (map keyword colNames)) {} {} {} {} {} {} {}) + row-info (RowInfo. [] [] [] [] {}) + stat (compute-stat path (fn [] probe)) + ;; arity 1 exposes {:data ... :close ...} so partial consumers + ;; (preview / errorPredetect) can close the reader they abandon + func (if if-header + (fn ([] (rest (read-func))) + ([_] (let [res (io-func)] {:data (rest (:data res)) :close (:close res)}))) + (fn ([] (read-func)) + ([_] (let [res (io-func)] {:data (:data res) :close (:close res)})))) + output-func (or (:output probe) (fn [wtr rows] (doseq [msg rows] (.write wtr (str (str/join "," msg) "\n")))))] + (.init col-info colNames) + (when-let [close (:close probe)] (close)) + (DataFrame. (:path probe) func 300 col-info row-info stat (atom output-func) if-header)) + ;; if the is the lazy seq function + (let [headers (first probe) + colNames (u/check-duplicate-col (if if-header headers (generate-col (count headers)))) + col-info (ColInfo. (doall (map keyword colNames)) {} {} {} {} {} {} {}) + row-info (RowInfo. [] [] [] [] {}) + stat (compute-stat path) + func (if if-header + (fn ([] (rest (path))) + ([_] {:data (rest (path))})) + (fn ([] (path)) + ([_] {:data (path)})))] + (.init col-info colNames) + (DataFrame. nil func 300 col-info row-info stat (atom (fn [wtr rows] (doseq [msg rows] (.write wtr (str (str/join "," msg) "\n"))))) if-header)))) ;; if the input is the path string (let [io-func (fn [] (read-file path :stat true :output true)) + probe (io-func) read-func (fn [] (:data (io-func))) - colNames (u/check-duplicate-col (if if-header (doall (first (read-func))) (generate-col (count (first (read-func)))))) + colNames (u/check-duplicate-col (if if-header (doall (first (:data probe))) (generate-col (count (first (:data probe)))))) col-info (ColInfo. (doall (map keyword colNames)) {} {} {} {} {} {} {}) row-info (RowInfo. [] [] [] [] {}) - stat (compute-stat path io-func) - func (if if-header (fn [] (rest (read-func))) read-func)] + stat (compute-stat path (fn [] probe)) + func (if if-header + (fn ([] (rest (read-func))) + ([_] (let [res (io-func)] {:data (rest (:data res)) :close (:close res)}))) + (fn ([] (read-func)) + ([_] (let [res (io-func)] {:data (:data res) :close (:close res)})))) + output-func (or (:output probe) (fn [wtr rows] (doseq [msg rows] (.write wtr (str (str/join "," msg) "\n")))))] (.init col-info colNames) - (DataFrame. path func 300 col-info row-info stat (atom (or (:output (io-func)) (fn [wtr rows] (doseq [msg rows] (.write wtr (str (str/join "," msg) "\n")))))) if-header) - )) + (when-let [close (:close probe)] (close)) + (DataFrame. path func 300 col-info row-info stat (atom output-func) if-header))) (catch Exception e (do (throw (TypeException. "Error in initializing the dataframe." e)) diff --git a/src/main/clojure/clojask/preview.clj b/src/main/clojure/clojask/preview.clj index c73a77d..cb5a379 100644 --- a/src/main/clojure/clojask/preview.clj +++ b/src/main/clojure/clojask/preview.clj @@ -10,7 +10,8 @@ formatters (.getFormatter (:col-info dataframe)) index (.getColIndex dataframe) header (.getColNames dataframe) - csv-data ((.getFunc dataframe)) + src ((.getFunc dataframe) :map) + csv-data (:data src) data (map zipmap (repeat [:id :d]) (map vector (iterate inc 0) csv-data)) sample (take sample-size data) ;; lazy source data (take sample size) ;; define the variables needed in the following functions @@ -37,17 +38,21 @@ (:d row))) ;; the function body of output operation (take over the work in output node) without formatting ;; ========== no need to change =========== - compute-res (loop [rows sample res (transient [])] ;; the result of normal compute - (if (= rows []) ;; exceed sample size - (persistent! res) - (let [row (first rows) - rest (rest rows) - row (preview-work-func row) - row-res (preview-output-func row) - res (if row-res (conj! res row-res) res)] - (if (>= (count res) return-size) - (persistent! res) - (recur rest res)))))] + compute-res (try + (loop [rows sample res (transient [])] ;; the result of normal compute + (if (= rows []) ;; exceed sample size + (persistent! res) + (let [row (first rows) + rest (rest rows) + row (preview-work-func row) + row-res (preview-output-func row) + res (if row-res (conj! res row-res) res)] + (if (>= (count res) return-size) + (persistent! res) + (recur rest res))))) + ;; the reader behind :data stops mid-file whenever + ;; sample-size or return-size is hit before the end + (finally (when-let [close (:close src)] (close))))] (if (and no-groupby no-aggre) (mapv (fn [row-v] (zipmap header row-v)) compute-res) ;; need to do aggregate diff --git a/test/clojask/api_test.clj b/test/clojask/api_test.clj index 0a45bf5..1fd4edc 100644 --- a/test/clojask/api_test.clj +++ b/test/clojask/api_test.clj @@ -6,7 +6,8 @@ [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] + [clojask-io.input :as cio]) (:import [com.clojask.exception OperationException TypeException])) (def input "test/clojask/Employees-example.csv") @@ -134,3 +135,22 @@ (is (= [1 2] (gb/mode [1 2 1 2 3]))) (is (= [1.0 2.0] (gb/mode [2.0 1.0 2.0 1.0 3.0])) "used to come back in hash order") (is (= ["a" "b"] (gb/mode ["b" "a" "b" "a"])))) + +(deftest dataframe-and-preview-close-abandoned-readers + (testing "construction, preview and a failing errorPredetect close every reader they open" + (let [opened (atom 0) + closed (atom 0) + orig cio/read-file] + (with-redefs [cio/read-file + (fn [& args] + (let [res (apply orig args)] + (swap! opened inc) + (update res :close (fn [close] (fn [] (swap! closed inc) (when close (close)))))))] + (let [df (ck/dataframe input)] + (is (pos? @opened) "construction reads the input") + (is (= @opened @closed) "construction closes its readers") + (ck/preview df 5 5) + (is (= @opened @closed) "preview closes its reader") + (is (thrown? OperationException + (ck/filter df "Salary" (fn [_] (throw (Exception. "boom")))))) + (is (= @opened @closed) "a failing errorPredetect closes its reader"))))))