<commit_before>(ns pigpen.cascading.runtime
  (:import (org.apache.hadoop.io BytesWritable)
           (pigpen.cascading OperationUtil)
           (java.util ArrayList)
           (clojure.lang ISeq))
  (:require [pigpen.runtime :as rt]
            [pigpen.raw :as raw]
            [pigpen.oven :as oven]
            [taoensso.nippy :refer [freeze thaw]]))

(defmulti hybrid->clojure
          "Converts a hybrid cascading/clojure data structure into 100% clojure.

           DataByteArrays are assumed to be frozen clojure structures.

           Only raw pig types are expected here - anything normal (Boolean, String, etc)
           should be frozen."
          type)

(defmethod hybrid->clojure :default [value]
  (throw (IllegalStateException. (str "Unexpected value:" value))))

(defmethod hybrid->clojure nil [value]
  ;; nils are allowed because they occur in joins
  nil)

(defmethod hybrid->clojure Number [value]
  ;; Numbers are allowed because RANK produces them
  value)

(defmethod hybrid->clojure BytesWritable [^BytesWritable value]
  (-> value (OperationUtil/getBytes) thaw))

(defmethod hybrid->clojure ISeq [^ISeq value]
  (mapv hybrid->clojure value))

;; ******* Serialization ********
(defn ^:private cs-freeze [value]
  (BytesWritable. (freeze value {:skip-header? true, :legacy-mode true})))

(defmethod pigpen.runtime/pre-process [:cascading :frozen-with-nils]
  [_ _]
  (fn [args]
    (mapv #(thaw (.getBytes %)) args)))

(defmethod pigpen.runtime/pre-process [:cascading :frozen]
  [_ _]
  (fn [args]
    (mapv hybrid->clojure args)))

(defmethod pigpen.runtime/post-process [:cascading :native]
  [_ _]
  identity)

(defmethod pigpen.runtime/post-process [:cascading :frozen]
  [_ _]
  (fn [args]
    (mapv cs-freeze args)))

(defmethod pigpen.runtime/post-process [:cascading :frozen-with-nils]
  [_ _]
  (fn [args]
    (mapv cs-freeze args)))

(defmethod pigpen.runtime/post-process [:cascading :native-key-frozen-val]
  [_ _]
  (fn [[key value]]
    [key (cs-freeze value)]))<commit_msg>removed redundant method and implemented distinction for nils<commit_after>(ns pigpen.cascading.runtime
  (:import (org.apache.hadoop.io BytesWritable)
           (pigpen.cascading OperationUtil)
           (java.util ArrayList)
           (clojure.lang ISeq))
  (:require [pigpen.runtime :as rt]
            [pigpen.raw :as raw]
            [pigpen.oven :as oven]
            [taoensso.nippy :refer [freeze thaw]]))

(defmulti hybrid->clojure
          "Converts a hybrid cascading/clojure data structure into 100% clojure.

           DataByteArrays are assumed to be frozen clojure structures.

           Only raw pig types are expected here - anything normal (Boolean, String, etc)
           should be frozen."
          type)

(defmethod hybrid->clojure :default [value]
  (throw (IllegalStateException. (str "Unexpected value:" value))))

(defmethod hybrid->clojure nil [value]
  ;; nils are allowed because they occur in joins
  nil)

(defmethod hybrid->clojure Number [value]
  ;; Numbers are allowed because RANK produces them
  value)

(defmethod hybrid->clojure BytesWritable [^BytesWritable value]
  (-> value (OperationUtil/getBytes) thaw))

(defmethod hybrid->clojure ISeq [^ISeq value]
  (mapv hybrid->clojure value))

;; ******* Serialization ********
(defn ^:private cs-freeze [value]
  (BytesWritable. (freeze value {:skip-header? true, :legacy-mode true})))

(defn ^:private cs-freeze-with-nils [value]
  (if value (cs-freeze value)))

(defmethod pigpen.runtime/pre-process [:cascading :frozen]
  [_ _]
  (fn [args]
    (mapv hybrid->clojure args)))

(defmethod pigpen.runtime/post-process [:cascading :native]
  [_ _]
  identity)

(defmethod pigpen.runtime/post-process [:cascading :frozen]
  [_ _]
  (fn [args]
    (mapv cs-freeze args)))

(defmethod pigpen.runtime/post-process [:cascading :frozen-with-nils]
  [_ _]
  (fn [args]
    (mapv cs-freeze-with-nils args)))

(defmethod pigpen.runtime/post-process [:cascading :native-key-frozen-val]
  [_ _]
  (fn [[key value]]
    [key (cs-freeze value)]))
{'fraction_non_alphanumeric': 0.13346515076618884, 'fraction_numerical': 0.0014829461196243204, 'mean_word_length': 3.452145214521452, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5526249', 'n_tokens_mistral': 1414, 'n_tokens_neox': 1361, 'n_words': 354}