From 04305950679953363713e25b9ac3c1abf1c28ac0 Mon Sep 17 00:00:00 2001 From: Matt Buehlmaier Date: Fri, 25 Sep 2026 10:56:57 +0800 Subject: [PATCH] Close abandoned CSV readers; bump clojask-io to 1.0.7 clojask-io 1.0.7 closes its reader when :data is fully consumed and returns a :close function for consumers that stop early. Use both: - dataframe construction opened three to four readers per call (header probe, stat, output fn) and abandoned them all after at most one line. It now realizes the io result once, reuses it for the header, size and output function, and closes it. - The stored read function gains a second arity returning {:data ... :close ...} so partial consumers can reach the reader. - preview fetches through that arity and closes in a finally, which also covers every errorPredetect and print-df, including when a preview throws (the errorPredetect failure path). Readers abandoned when an Onyx job dies mid-compute still wait for GC; full computes close on exhaustion via 1.0.7. Test written first and seen failing (5 readers opened, 0 closed): counts opened vs closed readers across construction, preview, and a failing errorPredetect. --- project.clj | 2 +- src/main/clojure/clojask/dataframe.clj | 70 +++++++++++++++++--------- src/main/clojure/clojask/preview.clj | 29 ++++++----- test/clojask/api_test.clj | 22 +++++++- 4 files changed, 84 insertions(+), 39 deletions(-) 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"))))))