<commit_before>(ns onyx.plugin.input-log-test
  (:require [aero.core :refer [read-config]]
            [clojure.test :refer [deftest is testing]]
            [datomic.api :as d]
            [onyx api
             [job :refer [add-task]]
             [test-helper :refer [with-test-env]]]
            [onyx.plugin datomic
             [core-async :refer [take-segments! get-core-async-channels]]]
            [onyx.tasks
             [datomic :refer [read-log]]
             [core-async :as core-async]]))

(defn build-job [db-uri log-end-tx batch-size batch-timeout]
  (let [batch-settings {:onyx/batch-size batch-size :onyx/batch-timeout batch-timeout}
        base-job (merge {:workflow [[:read-log :persist]]
                         :catalog []
                         :lifecycles []
                         :windows []
                         :triggers []
                         :flow-conditions []
                         :task-scheduler :onyx.task-scheduler/balanced})]
    (-> base-job
        (add-task (read-log :read-log
                            (merge {:datomic/uri db-uri
                                    :checkpoint/key "checkpoint"
                                    :checkpoint/force-reset? false
                                    :onyx/max-peers 1
                                    :datomic/log-end-tx log-end-tx}
                                   batch-settings)))
        (add-task (core-async/output :persist batch-settings 100000)))))

(defn ensure-datomic!
  ([db-uri data]
   (d/create-database db-uri)
   @(d/transact
     (d/connect db-uri)
     data)))

(def schema
  [{:db/id #db/id [:db.part/db]
    :db/ident :com.mdrogalis/people
    :db.install/_partition :db.part/db}

   {:db/id #db/id [:db.part/db]
    :db/ident :user/name
    :db/valueType :db.type/string
    :db/cardinality :db.cardinality/one
    :db.install/_attribute :db.part/db}])

(def people
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Derek"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Kristen"}])

(def people2
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike2"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene2"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti2"}])

(def people3
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike3"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene3"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti3"}])

(def people4
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike4"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene4"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti4"}])

(deftest datomic-input-log-test
  (let [{:keys [env-config peer-config datomic-config]}
        (read-config (clojure.java.io/resource "config.edn") {:profile :test})
        db-uri (str "datomic:mem://" (java.util.UUID/randomUUID))]
    (try
      (with-test-env [test-env [4 env-config peer-config]]
        (testing "That we can read the initial transaction log"
          (let [job (build-job db-uri 1002 10 1000)
                {:keys [persist]} (get-core-async-channels job)
                _ (mapv (partial ensure-datomic! db-uri) [schema people])
                job-id (:job-id (onyx.api/submit-job peer-config job))]
            (ensure-datomic! db-uri people2)
            (onyx.api/await-job-completion peer-config job-id)
            (is (= [{:data '([63 10 :com.mdrogalis/people 13194139534312 true]
                             [0 11 63 13194139534312 true]
                             [64 10 :user/name 13194139534312 true]
                             [64 40 23 13194139534312 true]
                             [64 41 35 13194139534312 true]
                             [0 13 64 13194139534312 true]) :t 1000}
                    {:data '([277076930200554 64 "Mike" 13194139534313 true]
                             [277076930200555 64 "Dorrene" 13194139534313 true]
                             [277076930200556 64 "Benti" 13194139534313 true]
                             [277076930200557 64 "Derek" 13194139534313 true]
                             [277076930200558 64 "Kristen" 13194139534313 true]) :t 1001}]
                   (map (fn [result]
                          (-> result
                              (update :data rest)
                              (dissoc :id)))
                        (take-segments! persist 50))))
            ;; recover from resume point, should only get the latest txes
            (let [new-job (build-job db-uri 1014 10 1000)
                  job2 (->> job-id
                            (onyx.api/job-snapshot-coordinates peer-config (:onyx/tenancy-id peer-config))<commit_msg>Show that read-log is inclusive of end-tx<commit_after>(ns onyx.plugin.input-log-test
  (:require [aero.core :refer [read-config]]
            [clojure.test :refer [deftest is testing]]
            [datomic.api :as d]
            [onyx api
             [job :refer [add-task]]
             [test-helper :refer [with-test-env]]]
            [onyx.plugin datomic
             [core-async :refer [take-segments! get-core-async-channels]]]
            [onyx.tasks
             [datomic :refer [read-log]]
             [core-async :as core-async]]))

(defn build-job [db-uri log-end-tx batch-size batch-timeout]
  (let [batch-settings {:onyx/batch-size batch-size :onyx/batch-timeout batch-timeout}
        base-job (merge {:workflow [[:read-log :persist]]
                         :catalog []
                         :lifecycles []
                         :windows []
                         :triggers []
                         :flow-conditions []
                         :task-scheduler :onyx.task-scheduler/balanced})]
    (-> base-job
        (add-task (read-log :read-log
                            (merge {:datomic/uri db-uri
                                    :checkpoint/key "checkpoint"
                                    :checkpoint/force-reset? false
                                    :onyx/max-peers 1
                                    :datomic/log-end-tx log-end-tx}
                                   batch-settings)))
        (add-task (core-async/output :persist batch-settings 100000)))))

(defn ensure-datomic!
  ([db-uri data]
   (d/create-database db-uri)
   @(d/transact
     (d/connect db-uri)
     data)))

(def schema
  [{:db/id #db/id [:db.part/db]
    :db/ident :com.mdrogalis/people
    :db.install/_partition :db.part/db}

   {:db/id #db/id [:db.part/db]
    :db/ident :user/name
    :db/valueType :db.type/string
    :db/cardinality :db.cardinality/one
    :db.install/_attribute :db.part/db}])

(def people
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Derek"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Kristen"}])

(def people2
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike2"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene2"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti2"}])

(def people3
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike3"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene3"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti3"}])

(def people4
  [{:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Mike4"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Dorrene4"}
   {:db/id (d/tempid :com.mdrogalis/people)
    :user/name "Benti4"}])

(deftest datomic-input-log-test
  (let [{:keys [env-config peer-config datomic-config]}
        (read-config (clojure.java.io/resource "config.edn") {:profile :test})
        db-uri (str "datomic:mem://" (java.util.UUID/randomUUID))]
    (try
      (with-test-env [test-env [4 env-config peer-config]]
        (testing "That we can read the initial transaction log"
          (let [job (build-job db-uri 1001 10 1000)
                {:keys [persist]} (get-core-async-channels job)
                _ (mapv (partial ensure-datomic! db-uri) [schema people])
                job-id (:job-id (onyx.api/submit-job peer-config job))]
            (ensure-datomic! db-uri people2)
            (onyx.api/await-job-completion peer-config job-id)
            (is (= [{:data '([63 10 :com.mdrogalis/people 13194139534312 true]
                             [0 11 63 13194139534312 true]
                             [64 10 :user/name 13194139534312 true]
                             [64 40 23 13194139534312 true]
                             [64 41 35 13194139534312 true]
                             [0 13 64 13194139534312 true]) :t 1000}
                    {:data '([277076930200554 64 "Mike" 13194139534313 true]
                             [277076930200555 64 "Dorrene" 13194139534313 true]
                             [277076930200556 64 "Benti" 13194139534313 true]
                             [277076930200557 64 "Derek" 13194139534313 true]
                             [277076930200558 64 "Kristen" 13194139534313 true]) :t 1001}]
                   (map (fn [result]
                          (-> result
                              (update :data rest)
                              (dissoc :id)))
                        (take-segments! persist 50))))
            ;; recover from resume point, should only get the latest txes
            (let [new-job (build-job db-uri 1014 10 1000)
                  job2 (->> job-id
                            (onyx.api/job-snapshot-coordinates peer-config (:onyx/tenancy-id peer-config))
{'fraction_non_alphanumeric': 0.1377233487251556, 'fraction_numerical': 0.06444489058422004, 'mean_word_length': 1.6746308724832215, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '956354', 'n_tokens_mistral': 3837, 'n_tokens_neox': 3334, 'n_words': 693}