Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 19 additions & 7 deletions src/main/clojure/clojask/classes/JoinedDataFrame.clj
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
type
limit
prefix
swapped
output-func]

GenDFIntf
Expand All @@ -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]
Expand All @@ -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)]
Expand All @@ -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))))
]
Expand Down
16 changes: 8 additions & 8 deletions src/main/clojure/clojask/dataframe.clj
Original file line number Diff line number Diff line change
Expand Up @@ -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"]}}]
Expand All @@ -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"]}}]
Expand All @@ -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"]}}]
Expand All @@ -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}}]
Expand All @@ -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
Expand All @@ -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}}]
Expand Down
13 changes: 13 additions & 0 deletions test/clojask/api_test.clj
Original file line number Diff line number Diff line change
Expand Up @@ -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")))))))
21 changes: 21 additions & 0 deletions test/clojask/regression_test.clj
Original file line number Diff line number Diff line change
Expand Up @@ -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))))))))
Loading