metabase.driver.mysql

(ns metabase.driver.mysql
  (:require
   [clojure.java.io :as jio]
   [clojure.java.jdbc :as jdbc]
   [clojure.set :as set]
   [clojure.string :as str]
   [clojure.walk :as walk]
   [honey.sql :as sql]
   [java-time.api :as t]
   [medley.core :as m]
   [metabase.config :as config]
   [metabase.db :as mdb]
   [metabase.driver :as driver]
   [metabase.driver.common :as driver.common]
   [metabase.driver.mysql.actions :as mysql.actions]
   [metabase.driver.mysql.ddl :as mysql.ddl]
   [metabase.driver.sql :as driver.sql]
   [metabase.driver.sql-jdbc.common :as sql-jdbc.common]
   [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
   [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
   [metabase.driver.sql-jdbc.quoting :refer [quote-columns]]
   [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
   [metabase.driver.sql.query-processor :as sql.qp]
   [metabase.driver.sql.query-processor.util :as sql.qp.u]
   [metabase.driver.sql.util :as sql.u]
   [metabase.lib.field :as lib.field]
   [metabase.lib.metadata :as lib.metadata]
   [metabase.query-processor.store :as qp.store]
   [metabase.query-processor.timezone :as qp.timezone]
   [metabase.query-processor.util.add-alias-info :as add]
   [metabase.upload :as upload]
   [metabase.util :as u]
   [metabase.util.honey-sql-2 :as h2x]
   [metabase.util.i18n :refer [deferred-tru]]
   [metabase.util.log :as log])
  (:import
   (java.io File)
   (java.sql DatabaseMetaData ResultSet ResultSetMetaData Types)
   (java.time LocalDateTime OffsetDateTime OffsetTime ZonedDateTime ZoneOffset)
   (java.time.format DateTimeFormatter)))

(set! *warn-on-reflection* true)

(comment
  ;; method impls live in these namespaces.
  mysql.actions/keep-me
  mysql.ddl/keep-me)

(driver/register! :mysql, :parent :sql-jdbc)

(def ^:private ^:const min-supported-mysql-version 5.7)
(def ^:private ^:const min-supported-mariadb-version 10.2)

(defmethod driver/display-name :mysql [_] "MySQL")

(doseq [[feature supported?] {;; MySQL LIKE clauses are case-sensitive or not based on whether the collation of the
                              ;; server and the columns themselves. Since this isn't something we can really change in
                              ;; the query itself don't present the option to the users in the UI
                              :case-sensitivity-string-filter-options false
                              :describe-fields                        true
                              :describe-fks                           true
                              :convert-timezone                       true
                              :datetime-diff                          true
                              :full-join                              false
                              :index-info                             true
                              :now                                    true
                              :percentile-aggregations                false
                              :persist-models                         true
                              :schemas                                false
                              :uploads                                true
                              :identifiers-with-spaces                true
                              ;; MySQL doesn't let you have lag/lead in the same part of a query as a `GROUP BY`; to
                              ;; fully support `offset` we need to do some kooky query transformations just for MySQL
                              ;; and make this work.
                              :window-functions/offset                false}]
  (defmethod driver/database-supports? [:mysql feature] [_driver _feature _db] supported?))

(defmethod driver/database-supports? [:mysql :nested-field-columns] [_driver _feat db]
  (driver.common/json-unfolding-default db))

(doseq [feature [:actions :actions/custom]]
  (defmethod driver/database-supports? [:mysql feature]
    [driver _feat _db]
    ;; Only supported for MySQL right now. Revise when a child driver is added.
    (= driver :mysql)))

(defn mariadb?
  [database]
  (-> database :dbms_version :flavor (= "MariaDB")))

(defn mariadb-connection?
  [driver conn]
  (->> conn (sql-jdbc.sync/dbms-version driver) :flavor (= "MariaDB")))

(defmethod driver/database-supports? [:mysql :table-privileges]
  [_driver _feat _db]
  ;; Disabled completely due to errors when dealing with partial revokes (metabase#38499)
  false
  #_(and (= driver :mysql) (not (mariadb? db))))

(defn- db-version [^DatabaseMetaData metadata]
  (Double/parseDouble
   (format "%d.%d" (.getDatabaseMajorVersion metadata) (.getDatabaseMinorVersion metadata))))

(defn- unsupported-version? [^DatabaseMetaData metadata]
  (let [mariadb? (= (.getDatabaseProductName metadata) "MariaDB")]
    (< (db-version metadata)
       (if mariadb?
         min-supported-mariadb-version
         min-supported-mysql-version))))

(defn- warn-on-unsupported-versions [driver details]
  (sql-jdbc.conn/with-connection-spec-for-testing-connection [jdbc-spec [driver details]]
    (sql-jdbc.execute/do-with-connection-with-options
     driver
     jdbc-spec
     nil
     (fn [^java.sql.Connection conn]
       (when (unsupported-version? (.getMetaData conn))
         (log/warn
          (u/format-color :red
                          (str "\n\n********************************************************************************\n"
                               (format
                                "WARNING: Metabase only officially supports MySQL %s/MariaDB %s and above."
                                min-supported-mysql-version
                                min-supported-mariadb-version)
                               "\n"
                               "All Metabase features may not work properly when using an unsupported version."
                               "\n********************************************************************************\n"))))))))

(defmethod driver/can-connect? :mysql
  [driver details]
  ;; delegate to parent method to check whether we can connect; if so, check if it's an unsupported version and issue
  ;; a warning if it is
  (when ((get-method driver/can-connect? :sql-jdbc) driver details)
    (warn-on-unsupported-versions driver details)
    true))

(declare table-names->privileges)
(declare privilege-grants-for-user)

(defmethod sql-jdbc.sync/current-user-table-privileges :mysql
  [driver conn & {:as _options}]
  ;; MariaDB doesn't allow users to query the privileges of roles a user might have (unless they have select privileges
  ;; for the mysql database), so we can't query the full privileges of the current user.
  (when-not (mariadb-connection? driver conn)
    (let [sql->tuples (fn [sql] (drop 1 (jdbc/query conn sql {:as-arrays? true})))
          db-name     (ffirst (sql->tuples "SELECT DATABASE()"))
          table-names (map first (sql->tuples "SHOW TABLES"))]
      (for [[table-name privileges] (table-names->privileges (privilege-grants-for-user conn "CURRENT_USER()")
                                                             db-name
                                                             table-names)]
        {:role   nil
         :schema nil
         :table  table-name
         :select (contains? privileges :select)
         :update (contains? privileges :update)
         :insert (contains? privileges :insert)
         :delete (contains? privileges :delete)}))))

(def default-ssl-cert-details
  {:name         "ssl-cert"
   :display-name (deferred-tru "Server SSL certificate chain")
   :placeholder  ""
   :visible-if   {"ssl" true}})

(defmethod driver/connection-properties :mysql
  [_]
  (->>
   [driver.common/default-host-details
    (assoc driver.common/default-port-details :placeholder 3306)
    driver.common/default-dbname-details
    driver.common/default-user-details
    driver.common/default-password-details
    driver.common/cloud-ip-address-info
    driver.common/default-ssl-details
    default-ssl-cert-details
    driver.common/ssh-tunnel-preferences
    driver.common/advanced-options-start
    driver.common/json-unfolding
    (assoc driver.common/additional-options
           :placeholder  "tinyInt1isBit=false")
    driver.common/default-advanced-options]
   (map u/one-or-many)
   (apply concat)))

(defmethod sql.qp/add-interval-honeysql-form :mysql
  [driver hsql-form amount unit]
  ;; MySQL doesn't support `:millisecond` as an option, but does support fractional seconds
  (if (= unit :millisecond)
    (recur driver hsql-form (/ amount 1000.0) :second)
    [:date_add hsql-form [:raw (format "INTERVAL %s %s" amount (name unit))]]))

(defmethod sql.qp/current-datetime-honeysql-form :mysql
  [_]
  (h2x/with-database-type-info [:now [:inline 6]] "timestamp"))

(defmethod driver/humanize-connection-error-message :mysql
  [_ message]
  (condp re-matches message
    #"^Communications link failure\s+The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server.$"
    :cannot-connect-check-host-and-port

    #"^Unknown database .*$"
    :database-name-incorrect

    #"Access denied for user.*$"
    :username-or-password-incorrect

    #"Must specify port after ':' in connection string"
    :invalid-hostname

    ;; else
    message))

#_{:clj-kondo/ignore [:deprecated-var]}
(defmethod sql-jdbc.sync/db-default-timezone :mysql
  [_ spec]
  (let [sql                                    (str "SELECT @@GLOBAL.time_zone AS global_tz,"
                                                    " @@system_time_zone AS system_tz,"
                                                    " time_format("
                                                    "   timediff("
                                                    "      now(), convert_tz(now(), @@GLOBAL.time_zone, '+00:00')"
                                                    "   ),"
                                                    "   '%H:%i'"
                                                    " ) AS 'offset';")
        [{:keys [global_tz system_tz offset]}] (jdbc/query spec sql)
        the-valid-id                           (fn [zone-id]
                                                 (when zone-id
                                                   (try
                                                     (.getId (t/zone-id zone-id))
                                                     (catch Throwable _))))]
    (or
     ;; if global timezone ID is 'SYSTEM', then try to use the system timezone ID
     (when (= global_tz "SYSTEM")
       (the-valid-id system_tz))
     ;; otherwise try to use the global ID
     (the-valid-id global_tz)
     ;; failing that, calculate the offset between now in the global timezone and now in UTC. Non-negative offsets
     ;; don't come back with `+` so add that if needed
     (if (str/starts-with? offset "-")
       offset
       (str \+ offset)))))

(defmethod driver/db-start-of-week :mysql
  [_]
  :sunday)

(defmethod driver.sql/json-field-length :mysql
  [_ json-field-identifier]
  [:length [:cast json-field-identifier :char]])

(defmethod sql.qp/unix-timestamp->honeysql [:mysql :seconds] [_ _ expr]
  [:from_unixtime expr])

(defmethod sql.qp/cast-temporal-string [:mysql :Coercion/ISO8601->DateTime]
  [_driver _coercion-strategy expr]
  (h2x/->datetime expr))

(defmethod sql.qp/cast-temporal-string [:mysql :Coercion/YYYYMMDDHHMMSSString->Temporal]
  [_driver _coercion-strategy expr]
  [:convert expr [:raw "DATETIME"]])

(defmethod sql.qp/cast-temporal-byte [:mysql :Coercion/YYYYMMDDHHMMSSBytes->Temporal]
  [driver _coercion-strategy expr]
  (sql.qp/cast-temporal-string driver :Coercion/YYYYMMDDHHMMSSString->Temporal expr))

(defn- date-format [format-str expr]
  [:date_format expr (h2x/literal format-str)])

(defn- str-to-date
  [format-str expr]
  (let [contains-date-parts? (some #(str/includes? format-str %)
                                   ["%a" "%b" "%c" "%D" "%d" "%e" "%j" "%M" "%m" "%U"
                                    "%u" "%V" "%v" "%W" "%w" "%X" "%x" "%Y" "%y"])
        contains-time-parts? (some #(str/includes? format-str %)
                                   ["%f" "%H" "%h" "%I" "%i" "%k" "%l" "%p" "%r" "%S" "%s" "%T"])
        database-type        (cond
                               (and contains-date-parts? (not contains-time-parts?)) "date"
                               (and contains-time-parts? (not contains-date-parts?)) "time"
                               :else                                                 "datetime")]
    (-> [:str_to_date expr (h2x/literal format-str)]
        (h2x/with-database-type-info database-type))))

(defmethod sql.qp/->float :mysql
  [_ value]
  ;; no-op as MySQL doesn't support cast to float
  value)

(defmethod sql.qp/->integer :mysql
  [_ value]
  (h2x/maybe-cast :signed value))

(defmethod sql.qp/->honeysql [:mysql :regex-match-first]
  [driver [_ arg pattern]]
  [:regexp_substr (sql.qp/->honeysql driver arg) (sql.qp/->honeysql driver pattern)])

(defmethod sql.qp/->honeysql [:mysql :length]
  [driver [_ arg]]
  [:char_length (sql.qp/->honeysql driver arg)])

(def ^:private database-type->mysql-cast-type-name
  {"integer"          "signed"
   "text"             "char"
   "double precision" "double"
   "bigint"           "unsigned"})

(defmethod sql.qp/json-query :mysql
  [_driver unwrapped-identifier stored-field]
  {:pre [(h2x/identifier? unwrapped-identifier)]}
  (letfn [(handle-name [x] (str "\"" (if (number? x) (str x) (name x)) "\""))]
    (let [field-type            (:database-type stored-field)
          field-type            (get database-type->mysql-cast-type-name field-type field-type)
          nfc-path              (:nfc-path stored-field)
          parent-identifier     (sql.qp.u/nfc-field->parent-identifier unwrapped-identifier stored-field)
          jsonpath-query        (format "$.%s" (str/join "." (map handle-name (rest nfc-path))))
          json-extract+jsonpath [:json_extract parent-identifier jsonpath-query]]
      (case (u/lower-case-en field-type)
        ;; If we see JSON datetimes we expect them to be in ISO8601. However, MySQL expects them as something different.
        ;; We explicitly tell MySQL to go and accept ISO8601, because that is JSON datetimes, although there is no real standard for JSON, ISO8601 is the de facto standard.
        "timestamp" [:convert
                     [:str_to_date json-extract+jsonpath "\"%Y-%m-%dT%T.%fZ\""]
                     [:raw "DATETIME"]]

        "boolean" json-extract+jsonpath

        ;; in older versions of MySQL you can't do `convert(<string>, double)` or `cast(<string> AS double)` which is
        ;; equivalent; instead you can do `<string> + 0.0` =(
        ("float" "double") [:+ json-extract+jsonpath [:inline 0.0]]

        [:convert json-extract+jsonpath [:raw (u/upper-case-en field-type)]]))))

(defmethod sql.qp/->honeysql [:mysql :field]
  [driver [_ id-or-name opts :as mbql-clause]]
  (let [stored-field  (when (integer? id-or-name)
                        (lib.metadata/field (qp.store/metadata-provider) id-or-name))
        parent-method (get-method sql.qp/->honeysql [:sql :field])
        honeysql-expr (parent-method driver mbql-clause)]
    (cond
      (not (lib.field/json-field? stored-field))
      honeysql-expr

      (::sql.qp/forced-alias opts)
      (keyword (::add/source-alias opts))

      :else
      (walk/postwalk #(if (h2x/identifier? %)
                        (sql.qp/json-query :mysql % stored-field)
                        %)
                     honeysql-expr))))

(defn- trunc-with-format [format-str expr]
  (str-to-date format-str (date-format format-str (h2x/->datetime expr))))

(defn- ->date [expr]
  (if (h2x/is-of-type? expr "date")
    expr
    (-> [:date expr]
        (h2x/with-database-type-info "date"))))

(defn make-date
  [year-expr number-of-days]
  (-> [:makedate year-expr (sql.qp/inline-num number-of-days)]
      (h2x/with-database-type-info "date")))

(defmethod sql.qp/date [:mysql :minute]
  [_driver _unit expr]
  (let [format-str (if (= (h2x/database-type expr) "time")
                     "%H:%i"
                     "%Y-%m-%d %H:%i")]
    (trunc-with-format format-str expr)))

(defmethod sql.qp/date [:mysql :hour]
  [_driver _unit expr]
  (let [format-str (if (= (h2x/database-type expr) "time")
                     "%H"
                     "%Y-%m-%d %H")]
    (trunc-with-format format-str expr)))

(defmethod sql.qp/date [:mysql :default]         [_ _ expr] expr)
(defmethod sql.qp/date [:mysql :minute-of-hour]  [_ _ expr] (h2x/minute expr))
(defmethod sql.qp/date [:mysql :hour-of-day]     [_ _ expr] (h2x/hour expr))
(defmethod sql.qp/date [:mysql :day]             [_ _ expr] (->date expr))
(defmethod sql.qp/date [:mysql :day-of-month]    [_ _ expr] [:dayofmonth expr])
(defmethod sql.qp/date [:mysql :day-of-year]     [_ _ expr] [:dayofyear expr])
(defmethod sql.qp/date [:mysql :month-of-year]   [_ _ expr] (h2x/month expr))
(defmethod sql.qp/date [:mysql :quarter-of-year] [_ _ expr] (h2x/quarter expr))
(defmethod sql.qp/date [:mysql :year]            [_ _ expr] (make-date (h2x/year expr) 1))

(defmethod sql.qp/date [:mysql :day-of-week]
  [driver _unit expr]
  (sql.qp/adjust-day-of-week driver [:dayofweek expr]))

(defmethod sql.qp/date [:mysql :week] [_ _ expr]
  (let [extract-week-fn (fn [expr]
                          (str-to-date "%X%V %W"
                                       (h2x/concat [:yearweek expr]
                                                   (h2x/literal " Sunday"))))]
    (sql.qp/adjust-start-of-week :mysql extract-week-fn expr)))

(defmethod sql.qp/date [:mysql :week-of-year-iso] [_ _ expr] (h2x/week expr 3))

(defmethod sql.qp/date [:mysql :month] [_ _ expr]
  (str-to-date "%Y-%m-%d"
               (h2x/concat (date-format "%Y-%m" expr)
                           (h2x/literal "-01"))))

(defmethod sql.qp/date [:mysql :quarter] [_ _ expr]
  (str-to-date "%Y-%m-%d"
               (h2x/concat (h2x/year expr)
                           (h2x/literal "-")
                           (h2x/- (h2x/* (h2x/quarter expr)
                                         3)
                                  2)
                           (h2x/literal "-01"))))

(defmethod sql.qp/->honeysql [:mysql :convert-timezone]
  [driver [_ arg target-timezone source-timezone]]
  (let [expr       (sql.qp/->honeysql driver arg)
        timestamp? (h2x/is-of-type? expr "timestamp")]
    (sql.u/validate-convert-timezone-args timestamp? target-timezone source-timezone)
    (h2x/with-database-type-info
     [:convert_tz expr (or source-timezone (qp.timezone/results-timezone-id)) target-timezone]
     "datetime")))

(defn- timestampdiff-dates [unit x y]
  [:timestampdiff [:raw (name unit)] (h2x/->date x) (h2x/->date y)])

(defn- timestampdiff [unit x y]
  [:timestampdiff [:raw (name unit)] x y])

(defmethod sql.qp/datetime-diff [:mysql :year]    [_driver _unit x y] (timestampdiff-dates :year x y))
(defmethod sql.qp/datetime-diff [:mysql :quarter] [_driver _unit x y] (timestampdiff-dates :quarter x y))
(defmethod sql.qp/datetime-diff [:mysql :month]   [_driver _unit x y] (timestampdiff-dates :month x y))
(defmethod sql.qp/datetime-diff [:mysql :week]    [_driver _unit x y] (timestampdiff-dates :week x y))
(defmethod sql.qp/datetime-diff [:mysql :day]     [_driver _unit x y] [:datediff y x])
(defmethod sql.qp/datetime-diff [:mysql :hour]    [_driver _unit x y] (timestampdiff :hour x y))
(defmethod sql.qp/datetime-diff [:mysql :minute]  [_driver _unit x y] (timestampdiff :minute x y))
(defmethod sql.qp/datetime-diff [:mysql :second]  [_driver _unit x y] (timestampdiff :second x y))

(defmethod sql-jdbc.sync/database-type->base-type :mysql
  [_ database-type]
  ({:BIGINT     :type/BigInteger
    :BINARY     :type/*
    :BIT        :type/Boolean
    :BLOB       :type/*
    :CHAR       :type/Text
    :DATE       :type/Date
    :DATETIME   :type/DateTime
    :DECIMAL    :type/Decimal
    :DOUBLE     :type/Float
    :ENUM       :type/MySQLEnum
    :FLOAT      :type/Float
    :INT        :type/Integer
    :INTEGER    :type/Integer
    :LONGBLOB   :type/*
    :LONGTEXT   :type/Text
    :MEDIUMBLOB :type/*
    :MEDIUMINT  :type/Integer
    :MEDIUMTEXT :type/Text
    :NUMERIC    :type/Decimal
    :REAL       :type/Float
    :SET        :type/*
    :SMALLINT   :type/Integer
    :TEXT       :type/Text
    :TIME       :type/Time
    :TIMESTAMP  :type/DateTimeWithLocalTZ ; stored as UTC in the database
    :TINYBLOB   :type/*
    :TINYINT    :type/Integer
    :TINYTEXT   :type/Text
    :VARBINARY  :type/*
    :VARCHAR    :type/Text
    :YEAR       :type/Integer
    :JSON       :type/JSON}
   ;; strip off " UNSIGNED" from end if present
   (keyword (str/replace (name database-type) #"\sUNSIGNED$" ""))))

(defmethod sql-jdbc.sync/column->semantic-type :mysql
  [_ database-type _]
  ;; More types to be added when we start caring about them
  (case database-type
    "JSON"  :type/SerializedJSON
    nil))

(def ^:private default-connection-args
  {;; 0000-00-00 dates are valid in MySQL; convert these to `null` when they come back because they're illegal in Java
   :zeroDateTimeBehavior "convertToNull"
   ;; Force UTF-8 encoding of results
   :useUnicode           true
   :characterEncoding    "UTF8"
   :characterSetResults  "UTF8"
   ;; GZIP compress packets sent between Metabase server and MySQL/MariaDB database
   :useCompression       true
   ;; record transaction isolation level and auto-commit locally, and avoid hitting the DB if we do something like
   ;; `.setTransactionIsolation()` to something we previously set it to. Since we do this every time we run a
   ;; query (see [[metabase.driver.sql-jdbc.execute/set-best-transaction-level!]]) this should speed up things a bit by
   ;; removing that overhead. See also
   ;; https://dev.mysql.com/doc/connector-j/en/connector-j-connp-props-performance-extensions.html#cj-conn-prop_useLocalSessionState
   ;; and #44507
   :useLocalSessionState true})

(defn- maybe-add-program-name-option [jdbc-spec additional-options-map]
  ;; connectionAttributes (if multiple) are separated by commas, so values that contain spaces are OK, so long as they
  ;; don't contain a comma; our mb-version-and-process-identifier shouldn't contain one, but just to be on the safe side
  (let [set-prog-nm-fn (fn []
                         (let [prog-name (str/replace config/mb-version-and-process-identifier "," "_")]
                           (assoc jdbc-spec :connectionAttributes (str "program_name:" prog-name))))]
    (if-let [conn-attrs (get additional-options-map "connectionAttributes")]
      (if (str/includes? conn-attrs "program_name")
        jdbc-spec ; additional-options already includes the program_name; don't set it here
        (set-prog-nm-fn))
      (set-prog-nm-fn)))) ; additional-options did not contain connectionAttributes at all; set it

(defmethod sql-jdbc.conn/connection-details->spec :mysql
  [_ {ssl? :ssl, :keys [additional-options ssl-cert], :as details}]
  ;; In versions older than 0.32.0 the MySQL driver did not correctly save `ssl?` connection status. Users worked
  ;; around this by including `useSSL=true`. Check if that's there, and if it is, assume SSL status. See #9629
  ;;
  ;; TODO - should this be fixed by a data migration instead?
  (let [addl-opts-map (sql-jdbc.common/additional-options->map additional-options :url "=" false)
        ssl?          (or ssl? (= "true" (get addl-opts-map "useSSL")))
        ssl-cert?     (and ssl? (some? ssl-cert))]
    (when (and ssl? (not (contains? addl-opts-map "trustServerCertificate")))
      (log/info "You may need to add 'trustServerCertificate=true' to the additional connection options to connect with SSL."))
    (merge
     default-connection-args
     ;; newer versions of MySQL will complain if you don't specify this when not using SSL
     {:useSSL (boolean ssl?)}
     (let [details (-> (if ssl-cert? (set/rename-keys details {:ssl-cert :serverSslCert}) details)
                       (set/rename-keys {:dbname :db})
                       (dissoc :ssl))]
       (-> (mdb/spec :mysql details)
           (maybe-add-program-name-option addl-opts-map)
           (sql-jdbc.common/handle-additional-options details))))))

(defmethod sql-jdbc.sync/active-tables :mysql
  [& args]
  (apply sql-jdbc.sync/post-filtered-active-tables args))

(defmethod sql-jdbc.sync/excluded-schemas :mysql
  [_]
  #{"INFORMATION_SCHEMA"})

(defmethod sql.qp/quote-style :mysql [_] :mysql)

(defmethod sql-jdbc.execute/set-timezone-sql :mysql
  [_]
  "SET @@session.time_zone = %s;")

(defmethod sql-jdbc.execute/set-parameter [:mysql OffsetTime]
  [driver ps i t]
  ;; convert to a LocalTime so MySQL doesn't get F U S S Y
  (sql-jdbc.execute/set-parameter driver ps i (t/local-time (t/with-offset-same-instant t (t/zone-offset 0)))))

(defmethod sql-jdbc.execute/set-parameter [:mysql OffsetDateTime]
  [driver ^java.sql.PreparedStatement ps ^Integer i t]
  (let [zone   (t/zone-id (qp.timezone/results-timezone-id))
        offset (.. zone getRules (getOffset (t/instant t)))
        t      (t/local-date-time (t/with-offset-same-instant t offset))]
    (sql-jdbc.execute/set-parameter driver ps i t)))

(defmethod sql-jdbc.execute/read-column-thunk [:mysql Types/TIMESTAMP]
  [_ ^ResultSet rs ^ResultSetMetaData rsmeta ^Integer i]
  ;; Check and see if the column type is `TIMESTAMP` (as opposed to `DATETIME`, which is the equivalent of
  ;; LocalDateTime), and normalize it to a UTC timestamp if so.
  (if (= (.getColumnTypeName rsmeta i) "TIMESTAMP")
    (fn read-timestamp-thunk []
      (when-let [t (.getObject rs i LocalDateTime)]
        (t/with-offset-same-instant (t/offset-date-time t (t/zone-id (qp.timezone/results-timezone-id))) (t/zone-offset 0))))
    (fn read-datetime-thunk []
      (.getObject rs i LocalDateTime))))

(defmethod sql-jdbc.execute/read-column-thunk [:mysql Types/TIME]
  [driver ^ResultSet rs rsmeta ^Integer i]
  (let [parent-thunk ((get-method sql-jdbc.execute/read-column-thunk [:sql-jdbc Types/TIME]) driver rs rsmeta i)]
    (fn read-time-thunk []
      (try
        (parent-thunk)
        (catch Throwable _
          (.getString rs i))))))

(defmethod sql-jdbc.execute/read-column-thunk [:mysql Types/DATE]
  [driver ^ResultSet rs ^ResultSetMetaData rsmeta ^Integer i]
  (if (= "YEAR" (.getColumnTypeName rsmeta i))
    (fn read-time-thunk []
      (when-let [x (.getObject rs i)]
        (.getYear (.toLocalDate ^java.sql.Date x))))
    (let [parent-thunk ((get-method sql-jdbc.execute/read-column-thunk [:sql-jdbc Types/DATE]) driver rs rsmeta i)]
      parent-thunk)))

(defn- format-offset [t]
  (let [offset (t/format "ZZZZZ" (t/zone-offset t))]
    (if (= offset "Z")
      "UTC"
      offset)))

(defmethod sql.qp/inline-value [:mysql OffsetTime]
  [_ t]
  ;; MySQL doesn't support timezone offsets in literals so pass in a local time literal wrapped in a call to convert
  ;; it to the appropriate timezone
  (format "convert_tz('%s', '%s', @@session.time_zone)"
          (t/format "HH:mm:ss.SSS" t)
          (format-offset t)))

(defmethod sql.qp/inline-value [:mysql OffsetDateTime]
  [_ t]
  (format "convert_tz('%s', '%s', @@session.time_zone)"
          (t/format "yyyy-MM-dd HH:mm:ss.SSS" t)
          (format-offset t)))

(defmethod sql.qp/inline-value [:mysql ZonedDateTime]
  [_ t]
  (format "convert_tz('%s', '%s', @@session.time_zone)"
          (t/format "yyyy-MM-dd HH:mm:ss.SSS" t)
          (str (t/zone-id t))))

(defmethod driver/upload-type->database-type :mysql
  [_driver upload-type]
  (case upload-type
    ::upload/varchar-255              [[:varchar 255]]
    ::upload/text                     [:text]
    ::upload/int                      [:bigint]
    ::upload/auto-incrementing-int-pk [:bigint :not-null :auto-increment]
    ::upload/float                    [:double]
    ::upload/boolean                  [:boolean]
    ::upload/date                     [:date]
    ::upload/datetime                 [:datetime]
    ::upload/offset-datetime          [:timestamp]))

(defmethod driver/create-auto-pk-with-append-csv? :mysql [_driver] true)

(defmethod driver/table-name-length-limit :mysql
  [_driver]
  ;; https://dev.mysql.com/doc/refman/8.0/en/identifier-length.html
  64)

(defn- format-load
  [_clause [file-path table-name]]
  [(format "LOAD DATA LOCAL INFILE '%s' INTO TABLE %s" file-path (sql/format-entity table-name))])

(sql/register-clause! ::load format-load :insert-into)

(defn- offset-datetime->unoffset-datetime
  [driver database ^OffsetDateTime offset-time]
  (let [zone-id (t/zone-id (driver/db-default-timezone driver database))]
    (t/local-date-time offset-time zone-id)))

(defmulti ^:private value->string
  {:arglists '([driver val])}
  (fn [_ val] (type val)))

(defmethod value->string :default
  [_driver val]
  (str val))

(defmethod value->string nil
  [_driver _val]
  nil)

(defmethod value->string Boolean
  [_driver val]
  (if val
    "1"
    "0"))

(defmethod value->string LocalDateTime
  [_driver val]
  (t/format :iso-local-date-time val))

(let [zulu-fmt         "yyyy-MM-dd'T'HH:mm:ss"
      offset-fmt       "XXX"
      zulu-formatter   (DateTimeFormatter/ofPattern zulu-fmt)
      offset-formatter (DateTimeFormatter/ofPattern (str zulu-fmt offset-fmt))]
  (defmethod value->string OffsetDateTime
    [driver ^OffsetDateTime val]
    (let [uploads-db (upload/current-database)]
      (if (mariadb? uploads-db)
        (offset-datetime->unoffset-datetime driver uploads-db val)
        (t/format (if (.equals (.getOffset val) ZoneOffset/UTC)
                    zulu-formatter
                    offset-formatter)
                  val)))))

(defn- sanitize-value
  ;; Per https://dev.mysql.com/doc/refman/8.0/en/load-data.html#load-data-field-line-handling
  ;; Backslash is the MySQL escape character within strings in SQL statements. Thus, to specify a literal backslash,
  ;; you must specify two backslashes for the value to be interpreted as a single backslash. The escape sequences
  ;; '\t' and '\n' specify tab and newline characters, respectively.
  [v]
  (if (nil? v)
    "\\N"
    (str/replace v #"\\|\n|\r|\t" {"\\" "\\\\"
                                   "\n" "\\n"
                                   "\r" "\\r"
                                   "\t" "\\t"})))

(defn- row->tsv
  [driver column-count row]
  (when (not= column-count (count row))
    (throw (Exception. (format "ERROR: missing data in row \"%s\ (str/join "," row)))))
  (->> row
       (map (comp sanitize-value (partial value->string driver)))
       (str/join "\t")))

(defn- get-global-variable
  [db-id var-name]
  (:value
   (first
    (jdbc/query (sql-jdbc.conn/db->pooled-connection-spec db-id)
                ["show global variables like ?" var-name]))))

(defmethod driver/insert-into! :mysql
  [driver db-id ^String table-name column-names values]
  ;; `local_infile` must be turned on per
  ;; https://dev.mysql.com/doc/refman/8.0/en/load-data.html#load-data-local
  (if (not= (get-global-variable db-id "local_infile") "ON")
    ;; If it isn't turned on, fall back to the generic "INSERT INTO ..." way
    ((get-method driver/insert-into! :sql-jdbc) driver db-id table-name column-names values)
    (let [temp-file (File/createTempFile table-name ".tsv")
          file-path (.getAbsolutePath temp-file)]
      (try
        (let [tsvs    (map (partial row->tsv driver (count column-names)) values)
              dialect (sql.qp/quote-style driver)
              sql     (sql/format {::load   [file-path (keyword table-name)]
                                   :columns (quote-columns driver column-names)}
                                  :quoted true
                                  :dialect dialect)]
          (with-open [^java.io.Writer writer (jio/writer file-path)]
            (doseq [value (interpose \newline tsvs)]
              (.write writer (str value))))
          (sql-jdbc.execute/do-with-connection-with-options
           driver
           db-id
           nil
           (fn [conn]
             (jdbc/execute! {:connection conn} sql))))
        (finally
          (.delete temp-file))))))

(defn- parse-grant
  [grant]
  (condp re-find grant
    #"^GRANT PROXY ON "
    nil
    #"^GRANT (.+) ON FUNCTION "
    nil
    #"^GRANT (.+) ON PROCEDURE "
    nil
    ;; GRANT
    ;;     priv_type [(column_list)]
    ;;       [, priv_type [(column_list)]] ...
    ;;     ON object
    ;;     TO user etc.
    ;; }
    ;; For now we ignore column-level privileges. But this is how we could get them in the future.
    #"^GRANT (.+) ON (.+) TO "
    :>>
    (fn [[_ priv-types object]]
      (when-let [priv-types' (if (= priv-types "ALL PRIVILEGES")
                               #{:select :update :delete :insert}
                               (let [split-priv-types (->> (str/split priv-types #", ")
                                                           (map (comp keyword u/lower-case-en))
                                                           set)]
                                 (set/intersection #{:select :update :delete :insert} split-priv-types)))]
        {:type             :privileges
         :privilege-types  (not-empty priv-types')
         :level            (cond
                             (= object "*.*")             :global
                             (str/ends-with? object ".*") :database
                             :else                        :table)
         :object           object}))
    ;; GRANT role [, role] ... TO user etc.
    #"^GRANT (.+) TO "
    :>>
    (fn [[_ roles]]
      {:type  :roles
       :roles (set (map u/lower-case-en (str/split roles #",")))})))

(defn- privilege-grants-for-user
  [conn-spec user]
  (let [query  (fn [q] (->> (jdbc/query conn-spec q {:as-arrays? true})
                            (drop 1)
                            (map first)))
        grants (map parse-grant (query (str "SHOW GRANTS FOR " user)))
        {role-grants      :roles
         privilege-grants :privileges} (group-by :type grants)]
    (if (seq role-grants)
      (let [roles  (:roles (first role-grants))
            grants (map parse-grant (query (str "SHOW GRANTS FOR " user "USING " (str/join "," roles))))
            {privilege-grants :privileges} (group-by :type grants)]
        privilege-grants)
      privilege-grants)))

(defn- table-names->privileges
  [privilege-grants database-name table-names]
  (let [{global-grants   :global
         database-grants :database
         table-grants    :table} (group-by :level privilege-grants)
        lower-database-name  (u/lower-case-en database-name)
        all-table-privileges (set/union (:privilege-types (first global-grants))
                                        (:privilege-types (m/find-first #(= (:object %) (str "`" lower-database-name "`.*"))
                                                                        database-grants)))
        table-privileges (into {}
                               (keep (fn [grant]
                                       (when-let [match (re-find (re-pattern (str "^`" lower-database-name "`.`(.+)`")) (:object grant))]
                                         (let [[_ table-name] match]
                                           [table-name (:privilege-types grant)]))))
                               table-grants)]
    (into {}
          (keep (fn [table-name]
                  (when-let [privileges (not-empty (set/union all-table-privileges (get table-privileges table-name)))]
                    [table-name privileges])))
          table-names)))

(defmethod driver/describe-fields :mysql
  [driver database & args]
  (eduction
   (map (fn [col]
          (-> col
              (update :pk? pos?)
              (update :database-position int) ;; Comes in as biginteger
              (update :database-required pos?)
              (update :database-is-auto-increment pos?))))
   (apply (get-method driver/describe-fields :sql-jdbc) driver database args)))

(defmethod sql-jdbc.sync/describe-fields-sql :mysql
  [driver & {:keys [table-names details]}]
  (sql/format {:select [[:c.column_name :name]
                        [[:- :c.ordinal_position 1] :database-position]
                        [nil :table-schema]
                        [:c.table_name :table-name]
                        (if (some-> details :additional-options (str/includes? "tinyInt1isBit=false"))
                          [[:upper :c.data_type] :database-type]
                          [[:if [:= :column_type [:inline "tinyint(1)"]] [:inline "BIT"] [:upper :c.data_type]] :database-type])
                        [[:= :c.extra [:inline "auto_increment"]] :database-is-auto-increment]
                        [[:and
                          [:or [:= :column_default nil] [:= [:lower :column_default] [:inline "null"]]]
                          [:= :is_nullable [:inline "NO"]]
                          [:not [:= :c.extra [:inline "auto_increment"]]]]
                         :database-required]
                        [[:= :c.column_key [:inline "PRI"]] :pk?]
                        [[:nullif :c.column_comment [:inline ""]] :field-comment]]
               :from [[:information_schema.columns :c]]
               :where
               [:and [:raw "c.table_schema not in ('information_schema','performance_schema','sys','mysql')"]
                [:raw "c.table_name not in ('innodb_table_stats', 'innodb_index_stats')"]
                (when-let [db-name ((some-fn :db :dbname) details)]
                  [:= :c.table_schema db-name])
                (when (seq table-names) [:in [:lower :c.table_name] (map u/lower-case-en table-names)])]}
              :dialect (sql.qp/quote-style driver)))

(defmethod sql-jdbc.sync/describe-fks-sql :mysql
  [driver & {:keys [table-names]}]
  (sql/format {:select [[nil :pk-table-schema]
                        [:a.referenced_table_name :pk-table-name]
                        [:a.referenced_column_name :pk-column-name]
                        [nil :fk-table-schema]
                        [:a.table_name :fk-table-name]
                        [:a.column_name :fk-column-name]]
               :from [[:information_schema.key_column_usage :a]]
               :join [[:information_schema.table_constraints :b] [:using :constraint_schema :constraint_name :table_name]]
               :where [:and [:= :b.constraint_type [:inline "FOREIGN KEY"]]
                       [:!= :a.referenced_table_schema nil]
                       (when (seq table-names) [:in :a.table_name table-names])]
               :order-by [:a.table_name]}
              :dialect (sql.qp/quote-style driver)))