<commit_before>(ns onyx.schema
  (:require [schema.core :as s]))

(def NamespacedKeyword
  (s/pred (fn [kw]
                 (and (keyword? kw)
                      (namespace kw)))
               'keyword-namespaced?))

(def Function
  (s/pred fn? 'fn?))

(def TaskName
  (s/pred (fn [v]
                 (and (not= :all v)
                      (not= :none v)
                      (keyword? v)))
               'task-name?))

(defn ^{:private true} edge-two-nodes? [edge]
  (= (count edge) 2))

(def ^{:private true} edge-validator
  (s/->Both [(s/pred vector? 'vector?)
                  (s/pred edge-two-nodes? 'edge-two-nodes?)
                  [TaskName]]))

(def Workflow
  (s/->Both [(s/pred vector? 'vector?)
                  [edge-validator]]))

(def Language
  (s/enum :java :clojure))

(def ^{:private true} base-task-map
  {:onyx/name TaskName
   :onyx/type (s/enum :input :output :function)
   :onyx/batch-size (s/pred pos? 'pos?)
   (s/optional-key :onyx/uniqueness-key) s/Any
   (s/optional-key :onyx/restart-pred-fn) s/Keyword
   (s/optional-key :onyx/language) Language
   (s/optional-key :onyx/batch-timeout) (s/pred pos? 'pos?)
   (s/optional-key :onyx/doc) s/Str
   (s/optional-key :onyx/max-peers) (s/pred pos? 'pos?)
   s/Keyword s/Any})

(def FluxPolicy 
  (s/enum :continue :kill :recover))

(def ^{:private true} partial-grouping-task
  {(s/optional-key :onyx/group-by-key) s/Any
   (s/optional-key :onyx/group-by-fn) NamespacedKeyword
   :onyx/min-peers s/Int
   :onyx/flux-policy FluxPolicy})

(defn grouping-task? [task-map]
  (and (#{:function :output} (:onyx/type task-map))
       (or (not (nil? (:onyx/group-by-key task-map)))
           (not (nil? (:onyx/group-by-fn task-map))))))

(def ^{:private true} partial-input-output-task
  {:onyx/plugin NamespacedKeyword
   :onyx/medium s/Keyword
   (s/optional-key :onyx/fn) NamespacedKeyword})

(def ^{:private true} partial-fn-task
  {:onyx/fn NamespacedKeyword})

(def TaskMap
  (s/conditional #(or (= (:onyx/type %) :input) (= (:onyx/type %) :output))
                      (merge base-task-map partial-input-output-task)
                      grouping-task?
                      (merge base-task-map partial-fn-task partial-grouping-task)
                      #(= (:onyx/type %) :function)
                      (merge base-task-map partial-fn-task)))

(def Catalog
  [TaskMap])

(def Lifecycle
  {:lifecycle/task s/Keyword
   :lifecycle/calls NamespacedKeyword
   (s/optional-key :lifecycle/doc) s/Str
   s/Any s/Any})

(def LifecycleCall
  {(s/optional-key :lifecycle/doc) s/Str
   (s/optional-key :lifecycle/start-task?) Function
   (s/optional-key :lifecycle/before-task-start) Function
   (s/optional-key :lifecycle/before-batch) Function
   (s/optional-key :lifecycle/after-read-batch) Function
   (s/optional-key :lifecycle/after-batch) Function
   (s/optional-key :lifecycle/after-task-stop) Function
   (s/optional-key :lifecycle/after-ack-segment) Function
   (s/optional-key :lifecycle/after-retry-segment) Function})

(def FlowCondition
  {:flow/from s/Keyword
   :flow/to (s/either s/Keyword [s/Keyword])
   (s/optional-key :flow/short-circuit?) s/Bool
   (s/optional-key :flow/exclude-keys) [s/Keyword]
   (s/optional-key :flow/doc) s/Str
   (s/optional-key :flow/params) [s/Keyword]
   :flow/predicate (s/either s/Keyword [s/Any])
   s/Keyword s/Any})

(def Unit
  [(s/one s/Int "unit-count")
   (s/one s/Keyword "unit-type")])

(def WindowType
  (s/enum :fixed :sliding :global :session))

(def Window
  {:window/id s/Keyword
   :window/task s/Keyword
   :window/type WindowType
   :window/window-key s/Any
   :window/aggregation (s/either s/Keyword [s/Keyword])
   (s/optional-key :window/range) Unit
   (s/optional-key :window/slide) Unit
   (s/optional-key :window/timeout-gap) Unit
   (s/optional-key :window/session-key) s/Keyword
   (s/optional-key :window/doc) s/Str
   s/Keyword s/Any})

(def TriggerRefinement
  (s/enum :accumulating :discarding))

(def Trigger
  {:trigger/window-id s/Keyword
   :trigger/refinement TriggerRefinement
   :trigger/on s/Keyword
   :trigger/sync s/Keyword
   (s/optional-key :trigger/fire-all-extents?) s/Bool
   (s/optional-key :trigger/doc) s/Str
   s/Keyword s/Any})

(def Job
  {:catalog Catalog
   :workflow Workflow
   :task-scheduler TaskScheduler
   (s/optional-key :percentage) s/Int
   (s/optional-key :flow-conditions) [FlowCondition]
   (s/optional-key :windows) [Window]
   (s/optional-key :triggers) [Trigger]
   (s/optional-key :lifecycles) [Lifecycle]
   (s/optional-key :acker/percentage) s/Int
   (s/optional-key :acker/exempt-input-tasks?) s/Bool
   (s/optional-key :acker/exempt-output-tasks?) s/Bool
   (s/optional-key :acker/exempt-tasks) [s/Keyword]})

(def ClusterId
  (s/either s/Uuid s/Str))

(def EnvConfig
  {:zookeeper/address s/Str
   :onyx/id ClusterId
   (s/optional-key :zookeeper/server?) s/Bool
   (s/optional-key :zookeeper.server/port) s/Int
   (s/optional-key :bookkeeper/server?) s/Bool
   (s/optional-key :bookkeeper/local-quorum?) s/Bool
   (s/optional-key :bookkeeper/base-dir) s/Str
   s/Keyword s/Any})

(def AeronIdleStrategy
  (s/enum :busy-spin :low-restart-latency :high-restart-latency))

(def JobScheduler
  s/Keyword)

(def TaskScheduler
  s/Keyword)

(def Messaging
  (s/enum :aeron :dummy-messenger))

(def StateLogImpl
  (s/enum :bookkeeper :none))

(def StateFilterImpl
  (s/enum :set :rocksdb))

(def PeerConfig
  {:zookeeper/address s/Str
   :onyx/id ClusterId
   :onyx.peer/job-scheduler JobScheduler
   :onyx.messaging/impl Messaging
   :onyx.messaging/bind-addr s/Str
   (s/optional-key :onyx.messaging/peer-port) s/Int
   (s/optional-key :onyx.messaging/external-addr) s/Str<commit_msg>Code ordering. Fixes unresolved var.<commit_after>(ns onyx.schema
  (:require [schema.core :as s]))

(def NamespacedKeyword
  (s/pred (fn [kw]
                 (and (keyword? kw)
                      (namespace kw)))
               'keyword-namespaced?))

(def Function
  (s/pred fn? 'fn?))

(def TaskName
  (s/pred (fn [v]
                 (and (not= :all v)
                      (not= :none v)
                      (keyword? v)))
               'task-name?))

(defn ^{:private true} edge-two-nodes? [edge]
  (= (count edge) 2))

(def ^{:private true} edge-validator
  (s/->Both [(s/pred vector? 'vector?)
                  (s/pred edge-two-nodes? 'edge-two-nodes?)
                  [TaskName]]))

(def Workflow
  (s/->Both [(s/pred vector? 'vector?)
                  [edge-validator]]))

(def Language
  (s/enum :java :clojure))

(def ^{:private true} base-task-map
  {:onyx/name TaskName
   :onyx/type (s/enum :input :output :function)
   :onyx/batch-size (s/pred pos? 'pos?)
   (s/optional-key :onyx/uniqueness-key) s/Any
   (s/optional-key :onyx/restart-pred-fn) s/Keyword
   (s/optional-key :onyx/language) Language
   (s/optional-key :onyx/batch-timeout) (s/pred pos? 'pos?)
   (s/optional-key :onyx/doc) s/Str
   (s/optional-key :onyx/max-peers) (s/pred pos? 'pos?)
   s/Keyword s/Any})

(def FluxPolicy 
  (s/enum :continue :kill :recover))

(def ^{:private true} partial-grouping-task
  {(s/optional-key :onyx/group-by-key) s/Any
   (s/optional-key :onyx/group-by-fn) NamespacedKeyword
   :onyx/min-peers s/Int
   :onyx/flux-policy FluxPolicy})

(defn grouping-task? [task-map]
  (and (#{:function :output} (:onyx/type task-map))
       (or (not (nil? (:onyx/group-by-key task-map)))
           (not (nil? (:onyx/group-by-fn task-map))))))

(def ^{:private true} partial-input-output-task
  {:onyx/plugin NamespacedKeyword
   :onyx/medium s/Keyword
   (s/optional-key :onyx/fn) NamespacedKeyword})

(def ^{:private true} partial-fn-task
  {:onyx/fn NamespacedKeyword})

(def TaskMap
  (s/conditional #(or (= (:onyx/type %) :input) (= (:onyx/type %) :output))
                      (merge base-task-map partial-input-output-task)
                      grouping-task?
                      (merge base-task-map partial-fn-task partial-grouping-task)
                      #(= (:onyx/type %) :function)
                      (merge base-task-map partial-fn-task)))

(def Catalog
  [TaskMap])

(def Lifecycle
  {:lifecycle/task s/Keyword
   :lifecycle/calls NamespacedKeyword
   (s/optional-key :lifecycle/doc) s/Str
   s/Any s/Any})

(def LifecycleCall
  {(s/optional-key :lifecycle/doc) s/Str
   (s/optional-key :lifecycle/start-task?) Function
   (s/optional-key :lifecycle/before-task-start) Function
   (s/optional-key :lifecycle/before-batch) Function
   (s/optional-key :lifecycle/after-read-batch) Function
   (s/optional-key :lifecycle/after-batch) Function
   (s/optional-key :lifecycle/after-task-stop) Function
   (s/optional-key :lifecycle/after-ack-segment) Function
   (s/optional-key :lifecycle/after-retry-segment) Function})

(def FlowCondition
  {:flow/from s/Keyword
   :flow/to (s/either s/Keyword [s/Keyword])
   (s/optional-key :flow/short-circuit?) s/Bool
   (s/optional-key :flow/exclude-keys) [s/Keyword]
   (s/optional-key :flow/doc) s/Str
   (s/optional-key :flow/params) [s/Keyword]
   :flow/predicate (s/either s/Keyword [s/Any])
   s/Keyword s/Any})

(def Unit
  [(s/one s/Int "unit-count")
   (s/one s/Keyword "unit-type")])

(def WindowType
  (s/enum :fixed :sliding :global :session))

(def Window
  {:window/id s/Keyword
   :window/task s/Keyword
   :window/type WindowType
   :window/window-key s/Any
   :window/aggregation (s/either s/Keyword [s/Keyword])
   (s/optional-key :window/range) Unit
   (s/optional-key :window/slide) Unit
   (s/optional-key :window/timeout-gap) Unit
   (s/optional-key :window/session-key) s/Keyword
   (s/optional-key :window/doc) s/Str
   s/Keyword s/Any})

(def TriggerRefinement
  (s/enum :accumulating :discarding))

(def Trigger
  {:trigger/window-id s/Keyword
   :trigger/refinement TriggerRefinement
   :trigger/on s/Keyword
   :trigger/sync s/Keyword
   (s/optional-key :trigger/fire-all-extents?) s/Bool
   (s/optional-key :trigger/doc) s/Str
   s/Keyword s/Any})

(def JobScheduler
  s/Keyword)

(def TaskScheduler
  s/Keyword)

(def Job
  {:catalog Catalog
   :workflow Workflow
   :task-scheduler TaskScheduler
   (s/optional-key :percentage) s/Int
   (s/optional-key :flow-conditions) [FlowCondition]
   (s/optional-key :windows) [Window]
   (s/optional-key :triggers) [Trigger]
   (s/optional-key :lifecycles) [Lifecycle]
   (s/optional-key :acker/percentage) s/Int
   (s/optional-key :acker/exempt-input-tasks?) s/Bool
   (s/optional-key :acker/exempt-output-tasks?) s/Bool
   (s/optional-key :acker/exempt-tasks) [s/Keyword]})

(def ClusterId
  (s/either s/Uuid s/Str))

(def EnvConfig
  {:zookeeper/address s/Str
   :onyx/id ClusterId
   (s/optional-key :zookeeper/server?) s/Bool
   (s/optional-key :zookeeper.server/port) s/Int
   (s/optional-key :bookkeeper/server?) s/Bool
   (s/optional-key :bookkeeper/local-quorum?) s/Bool
   (s/optional-key :bookkeeper/base-dir) s/Str
   s/Keyword s/Any})

(def AeronIdleStrategy
  (s/enum :busy-spin :low-restart-latency :high-restart-latency))

(def Messaging
  (s/enum :aeron :dummy-messenger))

(def StateLogImpl
  (s/enum :bookkeeper :none))

(def StateFilterImpl
  (s/enum :set :rocksdb))

(def PeerConfig
  {:zookeeper/address s/Str
   :onyx/id ClusterId
   :onyx.peer/job-scheduler JobScheduler
   :onyx.messaging/impl Messaging
   :onyx.messaging/bind-addr s/Str
   (s/optional-key :onyx.messaging/peer-port) s/Int
   (s/optional-key :onyx.messaging/external-addr) s/Str
{'fraction_non_alphanumeric': 0.15987378385485143, 'fraction_numerical': 0.00017530020159523183, 'mean_word_length': 2.968695652173913, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '2806162', 'n_tokens_mistral': 4328, 'n_tokens_neox': 4268, 'n_words': 861}