From 56a2bc6c5fcc50ba2595255175d7eb74afef85e3 Mon Sep 17 00:00:00 2001 From: Matt Buehlmaier Date: Fri, 25 Sep 2026 11:11:13 +0800 Subject: [PATCH] Keep argument column order when a join swaps its sides inner-join and outer-join stream the larger input and group the smaller one, swapping a and b when b's file is bigger. The swap leaked into the output: without :select, the second dataframe's columns (2_*) came first, in the header, the data, and preview. JoinedDataFrame gains a swapped flag, set only by the swapping branches of inner-join and outer-join: - getColNames presents the caller's order (record b first when swapped), so headers, preview and :select name resolution all follow it. - compute translates the user-facing select indices to internal ones (rows are always record-a columns then record-b columns); the existing write-index machinery then reorders the emitted values, so the join workers are untouched. - preview maps each side's old names onto the right half of the new ones. right-join keeps its existing swapped output order (its swap is semantic, not a size heuristic), and rolling joins never swap. Tests written first and seen failing on the 2_*-first order: a fast api-test for getColNames and preview, and an Onyx-backed regression test over disk and in-memory paths, including a :select on a swapped join. Existing fixtures are unaffected: the join tests' first input is the larger file, and the swapped-order fixtures all use :select. --- .../clojask/classes/JoinedDataFrame.clj | 26 ++++++++++++++----- src/main/clojure/clojask/dataframe.clj | 16 ++++++------ test/clojask/api_test.clj | 13 ++++++++++ test/clojask/regression_test.clj | 21 +++++++++++++++ 4 files changed, 61 insertions(+), 15 deletions(-) diff --git a/src/main/clojure/clojask/classes/JoinedDataFrame.clj b/src/main/clojure/clojask/classes/JoinedDataFrame.clj index 2119c1d..be38f37 100644 --- a/src/main/clojure/clojask/classes/JoinedDataFrame.clj +++ b/src/main/clojure/clojask/classes/JoinedDataFrame.clj @@ -26,6 +26,7 @@ type limit prefix + swapped output-func] GenDFIntf @@ -43,7 +44,11 @@ b-col-set (.getColNames b) a-col-header (mapv #(str a-col-prefix "_" %) a-col-set) b-col-header (mapv #(str b-col-prefix "_" %) b-col-set)] - (concat a-col-header b-col-header))) + ;; a swapped join presents the original argument order: record b is + ;; the caller's first dataframe + (if swapped + (concat b-col-header a-col-header) + (concat a-col-header b-col-header)))) (setOutput [this output] @@ -67,8 +72,8 @@ data-b (.preview b sample-size output-size format) old-a (.getColNames a) old-b (.getColNames b) - rep-key-a (zipmap old-a (take (count old-a) (.getColNames this))) - rep-key-b (zipmap old-b (take-last (count old-b) (.getColNames this))) + rep-key-a (zipmap old-a (if swapped (take-last (count old-a) (.getColNames this)) (take (count old-a) (.getColNames this)))) + rep-key-b (zipmap old-b (if swapped (take (count old-b) (.getColNames this)) (take-last (count old-b) (.getColNames this)))) data-a (map #(set/rename-keys % rep-key-a) data-a) data-b (map #(set/rename-keys % rep-key-b) data-b) data (map (fn [row-a row-b] (merge row-a row-b)) data-a data-b)] @@ -79,18 +84,25 @@ (compute [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))) select (if (= select [nil]) - (vec (take (+ (count (.getKeyIndex (.col-info a))) (count (.getKeyIndex (.col-info b)))) (iterate inc 0))) + (vec (take (+ na nb) (iterate inc 0))) (mapv (fn [key] (.indexOf (.getColNames this) key)) select)) - a-index (vec (apply sorted-set (remove (fn [num] (>= num (count (.getKeyIndex (.col-info a))))) 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) + a-index (vec (apply sorted-set (remove (fn [num] (>= num na)) internal))) ;; a-write - b-index (mapv #(- % (count (.getKeyIndex (.col-info a)))) (apply sorted-set (remove (fn [num] (< num (count (.getKeyIndex (.col-info a))))) select))) + b-index (mapv #(- % na) (apply sorted-set (remove (fn [num] (< num na)) internal))) b-index (if b-roll (vec (apply sorted-set (conj b-index b-roll))) b-index) b-roll (if b-roll (count (remove #(>= % b-roll) b-index)) nil) ;; b-write a-format (clojask.utils/formatters-by-position (.getFormatter (.col-info a)) a-index) b-format (clojask.utils/formatters-by-position (.getFormatter (.col-info b)) b-index) - write-index (mapv (fn [num] (count (remove #(>= % num) (concat a-index (mapv #(+ % (count (.getKeyIndex (.col-info a)))) b-index))))) select) + write-index (mapv (fn [num] (count (remove #(>= % num) (concat a-index (mapv #(+ % na) b-index))))) internal) mgroup-a (MGroupJoinOuter. (transient {}) (transient {}) false) mgroup-b (if (not= type 3) (MGroupJoin. (transient {}) (transient {}) (or (= 4 type) (= 5 type))) (MGroupJoinOuter. (transient {}) (transient {}) (or (= 4 type) (= 5 type)))) ] diff --git a/src/main/clojure/clojask/dataframe.clj b/src/main/clojure/clojask/dataframe.clj index 3de1cc4..6f7d51e 100644 --- a/src/main/clojure/clojask/dataframe.clj +++ b/src/main/clojure/clojask/dataframe.clj @@ -223,8 +223,8 @@ (let [size-a (.getSize (:stat a)) size-b (.getSize (:stat b))] (if (>= (compare size-a size-b) 0) - (JoinedDataFrame. a b a-keys b-keys nil nil 1 nil col-prefix (atom (.getOutput a))) - (JoinedDataFrame. b a b-keys a-keys nil nil 1 nil [(nth col-prefix 1) (nth col-prefix 0)] (atom (.getOutput a))))))) + (JoinedDataFrame. a b a-keys b-keys nil nil 1 nil col-prefix false (atom (.getOutput a))) + (JoinedDataFrame. b a b-keys a-keys nil nil 1 nil [(nth col-prefix 1) (nth col-prefix 0)] true (atom (.getOutput a))))))) (defn left-join [a b a-keys b-keys & {:keys [col-prefix] :or {col-prefix ["1" "2"]}}] @@ -242,7 +242,7 @@ (throw (TypeException. "The length of col-prefix should be equal to 2."))) (cond (some nil? (map second (concat a-keys b-keys))) (throw (TypeException. "Input includes non-existent column name(s)."))) - (JoinedDataFrame. a b a-keys b-keys nil nil 2 nil col-prefix (atom (.getOutput a))))) + (JoinedDataFrame. a b a-keys b-keys nil nil 2 nil col-prefix false (atom (.getOutput a))))) (defn right-join [a b a-keys b-keys & {:keys [col-prefix] :or {col-prefix ["1" "2"]}}] @@ -258,7 +258,7 @@ (throw (TypeException. "The length of left keys and right keys should be equal."))) (cond (some nil? (map second (concat a-keys b-keys))) (throw (TypeException. "Input includes non-existent column name(s)."))) - (JoinedDataFrame. b a b-keys a-keys nil nil 2 nil [(nth col-prefix 1) (nth col-prefix 0)] (atom (.getOutput a))))) + (JoinedDataFrame. b a b-keys a-keys nil nil 2 nil [(nth col-prefix 1) (nth col-prefix 0)] false (atom (.getOutput a))))) (defn outer-join [a b a-keys b-keys & {:keys [col-prefix] :or {col-prefix ["1" "2"]}}] @@ -277,8 +277,8 @@ (let [size-a (.getSize (:stat a)) size-b (.getSize (:stat b))] (if (>= (compare size-a size-b) 0) - (JoinedDataFrame. a b a-keys b-keys nil nil 3 nil col-prefix (atom (.getOutput a))) - (JoinedDataFrame. b a b-keys a-keys nil nil 3 nil [(nth col-prefix 1) (nth col-prefix 0)] (atom (.getOutput a))))))) + (JoinedDataFrame. a b a-keys b-keys nil nil 3 nil col-prefix false (atom (.getOutput a))) + (JoinedDataFrame. b a b-keys a-keys nil nil 3 nil [(nth col-prefix 1) (nth col-prefix 0)] true (atom (.getOutput a))))))) (defn rolling-join-forward [a b a-keys b-keys a-roll b-roll & {:keys [col-prefix limit] :or {col-prefix ["1" "2"] limit nil}}] @@ -300,7 +300,7 @@ (do (cond (not (and (not= a-roll nil) (not= b-roll nil))) (throw (TypeException. "Rolling keys include non-existent column name(s)."))) - (JoinedDataFrame. a b a-keys b-keys a-roll b-roll 4 limit col-prefix (atom (.getOutput a))))))) + (JoinedDataFrame. a b a-keys b-keys a-roll b-roll 4 limit col-prefix false (atom (.getOutput a))))))) ;; all of the code is the same as above except for the last line (defn rolling-join-backward @@ -323,7 +323,7 @@ (do (cond (not (and (not= a-roll nil) (not= b-roll nil))) (throw (TypeException. "Rolling keys include non-existent column name(s)."))) - (JoinedDataFrame. a b a-keys b-keys a-roll b-roll 5 limit col-prefix (atom (.getOutput a))))))) + (JoinedDataFrame. a b a-keys b-keys a-roll b-roll 5 limit col-prefix false (atom (.getOutput a))))))) (defn compute [this num-worker output-dir & {:keys [exception order output select exclude melt header in-memory] :or {exception false order false output nil select nil exclude nil melt vector header true in-memory false}}] diff --git a/test/clojask/api_test.clj b/test/clojask/api_test.clj index 1fd4edc..7b73c01 100644 --- a/test/clojask/api_test.clj +++ b/test/clojask/api_test.clj @@ -154,3 +154,16 @@ (is (thrown? OperationException (ck/filter df "Salary" (fn [_] (throw (Exception. "boom")))))) (is (= @opened @closed) "a failing errorPredetect closes its reader")))))) + +(deftest swapped-join-presents-original-column-order + (testing "inner/outer joins keep argument column order even when the sides swap internally" + (let [small (ck/dataframe input-b) + large (ck/dataframe input) + expected ["1_Employee" "1_EmployeeName" "1_DayOff" "1_UpdateDate" + "2_Employee" "2_EmployeeName" "2_Department" "2_Salary" "2_UpdateDate"] + inner (ck/inner-join small large ["Employee"] ["Employee"])] + (is (= expected (ck/get-col-names inner))) + (is (= expected (ck/get-col-names (ck/outer-join small large ["Employee"] ["Employee"])))) + (let [row (first (ck/preview inner 3 3))] + (is (= "20" (get row "1_DayOff"))) + (is (= "300" (get row "2_Salary"))))))) diff --git a/test/clojask/regression_test.clj b/test/clojask/regression_test.clj index 5f227cc..00f53f8 100644 --- a/test/clojask/regression_test.clj +++ b/test/clojask/regression_test.clj @@ -94,3 +94,24 @@ (ck/compute (ck/inner-join a b ["Employee"] ["Employee"]) 2 path) (is (< 1 (count (lines path)))) (is (= [] (stray-tmp-files ".clojask/join"))))) + +(deftest join-swaps-larger-side-but-keeps-column-order + (let [a-path (out "order-a.csv") + b-path (out "order-b.csv")] + (spit a-path "k,av\nx,a1\n") + (spit b-path (str "k,bv,bw\nx,b1,b2\n" (str/join "" (repeat 20 "y,f1,f2\n")))) + (doseq [in-memory [false true]] + (testing (str "in-memory " in-memory) + (let [a (ck/dataframe a-path) + b (ck/dataframe b-path) + inner (out "order-inner.csv") + outer (out "order-outer.csv") + selected (out "order-selected.csv")] + (ck/compute (ck/inner-join a b ["k"] ["k"]) 1 inner :in-memory in-memory) + (is (= ["1_k,1_av,2_k,2_bv,2_bw" "x,a1,x,b1,b2"] (lines inner))) + (ck/compute (ck/outer-join a b ["k"] ["k"]) 1 outer :in-memory in-memory) + (is (= "1_k,1_av,2_k,2_bv,2_bw" (first (lines outer)))) + (is (contains? (set (lines outer)) "x,a1,x,b1,b2")) + (is (= 22 (count (lines outer))) "header, one match, twenty unmatched b rows") + (ck/compute (ck/inner-join a b ["k"] ["k"]) 1 selected :select ["2_bv" "1_av"] :in-memory in-memory) + (is (= ["2_bv,1_av" "b1,a1"] (lines selected))))))))