Compare commits
15 Commits
nodejs
...
search-cod
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0b5c4d664f | ||
|
|
7e953fce2b | ||
|
|
94daeccf1c | ||
|
|
cf73eed90b | ||
|
|
fb632d12fd | ||
|
|
477d839fb3 | ||
|
|
f1f97ef705 | ||
|
|
2591a195ec | ||
|
|
a1a441fb2c | ||
|
|
48c874ff98 | ||
|
|
339033aa5a | ||
|
|
423486b8b6 | ||
|
|
7cefb72acd | ||
|
|
e928fdbb0b | ||
|
|
436bb09ff9 |
@@ -7,9 +7,20 @@
|
||||
[metabase.permissions.core :as perms]
|
||||
[metabase.permissions.models.data-permissions :as data-perms]
|
||||
[metabase.search.core :as search]
|
||||
[metabase.search.test-util :as search.tu]
|
||||
[metabase.test :as mt]
|
||||
[metabase.util.random :refer [random-name]]))
|
||||
|
||||
;; These tests reindex against the appdb search index. Without isolation they would drive a real reindex/rotation
|
||||
;; on the SHARED global state store (and enqueue async ingestion on the background worker, which reads that global
|
||||
;; store), leaving a throwaway :active table cached for the TTL that later gets dropped -- "search_index__… does
|
||||
;; not exist" -> aborted connection -> cascading failures in unrelated tests. with-new-search-if-available* binds
|
||||
;; an isolated mock index + synchronous ingestion, so the rebuild never touches global state.
|
||||
(use-fixtures :each
|
||||
(fn [thunk]
|
||||
(search.tu/with-new-search-if-available-otherwise-legacy
|
||||
(thunk))))
|
||||
|
||||
(deftest published-table-collection-filter-test
|
||||
(mt/with-premium-features #{:library}
|
||||
(testing "Published tables are included in collection-filtered search"
|
||||
|
||||
@@ -4,12 +4,16 @@
|
||||
[environ.core :as env]
|
||||
[honey.sql.helpers :as sql.helpers]
|
||||
[java-time.api :as t]
|
||||
[metabase.app-db.cluster-lock :as cluster-lock]
|
||||
[metabase.app-db.core :as mdb]
|
||||
[metabase.config.core :as config]
|
||||
[metabase.events.core :as events]
|
||||
[metabase.search.appdb.index :as search.index]
|
||||
[metabase.search.appdb.index-state :as index-state]
|
||||
[metabase.search.appdb.scoring :as search.scoring]
|
||||
[metabase.search.appdb.specialization.postgres :as specialization.postgres]
|
||||
[metabase.search.appdb.table :as search.table]
|
||||
[metabase.search.appdb.writer :as search.writer]
|
||||
[metabase.search.config :as search.config]
|
||||
[metabase.search.engine :as search.engine]
|
||||
[metabase.search.filter :as search.filter]
|
||||
@@ -28,8 +32,7 @@
|
||||
[methodical.core :as methodical]
|
||||
[toucan2.core :as t2])
|
||||
(:import
|
||||
(java.time OffsetDateTime)
|
||||
(java.util Queue)))
|
||||
(java.time OffsetDateTime)))
|
||||
|
||||
;; Register the multimethods for each specialization
|
||||
(comment
|
||||
@@ -137,53 +140,63 @@
|
||||
(search.index/search-query search-string search-ctx select-items)
|
||||
(filter-layers search-ctx)))
|
||||
|
||||
(defn- results
|
||||
[{:keys [search-engine search-string] :as search-ctx}]
|
||||
;; Check whether there is a query-able index.
|
||||
(defn- ensure-active-table!
|
||||
"Throw an informative error if there is no active search index table.
|
||||
As a side effect, attempts a fresh sync and optionally triggers a background init."
|
||||
[{:keys [search-engine]}]
|
||||
(when-not (search.index/active-table)
|
||||
(let [index-state @@#'search.index/*indexes*
|
||||
;; Sync, in case we're just out of sync with the database.
|
||||
found-active (:active (#'search.index/sync-tracking-atoms!))
|
||||
;; If there's really no index, and we're running in prod - gulp, try to initialize now.
|
||||
(let [state-before (search.index/state-snapshot)
|
||||
found-active (:active (index-state/force-refresh! search.index/*state-store*))
|
||||
;; If there is really no index and we are in prod, trigger a background init to recover.
|
||||
init-now? (and (not found-active) config/is-prod?)]
|
||||
(when init-now?
|
||||
(log/warnf "Triggering a late initialization of the %s search index." search-engine)
|
||||
(try
|
||||
(future
|
||||
(search.engine/init! search-engine {:force-reset? false}))
|
||||
(future (search.engine/init! search-engine {:force-reset? false}))
|
||||
(catch Exception e
|
||||
(log/error e))))
|
||||
;; Even if the index exists now, return an error so that we don't obscure that there was an issue.
|
||||
(throw (ex-info "Search Index not found."
|
||||
{:search-engine search-engine
|
||||
:db-type (mdb/db-type)
|
||||
:version (search.spec/index-version-hash)
|
||||
:lang_code (i18n/site-locale-string)
|
||||
:forced-init? init-now?
|
||||
:index-state-before index-state
|
||||
:index-state-after @@#'search.index/*indexes*
|
||||
:index-metadata (t2/select :model/SearchIndexMetadata :engine :appdb)}))))
|
||||
:index-state-before state-before
|
||||
:index-state-after (search.index/state-snapshot)
|
||||
:index-metadata (t2/select :model/SearchIndexMetadata :engine :appdb)})))))
|
||||
|
||||
(defn- execute-search-query
|
||||
"Build and execute the appdb search query, returning rehydrated results."
|
||||
[{:keys [search-string] :as search-ctx}]
|
||||
(when (setting/string->boolean (:mb-experimental-search-block-on-queue env/env))
|
||||
;; wait for a bit for the queue to be drained
|
||||
(let [pending-updates #(search.ingestion/pending-count)]
|
||||
(when-not (u/poll {:thunk pending-updates
|
||||
:done? zero?
|
||||
:timeout-ms 2000
|
||||
:interval-ms 100})
|
||||
(log/warn "Returning search results even though they may be stale. Queue size:" (pending-updates)))))
|
||||
(let [weights (search.config/weights search-ctx)
|
||||
scorers (search.scoring/scorers search-ctx)
|
||||
query (->> (base-filtered-query search-ctx search-string [:legacy_input])
|
||||
(search.scoring/with-scores search-ctx scorers))]
|
||||
(->> (t2/query query)
|
||||
(map (partial rehydrate weights (keys scorers))))))
|
||||
|
||||
(defn- results
|
||||
;; Between the moment one pod drops a retired table and other pods' TTLs expire (up to 5 min),
|
||||
;; those pods may observe a table-not-found error on their first query. We retry once after a
|
||||
;; forced refresh to recover transparently.
|
||||
[{:keys [search-string] :as search-ctx}]
|
||||
(ensure-active-table! search-ctx)
|
||||
(tracing/with-span :search "search.appdb.query" {:search/query-length (count search-string)}
|
||||
(try
|
||||
(when (setting/string->boolean (:mb-experimental-search-block-on-queue env/env))
|
||||
;; wait for a bit for the queue to be drained
|
||||
(let [pending-updates #(.size ^Queue @#'search.ingestion/queue)]
|
||||
(when-not (u/poll {:thunk pending-updates
|
||||
:done? zero?
|
||||
:timeout-ms 2000
|
||||
:interval-ms 100})
|
||||
(log/warn "Returning search results even though they may be stale. Queue size:" (pending-updates)))))
|
||||
(let [weights (search.config/weights search-ctx)
|
||||
scorers (search.scoring/scorers search-ctx)
|
||||
query (->> (base-filtered-query search-ctx search-string [:legacy_input])
|
||||
(search.scoring/with-scores search-ctx scorers))]
|
||||
(->> (t2/query query)
|
||||
(map (partial rehydrate weights (keys scorers)))))
|
||||
(execute-search-query search-ctx)
|
||||
(catch Exception e
|
||||
;; Rule out the error coming from stale index metadata.
|
||||
(#'search.index/sync-tracking-atoms!)
|
||||
(throw e)))))
|
||||
(if (and (search.table/table-not-found-exception? e) (not (::already-retried search-ctx)))
|
||||
(do (index-state/force-refresh! search.index/*state-store*)
|
||||
(execute-search-query (assoc search-ctx ::already-retried true)))
|
||||
(throw e))))))
|
||||
|
||||
(defmethod search.engine/results :search.engine/appdb
|
||||
[search-ctx]
|
||||
@@ -205,6 +218,32 @@
|
||||
t2/query
|
||||
(into #{} (map :model)))))
|
||||
|
||||
(defn- populate-index! [mode]
|
||||
(search.writer/index-docs! mode (search.ingestion/searchable-documents)))
|
||||
|
||||
(def ^:private reindex-lock
|
||||
"Cluster-wide lock ensuring at most one appdb index build runs across the whole cluster at a time.
|
||||
Kept at its historical keyword name to work during rolling upgrades."
|
||||
:metabase.search.task.search-index/search-index-lock)
|
||||
|
||||
(defn- do-with-reindex-lock
|
||||
"Run `thunk` while holding [[reindex-lock]]. If another node or thread already holds it, log and skip
|
||||
(return nil) rather than queue: the in-progress build will produce a fresh index, and the periodic
|
||||
reindex covers anything a skipped trigger would have caught. Reentrant, so a locked [[init!]] may call
|
||||
[[reindex!]] without deadlocking."
|
||||
[thunk]
|
||||
(if-not (index-state/db-backed? search.index/*state-store*)
|
||||
(thunk)
|
||||
(try
|
||||
(cluster-lock/with-cluster-lock reindex-lock (thunk))
|
||||
(catch clojure.lang.ExceptionInfo e
|
||||
;; do-with-cluster-lock wraps a failed acquisition in an ex-info carrying :lock-names; anything else
|
||||
;; (a real error from the body) propagates.
|
||||
(if (contains? (ex-data e) :lock-names)
|
||||
(do (log/info "An appdb search index build is already running elsewhere; skipping this trigger.")
|
||||
nil)
|
||||
(throw e))))))
|
||||
|
||||
(defn- restrict-to-row [model id qry]
|
||||
(sql.helpers/where qry [:and
|
||||
[:= :search_index.model model]
|
||||
@@ -264,38 +303,46 @@
|
||||
[search-ctx model id]
|
||||
(appdb-diagnose search-ctx model id))
|
||||
|
||||
(defn- populate-index! [context]
|
||||
(search.index/index-docs! context (search.ingestion/searchable-documents)))
|
||||
|
||||
(defmethod search.engine/init! :search.engine/appdb
|
||||
[_ {:keys [re-populate?] :as opts}]
|
||||
(let [index-created (search.index/when-index-created)]
|
||||
(if (and index-created (< 3 (t/time-between (t/instant index-created) (t/instant) :days)))
|
||||
(do
|
||||
(log/info "Forcing early reindex because existing index is old")
|
||||
(search.engine/reindex! :search.engine/appdb {}))
|
||||
(let [created? (search.index/ensure-ready! opts)]
|
||||
(when (or created? re-populate?)
|
||||
(log/info "Populating index")
|
||||
(populate-index! (if created? :search/reindexing :search/updating)))))))
|
||||
(do-with-reindex-lock
|
||||
(fn []
|
||||
(let [index-created (search.index/when-index-created)]
|
||||
(if (and index-created (< 3 (t/time-between (t/instant index-created) (t/instant) :days)))
|
||||
(do
|
||||
(log/info "Forcing early reindex because existing index is old")
|
||||
(search.engine/reindex! :search.engine/appdb {}))
|
||||
(case (search.index/ensure-ready! opts)
|
||||
;; A replacement table was created as pending while the existing index keeps serving queries:
|
||||
;; populate the pending table, THEN rotate it in, so search is never blanked out mid-rebuild.
|
||||
:pending (do (log/info "Populating replacement index, then activating")
|
||||
(u/prog1 (populate-index! (search.index/background-mode))
|
||||
(search.index/activate-table!)))
|
||||
;; A fresh empty table was activated immediately (nothing was serving): just populate it so
|
||||
;; partial results appear as soon as possible.
|
||||
:activated (do (log/info "Populating index")
|
||||
(populate-index! (search.index/background-mode)))
|
||||
;; No reset was needed; optionally re-populate the existing active index in place.
|
||||
(when re-populate?
|
||||
(log/info "Re-populating index")
|
||||
(populate-index! (search.index/incremental-mode)))))))))
|
||||
|
||||
(defmethod search.engine/sync-from-restored-db! :search.engine/appdb [_]
|
||||
(search.index/sync-from-restored-db!))
|
||||
|
||||
(defmethod search.engine/reindex! :search.engine/appdb
|
||||
[_ {:keys [in-place?]}]
|
||||
(try
|
||||
(search.index/ensure-ready!)
|
||||
(if in-place?
|
||||
(when-let [table (search.index/active-table)]
|
||||
;; keep the current table, just delete its contents
|
||||
(t2/delete! table))
|
||||
(search.index/maybe-create-pending!))
|
||||
(u/prog1 (populate-index! (if in-place? :search/updating :search/reindexing))
|
||||
(search.index/activate-table!))
|
||||
(catch Throwable e
|
||||
(log/error e "Error during reindexing")
|
||||
(throw e))))
|
||||
(do-with-reindex-lock
|
||||
(fn []
|
||||
(try
|
||||
(let [mode (if in-place? (search.index/in-place-mode) (search.index/background-mode))]
|
||||
(search.index/ensure-ready!)
|
||||
(search.index/prepare! mode)
|
||||
(u/prog1 (populate-index! mode)
|
||||
(search.index/finish! mode)))
|
||||
(catch Throwable e
|
||||
(log/error e "Error during reindexing")
|
||||
(throw e))))))
|
||||
|
||||
(derive :event/setting-update ::settings-changed-event)
|
||||
|
||||
|
||||
39
src/metabase/search/appdb/document.clj
Normal file
39
src/metabase/search/appdb/document.clj
Normal file
@@ -0,0 +1,39 @@
|
||||
(ns metabase.search.appdb.document
|
||||
"Pure transformation of an ingested entity into a row ready to be upserted into a search index table.
|
||||
|
||||
Nothing here touches the database or the active/pending state machine — it is just data shaping, which
|
||||
makes it cheap to test in isolation. The one external call is [[specialization/extra-entry-fields]],
|
||||
which adds driver-specific generated columns (e.g. the Postgres tsvector)."
|
||||
(:require
|
||||
[clojure.set :as set]
|
||||
[clojure.string :as str]
|
||||
[metabase.search.appdb.specialization.api :as specialization]
|
||||
[metabase.search.spec :as search.spec]
|
||||
[metabase.util.json :as json]))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(defn- strip-junk-chars
|
||||
"Replace control characters (\\p{Cc}: C0 controls including \\t \\n \\r, DEL, C1 controls) and surrogate
|
||||
code points (\\p{Cs}) with a single space so they act as token boundaries for full-text indexing instead
|
||||
of accidentally fusing adjacent words. Postgres also outright rejects literal NUL (0x00) in text columns,
|
||||
so this is required to keep reindex batches from aborting. Non-string values pass through unchanged."
|
||||
[v]
|
||||
(cond-> v (string? v) (str/replace #"(?U)[\p{Cc}\p{Cs}]" " ")))
|
||||
|
||||
(defn document->entry
|
||||
"Shape an ingested `entity` map into a row for the index table: rename id/timestamp columns, JSON-encode
|
||||
the display/legacy payloads, strip junk characters from text, and add driver-specific generated fields."
|
||||
[entity]
|
||||
(let [entity (update-vals entity strip-junk-chars)]
|
||||
(-> entity
|
||||
(select-keys (conj search.spec/attr-columns :model :display_data :legacy_input))
|
||||
(set/rename-keys {:id :model_id
|
||||
:created_at :model_created_at
|
||||
:updated_at :model_updated_at})
|
||||
(assoc :updated_at :%now)
|
||||
(update :display_data json/encode)
|
||||
;; legacy_input is already JSON-encoded in ->document; encode only if it's still a map (e.g., in tests)
|
||||
(update :legacy_input #(if (string? %) % (json/encode %)))
|
||||
(dissoc :native_query)
|
||||
(merge (specialization/extra-entry-fields entity)))))
|
||||
@@ -1,234 +1,113 @@
|
||||
(ns metabase.search.appdb.index
|
||||
"The appdb search index state machine: which physical table is active/pending, how reindexes rotate them
|
||||
([[ReindexMode]]), and the search queries that read from the active table. The write pipeline that
|
||||
actually populates these tables lives in [[metabase.search.appdb.writer]].
|
||||
|
||||
## Concurrency
|
||||
|
||||
Three coordination mechanisms cooperate; keep them straight:
|
||||
|
||||
1. **Cluster lock** (acquired at the appdb engine's init!/reindex! choke points in
|
||||
[[metabase.search.appdb.core]]): serializes whole reindex runs across the cluster, so at most one node
|
||||
rebuilds at a time.
|
||||
2. **Node-local [[index-lock]]**: serializes the multi-step DDL sequences (create-pending → activate) and
|
||||
[[ensure-ready!]] within a single JVM, so two threads on one node never both build a new index.
|
||||
3. **TTL state cache** ([[*state-store*]]): each node caches the {:active :pending} table names for a few
|
||||
minutes; [[index-state/force-refresh!]] re-reads immediately after a node performs a rotation, and the
|
||||
write pipeline force-refreshes on a stale-table error.
|
||||
|
||||
## Reindex modes
|
||||
|
||||
Writes go to different physical tables depending on what is happening (see [[metabase.search.appdb.writer]]):
|
||||
- a **full reindex** builds a fresh :pending table while the old :active table keeps serving queries,
|
||||
then atomically rotates pending → active;
|
||||
- an **in-place reindex** clears and repopulates the :active table directly (no pending);
|
||||
- **incremental updates** write to *both* :active and :pending, so a rebuild in progress stays current."
|
||||
(:require
|
||||
[clojure.set :as set]
|
||||
[clojure.string :as str]
|
||||
[honey.sql.helpers :as sql.helpers]
|
||||
[metabase.analytics-interface.core :as analytics]
|
||||
[metabase.app-db.core :as mdb]
|
||||
[metabase.config.core :as config]
|
||||
[metabase.search.appdb.index-state :as index-state]
|
||||
[metabase.search.appdb.specialization.api :as specialization]
|
||||
[metabase.search.appdb.specialization.h2 :as h2]
|
||||
[metabase.search.appdb.specialization.postgres :as postgres]
|
||||
[metabase.search.config :as search.config]
|
||||
[metabase.search.appdb.table :as table]
|
||||
[metabase.search.engine :as search.engine]
|
||||
[metabase.search.ingestion :as search.ingestion]
|
||||
[metabase.search.models.search-index-metadata :as search-index-metadata]
|
||||
[metabase.search.spec :as search.spec]
|
||||
[metabase.tracing.core :as tracing]
|
||||
[metabase.util :as u]
|
||||
[metabase.util.i18n :as i18n]
|
||||
[metabase.util.json :as json]
|
||||
[metabase.util.log :as log]
|
||||
[metabase.util.string :as string]
|
||||
[toucan2.core :as t2])
|
||||
(:import
|
||||
(org.h2.jdbc JdbcSQLSyntaxErrorException)
|
||||
(org.postgresql.util PSQLException)))
|
||||
|
||||
(comment
|
||||
h2/keep-me
|
||||
postgres/keep-me)
|
||||
[toucan2.core :as t2]))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(def ^:private insert-batch-size 150)
|
||||
;;; --------------------------------------- Index state ---------------------------------------------
|
||||
|
||||
(def ^:private sync-tracking-period (long (* 5 #_minutes 60e9)))
|
||||
(defonce ^{:dynamic true
|
||||
:doc "Lazily-refreshed view of the search_index_metadata table.
|
||||
Tests bind this to a MockStateStore so no DB interaction is needed for index state."}
|
||||
*state-store*
|
||||
(index-state/db-backed-store
|
||||
(fn []
|
||||
(reduce-kv (fn [acc status table-name]
|
||||
(let [kw (keyword table-name)]
|
||||
(if (table/exists? kw)
|
||||
(assoc acc status kw)
|
||||
(do (log/debugf "Search index %s table %s not found; not tracking it" status table-name)
|
||||
acc))))
|
||||
{}
|
||||
(search-index-metadata/indexes :appdb (search.spec/index-version-hash))))))
|
||||
|
||||
(defonce ^:private next-sync-at (atom nil))
|
||||
|
||||
(defonce ^:dynamic ^:private ^{:doc "This atom is often reset! in threads, so modifications should be done only when locking it first."}
|
||||
*indexes*
|
||||
(atom {:active nil, :pending nil}))
|
||||
|
||||
(def ^:private ^:dynamic *mocking-tables* false)
|
||||
(defonce ^{:private true
|
||||
:doc "Node-local monitor guarding the multi-step DDL sequences (create-pending → activate) so
|
||||
concurrent threads on the same node don't both attempt to build a new index."}
|
||||
index-lock (Object.))
|
||||
|
||||
(defmethod search.engine/reset-tracking! :search.engine/appdb [_]
|
||||
(reset! *indexes* nil))
|
||||
|
||||
(declare exists?)
|
||||
|
||||
(defn- sync-tracking-atoms!
|
||||
"Sync the *indexes* atom with the current database metadata state."
|
||||
[]
|
||||
;; Locks the indexes so the reset! doesn't lose data written to the db by a different thread between the read and write
|
||||
(locking *indexes*
|
||||
(let [indexes (into {}
|
||||
(for [[status table-name] (search-index-metadata/indexes :appdb (search.spec/index-version-hash))]
|
||||
(if (exists? table-name)
|
||||
[status (keyword table-name)]
|
||||
;; For debugging, make it clear why we are not tracking the given metadata.
|
||||
[(keyword (name status) "not-found") (keyword table-name)])))]
|
||||
(log/debugf "Sync tracking atoms: %s" indexes)
|
||||
(reset! *indexes* indexes))))
|
||||
(index-state/set-state! *state-store* {:active nil :pending nil}))
|
||||
|
||||
(defn sync-from-restored-db!
|
||||
"Re-sync tracking atoms with the current database state.
|
||||
"Re-sync index tracking with the current database state.
|
||||
Used after snapshot restore where the index tables are already present."
|
||||
[]
|
||||
(reset! next-sync-at nil)
|
||||
(sync-tracking-atoms!))
|
||||
|
||||
;; This exists only to be mocked.
|
||||
(defn- now [] (System/nanoTime))
|
||||
|
||||
(defn- sync-tracking-atoms-if-stale! []
|
||||
(when-not *mocking-tables*
|
||||
(let [current @next-sync-at
|
||||
now-ns (now)]
|
||||
(when (or (nil? current) (> now-ns current))
|
||||
;; Use compare-and-set! to ensure only one thread wins the race and syncs
|
||||
(when (compare-and-set! next-sync-at current (+ now-ns sync-tracking-period))
|
||||
(sync-tracking-atoms!))))))
|
||||
(index-state/force-refresh! *state-store*))
|
||||
|
||||
(defn active-table
|
||||
"The table against which we should currently make search queries."
|
||||
[]
|
||||
(sync-tracking-atoms-if-stale!)
|
||||
(:active @*indexes*))
|
||||
(:active (index-state/current-state *state-store*)))
|
||||
|
||||
(defn- pending-table
|
||||
(defn pending-table
|
||||
"A partially populated table that will take over from [[active-table]] when it is done."
|
||||
[]
|
||||
(sync-tracking-atoms-if-stale!)
|
||||
(:pending @*indexes*))
|
||||
(:pending (index-state/current-state *state-store*)))
|
||||
|
||||
(defn gen-table-name
|
||||
"Generate a unique table name to use as a search index table. If no suffix is provided, none will be used"
|
||||
([]
|
||||
(gen-table-name ""))
|
||||
([suffix]
|
||||
(keyword (str (str/replace (str "search_index__" (u/lower-case-en (u/generate-nano-id))) #"-" "_") suffix))))
|
||||
(defn state-snapshot
|
||||
"Return the current cached {:active …, :pending …} state. For diagnostics only."
|
||||
[]
|
||||
(index-state/current-state *state-store*))
|
||||
|
||||
(defn- table-name [kw]
|
||||
(cond-> (name kw)
|
||||
(= :h2 (mdb/db-type)) u/upper-case-en))
|
||||
|
||||
(defn exists?
|
||||
"Whether the given index `table` actually exists in the appdb (the tracked active/pending table can be
|
||||
briefly stale relative to what has been dropped)."
|
||||
[table]
|
||||
(when table
|
||||
(t2/exists? :information_schema.tables :table_name (table-name table))))
|
||||
|
||||
(defn- drop-table! [table]
|
||||
(boolean
|
||||
(when table
|
||||
(t2/query (sql.helpers/drop-table :if-exists (keyword (table-name table)))))))
|
||||
|
||||
(defn- orphan-indexes []
|
||||
(map (comp keyword u/lower-case-en :table_name)
|
||||
(t2/query {:select [:table_name]
|
||||
:from :information_schema.tables
|
||||
:where [:and
|
||||
[:= :table_schema :%current_schema]
|
||||
[:or
|
||||
[:like [:lower :table_name] [:inline "search\\_index\\_\\_%"]]
|
||||
;; legacy table names
|
||||
[:in [:lower :table_name]
|
||||
(mapv #(vector :inline %) ["search_index" "search_index_next" "search_index_retired"])]]
|
||||
;; Exclude temp tables — they are managed by with-temp-index-table
|
||||
[:not-like [:lower :table_name] [:inline "%\\_temp"]]
|
||||
[:not-in [:lower :table_name]
|
||||
{:select [:%lower.index_name]
|
||||
:from [(t2/table-name :model/SearchIndexMetadata)]
|
||||
:where [:= :engine [:inline "appdb"]]}]]})))
|
||||
|
||||
(defn- delete-obsolete-tables! []
|
||||
;; Delete metadata around indexes that are no longer needed.
|
||||
(search-index-metadata/delete-obsolete! (search.spec/index-version-hash))
|
||||
;; Drop any indexes that are no longer referenced.
|
||||
(let [dropped (volatile! [])]
|
||||
(doseq [table (orphan-indexes)]
|
||||
(try
|
||||
(t2/query (sql.helpers/drop-table table))
|
||||
(vswap! dropped conj table)
|
||||
;; Deletion could fail if it races with other instances
|
||||
(catch Exception e
|
||||
(log/warnf e "Failed to drop stale index %s" table))))
|
||||
(log/infof "Dropped %d stale indexes: %s" (count @dropped) @dropped)))
|
||||
|
||||
(defn- ->db-type [t]
|
||||
(get {:pk :int, :timestamp :timestamp-with-time-zone} t t))
|
||||
|
||||
(defn- ->db-column [c]
|
||||
(or (get {:id :model_id
|
||||
:created-at :model_created_at
|
||||
:updated-at :model_updated_at}
|
||||
c)
|
||||
(keyword (u/->snake_case_en (name c)))))
|
||||
|
||||
(def ^:private not-null
|
||||
#{:archived :name})
|
||||
|
||||
(def ^:private default
|
||||
{:archived false})
|
||||
|
||||
;; If this fails, we'll need to increase the size of :model below
|
||||
(assert (>= 32 (transduce (map (comp count name)) max 0 search.config/all-models)))
|
||||
|
||||
(def ^:private base-schema
|
||||
(into [[:model [:varchar 32] :not-null]
|
||||
[:display_data :text :not-null]
|
||||
[:legacy_input :text :not-null]
|
||||
;; useful for tracking the speed and age of the index
|
||||
[:created_at :timestamp-with-time-zone
|
||||
[:default [:raw "CURRENT_TIMESTAMP"]]
|
||||
:not-null]
|
||||
[:updated_at :timestamp-with-time-zone :not-null]]
|
||||
(keep (fn [[k t]]
|
||||
(when t
|
||||
(into [(->db-column k) (->db-type t)]
|
||||
(concat
|
||||
(when (not-null k)
|
||||
[:not-null])
|
||||
(when-some [d (default k)]
|
||||
[[:default d]]))))))
|
||||
search.spec/attr-types))
|
||||
|
||||
(defn create-table!
|
||||
"Create an index table with the given name. Should fail if it already exists."
|
||||
[table-name]
|
||||
;; Create with a separate transaction so that postgresql will complete the index creations before returning,
|
||||
;; even when already running in a transaction
|
||||
(t2/with-transaction [_ (mdb/app-db)]
|
||||
(-> (sql.helpers/create-table table-name)
|
||||
(sql.helpers/with-columns (specialization/table-schema base-schema))
|
||||
t2/query)
|
||||
(let [table-name (name table-name)]
|
||||
(doseq [stmt (specialization/post-create-statements table-name table-name)]
|
||||
(t2/query stmt)))))
|
||||
;;; --------------------------------------- Rotation ------------------------------------------------
|
||||
|
||||
(defn maybe-create-pending!
|
||||
"Create a search index table if one doesn't exist. Record and return the name of the table, regardless."
|
||||
"Create a pending index table if one does not already exist. Returns the pending table name, or nil."
|
||||
[]
|
||||
(locking *indexes*
|
||||
(if *mocking-tables*
|
||||
;; In a test where the atoms are the source of truth, create a new table if necessary.
|
||||
(or (pending-table)
|
||||
(let [table-name (gen-table-name)]
|
||||
(create-table! table-name)
|
||||
(swap! *indexes* assoc :pending table-name) table-name))
|
||||
;; The database is the source of truth
|
||||
(let [{:keys [pending]} (sync-tracking-atoms!)]
|
||||
(or pending
|
||||
(let [table-name (gen-table-name)]
|
||||
(log/infof "Creating pending index %s for lang %s" table-name (i18n/site-locale-string))
|
||||
;; We may fail to insert a new metadata row if we lose a race with another instance.
|
||||
(when (search-index-metadata/create-pending! :appdb (search.spec/index-version-hash) table-name)
|
||||
(locking index-lock
|
||||
(or (pending-table)
|
||||
;; We may fail to insert a new metadata row if we lose a race with another instance.
|
||||
(let [table-name (table/gen-table-name)]
|
||||
(log/infof "Creating pending index %s for lang %s" table-name (i18n/site-locale-string))
|
||||
(when (search-index-metadata/create-pending! :appdb (search.spec/index-version-hash) table-name)
|
||||
(try
|
||||
(table/create-table! table-name)
|
||||
(catch Exception e
|
||||
(log/error e "Error creating pending index table, cleaning up metadata")
|
||||
(try
|
||||
(create-table! table-name)
|
||||
(catch Exception e
|
||||
(log/error e "Error creating pending index table, cleaning up metadata")
|
||||
(try
|
||||
(t2/with-connection [safe-conn (mdb/app-db)]
|
||||
(t2/delete! :conn safe-conn :model/SearchIndexMetadata :index_name (name table-name)))
|
||||
(catch Exception del-e
|
||||
(log/warn del-e "Error clearing out search metadata after failure")))
|
||||
(sync-tracking-atoms!))))
|
||||
(let [pending (:pending (sync-tracking-atoms!))]
|
||||
(log/infof "New pending index %s" pending)
|
||||
pending)))))))
|
||||
(t2/with-connection [safe-conn (mdb/app-db)]
|
||||
(t2/delete! :conn safe-conn :model/SearchIndexMetadata :index_name (name table-name)))
|
||||
(catch Exception del-e
|
||||
(log/warn del-e "Error clearing out search metadata after failure"))))))
|
||||
(let [pending (:pending (index-state/force-refresh! *state-store*))]
|
||||
(log/infof "New pending index %s" pending)
|
||||
pending)))))
|
||||
|
||||
(defn- analyze-table!
|
||||
"Refresh the table's statistics so size estimates are accurate as soon as it becomes active.
|
||||
@@ -240,182 +119,121 @@
|
||||
(log/warnf e "Failed to analyze index table %s" table-name))))
|
||||
|
||||
(defn activate-table!
|
||||
"Make the pending index active if it exists. Returns true if it did so."
|
||||
"Make the pending index active if it exists. Returns true if a rotation occurred."
|
||||
[]
|
||||
(locking *indexes*
|
||||
(if *mocking-tables*
|
||||
;; The atoms are the only source of truth, we must not update the metadata.
|
||||
(boolean
|
||||
(when-let [pending (:pending @*indexes*)]
|
||||
(analyze-table! pending)
|
||||
(reset! *indexes* {:pending nil, :active pending}) true))
|
||||
;; Ensure the metadata is updated and pruned.
|
||||
(let [{:keys [pending]} (sync-tracking-atoms!)]
|
||||
(log/infof "Activating pending index %s" pending)
|
||||
(when pending
|
||||
(analyze-table! pending)
|
||||
(let [active (keyword (search-index-metadata/active-pending! :appdb (search.spec/index-version-hash)))]
|
||||
(reset! *indexes* {:pending nil :active active})
|
||||
(log/infof "Activated pending index %s" active)))
|
||||
;; Clean up while we're here
|
||||
(delete-obsolete-tables!)
|
||||
;; Did *we* do a rotation?
|
||||
(boolean pending)))))
|
||||
(locking index-lock
|
||||
(let [{:keys [pending]} (index-state/force-refresh! *state-store*)]
|
||||
(log/infof "Activating pending index %s" pending)
|
||||
(when pending
|
||||
;; Refresh statistics before rotation so size estimates are accurate the moment it goes active.
|
||||
(analyze-table! pending)
|
||||
(let [active (keyword (search-index-metadata/active-pending! :appdb (search.spec/index-version-hash)))]
|
||||
(index-state/set-state! *state-store* {:pending nil :active active})
|
||||
(log/infof "Activated pending index %s" active)))
|
||||
;; Clean up while we're here — skip in mock mode to avoid touching production metadata
|
||||
(when (index-state/db-backed? *state-store*)
|
||||
(table/delete-obsolete-tables!))
|
||||
;; Did *we* do a rotation?
|
||||
(boolean pending))))
|
||||
|
||||
(defn- strip-junk-chars
|
||||
"Replace control characters (\\p{Cc}: C0 controls including \\t \\n \\r, DEL, C1 controls) and surrogate
|
||||
code points (\\p{Cs}) with a single space so they act as token boundaries for full-text indexing instead
|
||||
of accidentally fusing adjacent words. Postgres also outright rejects literal NUL (0x00) in text columns,
|
||||
so this is required to keep reindex batches from aborting. Non-string values pass through unchanged."
|
||||
[v]
|
||||
(cond-> v (string? v) (str/replace #"(?U)[\p{Cc}\p{Cs}]" " ")))
|
||||
(defn reset-index!
|
||||
"Ensure there is a fresh, empty index table to populate; in case the table schema or stored data format
|
||||
has changed. Returns:
|
||||
|
||||
(defn- document->entry [entity]
|
||||
(let [entity (update-vals entity strip-junk-chars)]
|
||||
(-> entity
|
||||
(select-keys (conj search.spec/attr-columns :model :display_data :legacy_input))
|
||||
(set/rename-keys {:id :model_id
|
||||
:created_at :model_created_at
|
||||
:updated_at :model_updated_at})
|
||||
(assoc :updated_at :%now)
|
||||
(update :display_data json/encode)
|
||||
;; legacy_input is already JSON-encoded in ->document; encode only if it's still a map (e.g., in tests)
|
||||
(update :legacy_input #(if (string? %) % (json/encode %)))
|
||||
(dissoc :native_query)
|
||||
(merge (specialization/extra-entry-fields entity)))))
|
||||
:activated — a new empty table was made active immediately, because no complete index was serving
|
||||
(genuine first-time creation). Partial results then appear as soon as the caller populates.
|
||||
:pending — a complete index is still serving, so the replacement was created as :pending and the old
|
||||
:active table is left serving. The caller MUST populate the pending table and then call
|
||||
[[activate-table!]], so search is never blanked out mid-rebuild.
|
||||
|
||||
(defn- table-not-found-exception? [e]
|
||||
;; Use with care, obviously this can give false positives if used with a query that's *actually* malformed.
|
||||
;; TODO we should handle the MySQL and MariaDB flavors here too
|
||||
(or (instance? PSQLException (ex-cause e))
|
||||
(instance? JdbcSQLSyntaxErrorException (ex-cause e))))
|
||||
The choice hinges on whether we have a usable index to keep serving — never throw away the only thing
|
||||
answering queries just to put an empty table in its place."
|
||||
[]
|
||||
(log/infof "Resetting appdb index for version %s, active table: %s" (search.spec/index-version-hash)
|
||||
(pr-str (active-table)))
|
||||
(letfn [(reset-logic []
|
||||
;; Discard any in-progress pending table before starting fresh.
|
||||
(when-let [table-name (pending-table)]
|
||||
(let [deleted (search-index-metadata/delete-index! :appdb (search.spec/index-version-hash) table-name)]
|
||||
(when (pos? deleted)
|
||||
(log/infof "Deleted %d pending indices" deleted)))
|
||||
(index-state/set-state! *state-store* {:active (active-table) :pending nil}))
|
||||
(let [serving? (boolean (and (active-table) (table/exists? (active-table))))]
|
||||
(maybe-create-pending!)
|
||||
(if serving?
|
||||
:pending
|
||||
;; Nothing usable is serving — activate the fresh empty table now for partial results ASAP.
|
||||
(do (activate-table!) :activated))))]
|
||||
(if search.ingestion/*force-sync*
|
||||
(reset-logic)
|
||||
;; Use a dedicated connection so the empty tables become visible to other connections
|
||||
;; even while the initial data load is happening in an outer transaction.
|
||||
(t2/with-connection [_ (mdb/data-source)]
|
||||
(reset-logic)))))
|
||||
|
||||
(defn- retry-upsert-ex [table-type table-name-before table-name-after e-before e-after]
|
||||
(ex-info "Failed retrying search index batch upsert"
|
||||
{:table-type table-type
|
||||
:table-name-before-refresh table-name-before
|
||||
:table-name-after-refresh table-name-after
|
||||
:initial-exception-class (class e-before)
|
||||
:initial-exception-message (ex-message e-before)}
|
||||
e-after))
|
||||
(defn ensure-ready!
|
||||
"Ensure the index is ready to be populated. Returns nil when an existing index was already in place and no
|
||||
reset was needed, otherwise the [[reset-index!]] signal (:activated or :pending) telling the caller how to
|
||||
populate it."
|
||||
[& {:keys [force-reset?]}]
|
||||
(locking index-lock
|
||||
(when (nil? (active-table))
|
||||
(index-state/force-refresh! *state-store*))
|
||||
(when (or force-reset? (not (table/exists? (active-table))))
|
||||
(reset-index!))))
|
||||
|
||||
(defn- safe-batch-upsert!
|
||||
"A version of batch-upsert! that no-ops for missing indexes, and handles stale index tracking metadata.
|
||||
;;; --------------------------------------- Reindex modes -------------------------------------------
|
||||
|
||||
Returns the name of the table that was written to, or nil if there is none being tracked, or nil
|
||||
if the upsert failed for any other reason — in which case the failure is logged at ERROR and we
|
||||
continue so the rest of the reindex can finish and activate whatever was successfully written.
|
||||
(defprotocol ReindexMode
|
||||
"Strategy describing how one population pass writes its batches and rotates tables. There are three:
|
||||
a background full rebuild, an in-place rebuild, and incremental live updates (see the namespace
|
||||
docstring's 'Reindex modes'). [[prepare!]] and [[finish!]] bracket a *reindex*; the live update path
|
||||
and the initial populate reuse only the per-batch write behavior ([[commits-per-batch?]], which feeds
|
||||
the writer's derived [[metabase.search.appdb.writer/write-plan]])."
|
||||
(prepare! [mode]
|
||||
"DDL to run before populating: create a fresh pending table, clear the active one, or nothing.")
|
||||
(commits-per-batch? [mode]
|
||||
"True if each batch should COMMIT in a dedicated connection so the partial build is visible to other
|
||||
connections and nodes during a long rebuild. Always false under [[search.ingestion/*force-sync*]],
|
||||
which runs synchronously inside the caller's own transaction.")
|
||||
(finish! [mode]
|
||||
"DDL to run after populating: activate the pending table (or just clean up). Returns true if a
|
||||
rotation occurred."))
|
||||
|
||||
We recover gracefully the first time if the tracking atom was stale, but do not check again on retry."
|
||||
[table-type table-name-fn entries]
|
||||
;; For convenience, no-op if we are not tracking any table.
|
||||
(when-let [table-name (table-name-fn)]
|
||||
(let [upsert! (fn [t] (specialization/batch-upsert! t entries) t)]
|
||||
(try
|
||||
(upsert! table-name)
|
||||
(catch InterruptedException ie
|
||||
(.interrupt (Thread/currentThread))
|
||||
(throw ie))
|
||||
(catch Exception e
|
||||
;; If the failure is a legitimately non-existent table, refresh tracking and retry once.
|
||||
(if (and (table-not-found-exception? e) (not (exists? table-name)))
|
||||
(when-let [refreshed-table-name (do (sync-tracking-atoms!) (table-name-fn))]
|
||||
(if (= table-name refreshed-table-name)
|
||||
(throw (ex-info "Currently tracked index does not exist" e {:table-name table-name}))
|
||||
(try
|
||||
(upsert! refreshed-table-name)
|
||||
(catch InterruptedException ie
|
||||
(.interrupt (Thread/currentThread))
|
||||
(throw ie))
|
||||
(catch Exception e2
|
||||
(if (table-not-found-exception? e2)
|
||||
(throw (retry-upsert-ex table-type table-name refreshed-table-name e e2))
|
||||
(do (analytics/inc! :metabase-search/appdb-index-batches-skipped {:table-type table-type})
|
||||
(log/errorf (retry-upsert-ex table-type table-name refreshed-table-name e e2)
|
||||
"Error upserting search index batch into %s table %s after refresh; skipping batch and continuing"
|
||||
(name table-type) refreshed-table-name)
|
||||
nil))))))
|
||||
;; Any other failure - log and continue so reindex can still finish.
|
||||
(do (analytics/inc! :metabase-search/appdb-index-batches-skipped {:table-type table-type})
|
||||
(log/errorf e "Error upserting search index batch into %s table %s; skipping batch and continuing"
|
||||
(name table-type) table-name)
|
||||
nil)))))))
|
||||
(defrecord BackgroundReindex []
|
||||
ReindexMode
|
||||
;; Build a brand-new pending table while the current active table keeps serving queries.
|
||||
(prepare! [_] (maybe-create-pending!) nil)
|
||||
(commits-per-batch? [_] (not search.ingestion/*force-sync*))
|
||||
(finish! [_] (activate-table!)))
|
||||
|
||||
(defn- batch-update!
|
||||
"Create the given search index entries in bulk. Commits after each batch.
|
||||
(defrecord InPlaceReindex []
|
||||
ReindexMode
|
||||
;; Empty the active table and repopulate it directly — no pending table, no rotation.
|
||||
(prepare! [_] (when-let [t (active-table)] (t2/delete! t)) nil)
|
||||
(commits-per-batch? [_] false)
|
||||
(finish! [_] (activate-table!)))
|
||||
|
||||
`reindex-table`, when non-nil, is the destination captured once at the start of a full reindex (see
|
||||
[[index-docs!]]) -- the pending table when a rebuild is staging one, otherwise the active table for an
|
||||
initial build. During a reindex we write ONLY to that captured table, so a concurrent resync that
|
||||
transiently blanks the tracking atom can't redirect writes elsewhere mid-rebuild. For incremental and
|
||||
in-place updates (`reindex-table` nil) we dual-write to the active and pending tables so an in-progress
|
||||
rebuild stays current."
|
||||
[reindex-table documents]
|
||||
;; Protect against tests that nuke the appdb
|
||||
(when config/is-test?
|
||||
(when-let [table (active-table)]
|
||||
(when (not (exists? table))
|
||||
(log/warnf "Unable to find table %s and no longer tracking it as active", table)
|
||||
(swap! *indexes* assoc :active nil)))
|
||||
(when-let [table (pending-table)]
|
||||
(when (not (exists? table))
|
||||
(log/warnf "Unable to find table %s and no longer tracking it as pending", table)
|
||||
(swap! *indexes* assoc :pending nil))))
|
||||
(defrecord IncrementalUpdate []
|
||||
ReindexMode
|
||||
;; Live edits: no DDL, just keep both tables current.
|
||||
(prepare! [_] nil)
|
||||
(commits-per-batch? [_] false)
|
||||
(finish! [_] false))
|
||||
|
||||
(let [reindexing? (some? reindex-table)
|
||||
do-writes (fn []
|
||||
(let [entries (map document->entry documents)
|
||||
updated (if reindexing?
|
||||
;; Full reindex: write only to the table captured for the whole run, so a
|
||||
;; transiently-blanked tracking atom can't redirect writes into the live
|
||||
;; active table (which would silently drop documents from the new index).
|
||||
(safe-batch-upsert! :pending (constantly reindex-table) entries)
|
||||
;; Incremental / in-place: dual-write so an in-progress rebuild stays
|
||||
;; current. Either table may legitimately be absent.
|
||||
(let [active-updated (safe-batch-upsert! :active active-table entries)
|
||||
pending-updated (safe-batch-upsert! :pending pending-table entries)]
|
||||
(or active-updated pending-updated)))]
|
||||
(when updated
|
||||
(u/prog1 (->> entries (map :model) frequencies)
|
||||
(when reindexing?
|
||||
(t2/query ["commit"]))
|
||||
(log/trace "indexed documents for " <>)))))]
|
||||
(if reindexing?
|
||||
;; New connection used for performing the updates which commit periodically without impacting any outer transactions.
|
||||
(t2/with-connection [_conn (mdb/data-source)]
|
||||
(do-writes))
|
||||
(do-writes))))
|
||||
(defn background-mode
|
||||
"A full reindex into a fresh pending table (the active table keeps serving until rotation)."
|
||||
[] (->BackgroundReindex))
|
||||
|
||||
(defn index-docs!
|
||||
"Indexes the documents. The context should be :search/updating or :search/reindexing.
|
||||
Context should be :search/updating or :search/reindexing to help control how to manage the updates"
|
||||
[context document-reducible]
|
||||
(tracing/with-span :search "search.appdb.index-docs" {:search/context (name context)}
|
||||
(let [reindexing? (and (= :search/reindexing context) (not search.ingestion/*force-sync*))
|
||||
;; Capture the destination table ONCE for the whole reindex: the pending table when a rebuild is
|
||||
;; staging one, otherwise the active table (initial creation populates the freshly-activated table
|
||||
;; directly, with no pending). Resolving this per batch is unsafe -- a concurrent TTL resync of the
|
||||
;; tracking atom can transiently blank :pending, which would otherwise flip the write target to the
|
||||
;; live active table mid-rebuild and silently drop documents from the index we are about to activate.
|
||||
reindex-table (when reindexing? (or (pending-table) (active-table)))]
|
||||
(transduce (comp (partition-all insert-batch-size)
|
||||
(map (partial batch-update! reindex-table)))
|
||||
(partial merge-with +)
|
||||
document-reducible))))
|
||||
(defn in-place-mode
|
||||
"A reindex that clears and repopulates the active table directly."
|
||||
[] (->InPlaceReindex))
|
||||
|
||||
(defmethod search.engine/update! :search.engine/appdb [_engine document-reducible]
|
||||
(index-docs! :search/updating document-reducible))
|
||||
(defn incremental-mode
|
||||
"The live-update path: dual-write to active and pending, no rotation."
|
||||
[] (->IncrementalUpdate))
|
||||
|
||||
(defmethod search.engine/delete! :search.engine/appdb [_engine search-model ids]
|
||||
(when (seq ids)
|
||||
(->> [(active-table) (pending-table)]
|
||||
(keep (fn [table-name]
|
||||
(when table-name
|
||||
{search-model (try (t2/delete! table-name :model search-model :model_id [:in (set ids)])
|
||||
;; Race conditions with table being deleted, especially in tests.
|
||||
(catch Exception e (if (table-not-found-exception? e) 0 (throw e))))})))
|
||||
(apply merge-with +)
|
||||
(into {}))))
|
||||
;;; ------------------------------------------ Search -----------------------------------------------
|
||||
|
||||
(defn when-index-created
|
||||
"Return creation time of the active index, or nil if there is none."
|
||||
@@ -442,57 +260,44 @@
|
||||
(map (juxt :model :name)
|
||||
(t2/query (search-query search-term search-ctx [:model :name]))))
|
||||
|
||||
(defn reset-index!
|
||||
"Ensure we have a blank slate; in case the table schema or stored data format has changed."
|
||||
[]
|
||||
(log/infof "Resetting appdb index for version %s, active table: %s" (search.spec/index-version-hash)
|
||||
(pr-str (active-table)))
|
||||
(letfn [(reset-logic []
|
||||
;; stop tracking any pending table
|
||||
(when-let [table-name (pending-table)]
|
||||
(when-not *mocking-tables*
|
||||
(let [deleted (search-index-metadata/delete-index! :appdb (search.spec/index-version-hash) table-name)]
|
||||
(when (pos? deleted)
|
||||
(log/infof "Deleted %d pending indices" deleted))))
|
||||
(swap! *indexes* assoc :pending nil))
|
||||
(maybe-create-pending!)
|
||||
(activate-table!))]
|
||||
(if search.ingestion/*force-sync*
|
||||
(reset-logic)
|
||||
;; Creates and tracks tables with a unique transaction so the empty tables are available to other threads
|
||||
;; even while the initial startup and data load may be happening
|
||||
(t2/with-connection [_ (mdb/data-source)]
|
||||
(reset-logic)))))
|
||||
|
||||
(defn ensure-ready!
|
||||
"Ensure the index is ready to be populated. Return false if it was already ready."
|
||||
[& {:keys [force-reset?]}]
|
||||
;; Be extra careful against races on initializing the setting
|
||||
(locking *indexes*
|
||||
(when-not *mocking-tables*
|
||||
(when (nil? (active-table))
|
||||
(sync-tracking-atoms!)))
|
||||
(when (or force-reset? (not (exists? (active-table))))
|
||||
(reset-index!))))
|
||||
;;; --------------------------------------- Test helpers --------------------------------------------
|
||||
|
||||
#_{:clj-kondo/ignore [:metabase/test-helpers-use-non-thread-safe-functions]}
|
||||
(defmacro with-temp-index-table
|
||||
"Create a temporary index table for the duration of the body. Uses the existing index if we're already mocking."
|
||||
"Create a single temporary index table for the duration of the body, for tests that write to and read from
|
||||
one fixed index table (ingestion, scoring, search).
|
||||
|
||||
Binds *state-store* to an isolated MockStateStore so DDL within the body doesn't touch the real index, and
|
||||
binds [[metabase.search.spec/*testing-only-index-version-hash*]] to a throwaway version so any metadata a
|
||||
build/rotate inside the body writes (e.g. via [[maybe-create-pending!]]) is scoped to that version and swept
|
||||
up on exit — nothing leaks into the real-version metadata or leaves orphan tables behind.
|
||||
|
||||
NOTE: because the mock's force-refresh! is a no-op, the in-memory active/pending tracking does not follow a
|
||||
rotation. Tests that need to assert against a real pending→active rotation should bind
|
||||
[[metabase.search.spec/*testing-only-index-version-hash*]] and use the real store directly.
|
||||
|
||||
Re-entrant: when already inside a with-temp-index-table (the state store is a mock), it reuses the existing
|
||||
temp table rather than creating a new one, so nested scopes share a single index and content accumulates
|
||||
across them — several callers (e.g. nested with-search-items-in-root-collection) rely on this."
|
||||
[& body]
|
||||
`(if @#'*mocking-tables*
|
||||
~@body
|
||||
(let [table-name# (gen-table-name "_temp")
|
||||
version# (str (string/random-string 8) "-temp")]
|
||||
(binding [*mocking-tables* true
|
||||
*indexes* (atom {:active table-name#})]
|
||||
`(if-not (index-state/db-backed? *state-store*)
|
||||
(do ~@body)
|
||||
(let [table-name# (table/gen-table-name "_temp")
|
||||
version# (str (string/random-string 8) "-temp")]
|
||||
(binding [*state-store* (index-state/mock-store {:active table-name#})
|
||||
search.spec/*testing-only-index-version-hash* version#]
|
||||
(try
|
||||
(t2/insert! :model/SearchIndexMetadata {:engine :appdb
|
||||
:version version#
|
||||
:lang_code (i18n/site-locale-string)
|
||||
:status :pending
|
||||
:index_name (name table-name#)})
|
||||
(create-table! table-name#)
|
||||
(table/create-table! table-name#)
|
||||
~@body
|
||||
(finally
|
||||
(#'drop-table! table-name#)
|
||||
;; Drop every physical table registered under the throwaway version — the temp table plus any a
|
||||
;; reindex inside the body created — then delete the metadata, so nothing leaks into the real index.
|
||||
(doseq [t# (t2/select-fn-set :index_name :model/SearchIndexMetadata :version version#)]
|
||||
(table/drop-table! (keyword t#)))
|
||||
(table/drop-table! table-name#)
|
||||
(t2/delete! :model/SearchIndexMetadata :version version#)))))))
|
||||
|
||||
105
src/metabase/search/appdb/index_state.clj
Normal file
105
src/metabase/search/appdb/index_state.clj
Normal file
@@ -0,0 +1,105 @@
|
||||
(ns metabase.search.appdb.index-state
|
||||
"In-memory cache of the active/pending search index table names, synchronized from the database.
|
||||
|
||||
State machine:
|
||||
:active — the table currently serving search queries; must always exist in DB when non-nil
|
||||
:pending — a table being built during a reindex; nil when no reindex is in progress
|
||||
retired — previous active tables, dropped after rotation; not tracked in memory
|
||||
|
||||
The pending→active transition is performed under a cluster lock.
|
||||
Other nodes observe the new active table on their next [[current-state]] call after TTL
|
||||
expiry, or immediately via [[force-refresh!]].
|
||||
|
||||
Two implementations:
|
||||
[[DbBackedStateStore]] — production; lazily syncs from DB at most once per TTL window
|
||||
[[MockStateStore]] — tests; pure in-memory, no DB interaction"
|
||||
(:import (clojure.lang Atom)))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(defprotocol IndexStateStore
|
||||
(current-state [store]
|
||||
"Return {:active table-or-nil, :pending table-or-nil}.
|
||||
The result may be up to TTL seconds stale; at most one DB read per TTL window.")
|
||||
(force-refresh! [store]
|
||||
"Unconditionally re-read from the backing store and return the new state.")
|
||||
(set-state! [store new-state]
|
||||
"Overwrite the cached :active and :pending values. Call after DDL operations to avoid
|
||||
serving stale data. Resets the TTL so the next DB read is deferred by one full window.")
|
||||
(db-backed? [store]
|
||||
"True if this store reads from the database. False for in-memory mock stores used in tests.
|
||||
When false, operations that modify global DB state (e.g. cleanup of obsolete tables) are
|
||||
skipped so that tests remain isolated from production index state."))
|
||||
|
||||
(def ^:private ttl-ns
|
||||
"How long (in nanoseconds) a cached state may be used before re-reading from the DB."
|
||||
(* 5 60 1000000000))
|
||||
|
||||
(defrecord DbBackedStateStore
|
||||
[^Atom state-atom
|
||||
sync-fn]
|
||||
|
||||
IndexStateStore
|
||||
|
||||
(current-state [_]
|
||||
(let [{:keys [next-sync-at-ns] :as s} @state-atom
|
||||
now-ns (System/nanoTime)]
|
||||
;; If the TTL has expired, try to claim the refresh slot via CAS.
|
||||
;; Only the thread whose CAS succeeds performs the DB read; all others return the
|
||||
;; (momentarily stale) cached value, which is an acceptable trade-off for a 5-minute TTL.
|
||||
(when (or (nil? next-sync-at-ns) (>= now-ns next-sync-at-ns))
|
||||
(when (compare-and-set! state-atom s (assoc s :next-sync-at-ns (+ now-ns ttl-ns)))
|
||||
(let [fresh (sync-fn)]
|
||||
(swap! state-atom #(assoc % :active (:active fresh) :pending (:pending fresh)
|
||||
:next-sync-at-ns (+ (System/nanoTime) ttl-ns)))))))
|
||||
(select-keys @state-atom [:active :pending]))
|
||||
|
||||
(force-refresh! [_]
|
||||
(let [fresh (sync-fn)]
|
||||
(swap! state-atom assoc
|
||||
:active (:active fresh)
|
||||
:pending (:pending fresh)
|
||||
:next-sync-at-ns (+ (System/nanoTime) ttl-ns))
|
||||
(select-keys @state-atom [:active :pending])))
|
||||
|
||||
(set-state! [_ new-state]
|
||||
(swap! state-atom assoc
|
||||
:active (:active new-state)
|
||||
:pending (:pending new-state)
|
||||
:next-sync-at-ns (+ (System/nanoTime) ttl-ns)))
|
||||
|
||||
(db-backed? [_] true))
|
||||
|
||||
(defrecord MockStateStore
|
||||
[^Atom state-atom]
|
||||
|
||||
IndexStateStore
|
||||
|
||||
(current-state [_]
|
||||
@state-atom)
|
||||
|
||||
(force-refresh! [_]
|
||||
;; No backing store to read from — the atom IS the truth in test mode.
|
||||
@state-atom)
|
||||
|
||||
(set-state! [_ new-state]
|
||||
(swap! state-atom assoc
|
||||
:active (:active new-state)
|
||||
:pending (:pending new-state)))
|
||||
|
||||
(db-backed? [_] false))
|
||||
|
||||
(defn db-backed-store
|
||||
"Create a production state store backed by `sync-fn`.
|
||||
`sync-fn` is called with no arguments and must return a map with :active and/or :pending keys
|
||||
whose values are table-name keywords (or nil when absent)."
|
||||
[sync-fn]
|
||||
(->DbBackedStateStore (atom {:active nil :pending nil :next-sync-at-ns nil}) sync-fn))
|
||||
|
||||
(defn mock-store
|
||||
"Create an in-memory state store for use in tests.
|
||||
Optionally accepts an initial state map with :active and/or :pending keys."
|
||||
([]
|
||||
(mock-store {}))
|
||||
([initial]
|
||||
(->MockStateStore (atom {:active (:active initial) :pending (:pending initial)}))))
|
||||
@@ -5,12 +5,13 @@
|
||||
[metabase.analytics.core :as analytics]
|
||||
[metabase.search.appdb.index :as search.index]
|
||||
[metabase.search.appdb.specialization.api :as specialization]
|
||||
[metabase.search.appdb.table :as search.table]
|
||||
[metabase.search.engine :as search.engine]))
|
||||
|
||||
(defn- active-index-size
|
||||
[]
|
||||
(when-let [table (search.index/active-table)]
|
||||
(when (search.index/exists? table)
|
||||
(when (search.table/exists? table)
|
||||
(specialization/index-size-estimate table))))
|
||||
|
||||
;; Keep the :metabase-search/appdb-index-size product gauge up to date (per-instance). Skips the work entirely
|
||||
|
||||
169
src/metabase/search/appdb/table.clj
Normal file
169
src/metabase/search/appdb/table.clj
Normal file
@@ -0,0 +1,169 @@
|
||||
(ns metabase.search.appdb.table
|
||||
"Physical search-index tables: creating, dropping, detecting, and cleaning them up.
|
||||
|
||||
This namespace knows how to manipulate the `search_index__*` tables themselves. It is deliberately
|
||||
unaware of the active/pending state machine (see [[metabase.search.appdb.index-state]]) and of the
|
||||
reindex write pipeline (see [[metabase.search.appdb.index]]) — those layers decide *which* table to
|
||||
touch; this one only knows *how* to create and destroy them."
|
||||
(:require
|
||||
[clojure.string :as str]
|
||||
[honey.sql.helpers :as sql.helpers]
|
||||
[metabase.app-db.core :as mdb]
|
||||
[metabase.search.appdb.specialization.api :as specialization]
|
||||
[metabase.search.appdb.specialization.h2 :as h2]
|
||||
[metabase.search.appdb.specialization.postgres :as postgres]
|
||||
[metabase.search.config :as search.config]
|
||||
[metabase.search.models.search-index-metadata :as search-index-metadata]
|
||||
[metabase.search.spec :as search.spec]
|
||||
[metabase.util :as u]
|
||||
[metabase.util.log :as log]
|
||||
[toucan2.core :as t2])
|
||||
(:import
|
||||
(org.h2.jdbc JdbcSQLSyntaxErrorException)
|
||||
(org.postgresql.util PSQLException)))
|
||||
|
||||
(comment
|
||||
h2/keep-me
|
||||
postgres/keep-me)
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(defn table-name
|
||||
"Normalize a table keyword to the literal name string used by the current app DB (H2 upper-cases)."
|
||||
[kw]
|
||||
(cond-> (name kw)
|
||||
(= :h2 (mdb/db-type)) u/upper-case-en))
|
||||
|
||||
(defn exists?
|
||||
"True if the given index table currently exists in the app DB."
|
||||
[table]
|
||||
(when table
|
||||
(t2/exists? :information_schema.tables :table_name (table-name table))))
|
||||
|
||||
(defn table-not-found-exception?
|
||||
"True if `e` looks like a missing-table error from the DB driver.
|
||||
Used to detect stale index table references — the caller should refresh state and retry.
|
||||
NOTE: can produce false positives on genuinely malformed queries, so only use in contexts
|
||||
where you have already verified the table name came from the index state store."
|
||||
[e]
|
||||
(or (instance? PSQLException (ex-cause e))
|
||||
(instance? JdbcSQLSyntaxErrorException (ex-cause e))))
|
||||
|
||||
(defn gen-table-name
|
||||
"Generate a unique table name to use as a search index table. If no suffix is provided, none will be used"
|
||||
([]
|
||||
(gen-table-name ""))
|
||||
([suffix]
|
||||
(keyword (str (str/replace (str "search_index__" (u/lower-case-en (u/generate-nano-id))) #"-" "_") suffix))))
|
||||
|
||||
(defn drop-table!
|
||||
"Drop the given index table if it exists. Returns true if a drop was issued."
|
||||
[table]
|
||||
(boolean
|
||||
(when table
|
||||
(t2/query (sql.helpers/drop-table :if-exists (keyword (table-name table)))))))
|
||||
|
||||
;;; -------------------------------------------- Schema --------------------------------------------
|
||||
|
||||
(defn- ->db-type [t]
|
||||
(get {:pk :int, :timestamp :timestamp-with-time-zone} t t))
|
||||
|
||||
(defn- ->db-column [c]
|
||||
(or (get {:id :model_id
|
||||
:created-at :model_created_at
|
||||
:updated-at :model_updated_at}
|
||||
c)
|
||||
(keyword (u/->snake_case_en (name c)))))
|
||||
|
||||
(def ^:private not-null
|
||||
#{:archived :name})
|
||||
|
||||
(def ^:private default
|
||||
{:archived false})
|
||||
|
||||
;; If this fails, we'll need to increase the size of :model below
|
||||
(assert (>= 32 (transduce (map (comp count name)) max 0 search.config/all-models)))
|
||||
|
||||
(def ^:private base-schema
|
||||
(into [[:model [:varchar 32] :not-null]
|
||||
[:display_data :text :not-null]
|
||||
[:legacy_input :text :not-null]
|
||||
;; useful for tracking the speed and age of the index
|
||||
[:created_at :timestamp-with-time-zone
|
||||
[:default [:raw "CURRENT_TIMESTAMP"]]
|
||||
:not-null]
|
||||
[:updated_at :timestamp-with-time-zone :not-null]]
|
||||
(keep (fn [[k t]]
|
||||
(when t
|
||||
(into [(->db-column k) (->db-type t)]
|
||||
(concat
|
||||
(when (not-null k)
|
||||
[:not-null])
|
||||
(when-some [d (default k)]
|
||||
[[:default d]]))))))
|
||||
search.spec/attr-types))
|
||||
|
||||
(defn create-table!
|
||||
"Create an index table with the given name. Should fail if it already exists."
|
||||
[table-name]
|
||||
;; Create with a separate transaction so that postgresql will complete the index creations before returning,
|
||||
;; even when already running in a transaction
|
||||
(t2/with-transaction [_ (mdb/app-db)]
|
||||
(-> (sql.helpers/create-table table-name)
|
||||
(sql.helpers/with-columns (specialization/table-schema base-schema))
|
||||
t2/query)
|
||||
(let [table-name (name table-name)]
|
||||
(doseq [stmt (specialization/post-create-statements table-name table-name)]
|
||||
(t2/query stmt)))))
|
||||
|
||||
;;; --------------------------------------- Orphan cleanup -----------------------------------------
|
||||
|
||||
(defn- orphan-indexes []
|
||||
(map (comp keyword u/lower-case-en :table_name)
|
||||
(t2/query {:select [:table_name]
|
||||
:from :information_schema.tables
|
||||
:where [:and
|
||||
[:= :table_schema :%current_schema]
|
||||
[:or
|
||||
[:like [:lower :table_name] [:inline "search\\_index\\_\\_%"]]
|
||||
;; legacy table names
|
||||
[:in [:lower :table_name]
|
||||
(mapv #(vector :inline %) ["search_index" "search_index_next" "search_index_retired"])]]
|
||||
;; Exclude temp tables — they are managed by with-temp-index-table
|
||||
[:not-like [:lower :table_name] [:inline "%\\_temp"]]
|
||||
[:not-in [:lower :table_name]
|
||||
{:select [:%lower.index_name]
|
||||
:from [(t2/table-name :model/SearchIndexMetadata)]
|
||||
:where [:= :engine [:inline "appdb"]]}]]})))
|
||||
|
||||
(defn- drop-orphan-table!
|
||||
"Drop an orphaned index `table`, bounding how long the DROP will wait for the table's lock.
|
||||
|
||||
[[delete-obsolete-tables!]] runs while the cluster reindex lock is held (it's called from `activate-table!`
|
||||
during a reindex). A `DROP TABLE` needs an exclusive lock, so if the orphan is still locked by a concurrent
|
||||
search transaction the DROP blocks indefinitely -- holding the cluster lock and deadlocking every other
|
||||
reindex behind it. Run the drop in its own transaction with a short `lock_timeout` so a contended drop fails
|
||||
fast (and is then skipped by the caller); the orphan is cleaned up on a later pass. Only Postgres needs this --
|
||||
H2 has a short default lock timeout and the appdb engine only runs on those two."
|
||||
[table]
|
||||
(t2/with-transaction [_ (mdb/app-db)]
|
||||
(when (= :postgres (mdb/db-type))
|
||||
(t2/query ["SET LOCAL lock_timeout = '5s'"]))
|
||||
(t2/query (sql.helpers/drop-table table))))
|
||||
|
||||
(defn delete-obsolete-tables!
|
||||
"Delete metadata for obsolete indexes and drop any physical tables no longer referenced by metadata."
|
||||
[]
|
||||
;; Delete metadata around indexes that are no longer needed.
|
||||
(search-index-metadata/delete-obsolete! (search.spec/index-version-hash))
|
||||
;; Drop any indexes that are no longer referenced.
|
||||
(let [dropped (volatile! [])]
|
||||
(doseq [table (orphan-indexes)]
|
||||
(try
|
||||
(drop-orphan-table! table)
|
||||
(vswap! dropped conj table)
|
||||
;; Deletion could fail if it races with other instances, or if the table is still locked (we use a short
|
||||
;; lock timeout to avoid wedging) -- either way just skip it; it'll be retried on a later pass.
|
||||
(catch Exception e
|
||||
(log/warnf e "Failed to drop stale index %s" table))))
|
||||
(log/infof "Dropped %d stale indexes: %s" (count @dropped) @dropped)))
|
||||
165
src/metabase/search/appdb/writer.clj
Normal file
165
src/metabase/search/appdb/writer.clj
Normal file
@@ -0,0 +1,165 @@
|
||||
(ns metabase.search.appdb.writer
|
||||
"The write pipeline that populates the appdb search index: turning a stream of documents into batched,
|
||||
resilient upserts against the active and/or pending tables.
|
||||
|
||||
This is the *mechanism* half of the indexer; the *policy* half — which table is active/pending and how a
|
||||
reindex rotates them — lives in [[metabase.search.appdb.index]]. [[write-plan]] turns a
|
||||
[[metabase.search.appdb.index/ReindexMode]] into the complete per-batch decision: which physical tables to
|
||||
write to and whether to COMMIT each batch in a dedicated connection.
|
||||
|
||||
The core resilience guarantee is [[safe-batch-upsert!]]: a single batch that lands on a table which was
|
||||
dropped out from under us (a concurrent rotation on another node) triggers one transparent state refresh
|
||||
and retry; any other write failure is logged and skipped so the rest of a reindex still completes."
|
||||
(:require
|
||||
[metabase.analytics-interface.core :as analytics]
|
||||
[metabase.app-db.core :as mdb]
|
||||
[metabase.search.appdb.document :as document]
|
||||
[metabase.search.appdb.index :as index]
|
||||
[metabase.search.appdb.index-state :as index-state]
|
||||
[metabase.search.appdb.specialization.api :as specialization]
|
||||
[metabase.search.appdb.table :as table]
|
||||
[metabase.search.engine :as search.engine]
|
||||
[metabase.tracing.core :as tracing]
|
||||
[metabase.util :as u]
|
||||
[metabase.util.log :as log]
|
||||
[toucan2.core :as t2]))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(def ^:private insert-batch-size 150)
|
||||
|
||||
(defn- write-plan
|
||||
"The complete per-batch write decision for `mode`: which `[label table-name-fn]` pairs to upsert into, and
|
||||
whether each batch should COMMIT in a dedicated connection.
|
||||
|
||||
A committing background rebuild targets ONLY the `reindex-table` captured once at the start of the run
|
||||
(see [[index-docs!]]) — the active table keeps serving and is swapped in by [[index/activate-table!]].
|
||||
Pinning to the captured table means a concurrent state resync that transiently blanks :pending can't
|
||||
redirect writes into the live active table mid-rebuild and silently drop whole models from the new index.
|
||||
Every other write dual-writes to active AND pending so that an in-flight rebuild stays current with live
|
||||
edits (pending may be nil, in which case it is simply skipped), and runs inside the caller's transaction
|
||||
with no per-batch COMMIT."
|
||||
[mode reindex-table]
|
||||
(let [commit? (index/commits-per-batch? mode)]
|
||||
{:commit? commit?
|
||||
:targets (if (and commit? reindex-table)
|
||||
[[:pending (constantly reindex-table)]]
|
||||
[[:active index/active-table] [:pending index/pending-table]])}))
|
||||
|
||||
(defn- do-batch-upsert!
|
||||
"Write entries to the given table. Returns the table name on success. No error handling."
|
||||
[table-name entries]
|
||||
(specialization/batch-upsert! table-name entries)
|
||||
table-name)
|
||||
|
||||
(defn- classify-upsert-error
|
||||
"Return :interrupted, :stale-table, or :unknown-error for a caught upsert exception."
|
||||
[e table-name]
|
||||
(cond
|
||||
(instance? InterruptedException e) :interrupted
|
||||
(and (table/table-not-found-exception? e) (not (table/exists? table-name))) :stale-table
|
||||
:else :unknown-error))
|
||||
|
||||
(defn- record-skipped-batch!
|
||||
"Log an upsert failure at ERROR level, increment analytics, and return nil so the reindex continues."
|
||||
[table-type name-before name-after e-before e-after]
|
||||
(let [wrapped (ex-info "Failed retrying search index batch upsert"
|
||||
{:table-type table-type
|
||||
:table-name-before-refresh name-before
|
||||
:table-name-after-refresh name-after
|
||||
:initial-exception-class (class e-before)
|
||||
:initial-exception-message (ex-message e-before)}
|
||||
e-after)]
|
||||
(analytics/inc! :metabase-search/appdb-index-batches-skipped {:table-type table-type})
|
||||
(log/errorf wrapped "Error upserting search index batch into %s table; skipping batch and continuing"
|
||||
(name table-type))
|
||||
nil))
|
||||
|
||||
(defn- try-retry-after-sync!
|
||||
"After a stale-table error, force-refresh state and retry the upsert once into the new table.
|
||||
Returns the new table name on success, or nil if the retry also fails."
|
||||
[table-type orig-name table-name-fn entries orig-ex]
|
||||
(index-state/force-refresh! index/*state-store*)
|
||||
(if-let [new-name (table-name-fn)]
|
||||
(if (= orig-name new-name)
|
||||
;; The refreshed state still points at the same missing table — something is genuinely wrong.
|
||||
(throw (ex-info "Currently tracked index does not exist" orig-ex {:table-name orig-name}))
|
||||
(try
|
||||
(do-batch-upsert! new-name entries)
|
||||
(catch InterruptedException ie
|
||||
(.interrupt (Thread/currentThread))
|
||||
(throw ie))
|
||||
(catch Exception e2
|
||||
(record-skipped-batch! table-type orig-name new-name orig-ex e2))))
|
||||
;; After refresh there is no table to write to — nothing to do.
|
||||
nil))
|
||||
|
||||
(defn- safe-batch-upsert!
|
||||
"Write entries to the tracked table, recovering once from stale index metadata.
|
||||
Returns the table name written to, or nil when there is no tracked table or the write failed.
|
||||
Failures are logged at ERROR and skipped so the rest of a reindex can still finish."
|
||||
[table-type table-name-fn entries]
|
||||
(when-let [table-name (table-name-fn)]
|
||||
(try
|
||||
(do-batch-upsert! table-name entries)
|
||||
(catch InterruptedException ie
|
||||
(.interrupt (Thread/currentThread))
|
||||
(throw ie))
|
||||
(catch Exception e
|
||||
(case (classify-upsert-error e table-name)
|
||||
:stale-table (try-retry-after-sync! table-type table-name table-name-fn entries e)
|
||||
:unknown-error (record-skipped-batch! table-type table-name nil e nil))))))
|
||||
|
||||
(defn- batch-update!
|
||||
"Upsert one batch of documents into the tables chosen by `mode` (see [[write-plan]]).
|
||||
When the mode commits per batch, the writes run in a dedicated connection and each batch issues an
|
||||
explicit COMMIT so the partial build is visible to other connections during a long rebuild."
|
||||
[mode reindex-table documents]
|
||||
(let [{:keys [commit? targets]} (write-plan mode reindex-table)
|
||||
do-writes (fn []
|
||||
(let [entries (map document/document->entry documents)
|
||||
written (into {} (map (fn [[label table-name-fn]]
|
||||
[label (safe-batch-upsert! label table-name-fn entries)]))
|
||||
targets)]
|
||||
(when (some some? (vals written))
|
||||
(u/prog1 (->> entries (map :model) frequencies)
|
||||
(when commit?
|
||||
(t2/query ["commit"]))
|
||||
(log/trace "indexed documents for " <>)))))]
|
||||
(if commit?
|
||||
;; New connection used for performing the updates which commit periodically without impacting any outer transactions.
|
||||
(t2/with-connection [_conn (mdb/data-source)]
|
||||
(do-writes))
|
||||
(do-writes))))
|
||||
|
||||
(defn index-docs!
|
||||
"Index the documents from `document-reducible` using the given reindex `mode`, returning per-model counts.
|
||||
`mode` is one of [[index/background-mode]], [[index/in-place-mode]], or [[index/incremental-mode]].
|
||||
|
||||
For a committing reindex the destination table is captured ONCE here — the pending table when a rebuild is
|
||||
staging one, otherwise the active table for an initial build — so that a concurrent TTL resync that
|
||||
transiently blanks the tracking state cannot redirect later batches into the live active table and silently
|
||||
drop whole models from the index we are about to activate (see [[write-plan]])."
|
||||
[mode document-reducible]
|
||||
(tracing/with-span :search "search.appdb.index-docs" {:search/mode (.getSimpleName (class mode))}
|
||||
(let [reindex-table (when (index/commits-per-batch? mode)
|
||||
(or (index/pending-table) (index/active-table)))]
|
||||
(transduce (comp (partition-all insert-batch-size)
|
||||
(map (partial batch-update! mode reindex-table)))
|
||||
(partial merge-with +)
|
||||
document-reducible))))
|
||||
|
||||
(defmethod search.engine/update! :search.engine/appdb [_engine document-reducible]
|
||||
(index-docs! (index/incremental-mode) document-reducible))
|
||||
|
||||
(defmethod search.engine/delete! :search.engine/appdb [_engine search-model ids]
|
||||
(when (seq ids)
|
||||
;; The index-size gauge is refreshed out-of-band by metabase.search.appdb.metrics, so we don't measure here.
|
||||
(->> [(index/active-table) (index/pending-table)]
|
||||
(keep (fn [table-name]
|
||||
(when table-name
|
||||
{search-model (try (t2/delete! table-name :model search-model :model_id [:in (set ids)])
|
||||
;; Race conditions with table being deleted, especially in tests.
|
||||
(catch Exception e (if (table/table-not-found-exception? e) 0 (throw e))))})))
|
||||
(apply merge-with +)
|
||||
(into {}))))
|
||||
135
src/metabase/search/index_queue.clj
Normal file
135
src/metabase/search/index_queue.clj
Normal file
@@ -0,0 +1,135 @@
|
||||
(ns metabase.search.index-queue
|
||||
"Owns the asynchronous search-index ingestion queue: the backing queue, the listener
|
||||
lifecycle, and in-flight message accounting so that idleness can be detected precisely.
|
||||
|
||||
Two implementations mirror the [[metabase.search.appdb.index-state]] pattern:
|
||||
[[AsyncIngestionQueue]] — production; a DelayQueue drained by a background listener.
|
||||
[[SyncIngestionQueue]] — alternative/test wiring; runs the handler inline on the
|
||||
calling thread, so it is always idle.
|
||||
|
||||
In-flight accounting: a message counts as pending from the moment it is enqueued until
|
||||
the batch that contains it finishes processing. `pending` is therefore (queued + being
|
||||
processed) with no gap, so [[idle?]] is exact — unlike polling the raw queue size, which
|
||||
reads zero in the window after the listener dequeues a batch but before it has finished
|
||||
writing it to the index."
|
||||
(:require
|
||||
[metabase.util.log :as log]
|
||||
[metabase.util.queue :as queue])
|
||||
(:import
|
||||
(java.util.concurrent DelayQueue)
|
||||
(java.util.concurrent.atomic AtomicLong)))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(defprotocol IngestionQueue
|
||||
(enqueue! [q updates]
|
||||
"Submit a sequence of updates for asynchronous processing. Returns true.")
|
||||
(pending-count [q]
|
||||
"Number of messages enqueued but not yet fully processed (queued + in-flight).")
|
||||
(await-idle! [q opts]
|
||||
"Block until the queue is idle, or until `:timeout-ms` (default 30000) elapses.
|
||||
Returns true if idle, false on timeout.")
|
||||
(clear! [q]
|
||||
"Discard all not-yet-taken messages and reset accounting. Intended for test/restore
|
||||
reset when the system is quiescent.")
|
||||
(running? [q]
|
||||
"True if the queue is actively processing (listener running; always true for the sync impl).")
|
||||
(start! [q]
|
||||
"Begin asynchronous processing. Logs and no-ops if already running.")
|
||||
(stop! [q]
|
||||
"Stop asynchronous processing. No-op for the sync impl."))
|
||||
|
||||
(defn idle?
|
||||
"True when the queue has no pending work."
|
||||
[q]
|
||||
(zero? (pending-count q)))
|
||||
|
||||
(defn- wrap-handler
|
||||
"Wrap the user handler so that finishing a batch decrements the in-flight counter and
|
||||
refreshes the size gauge, even if the handler throws."
|
||||
[handler ^AtomicLong pending ^DelayQueue queue gauge-fn]
|
||||
(fn [batch]
|
||||
(try
|
||||
(handler batch)
|
||||
(finally
|
||||
(.addAndGet pending (- (count batch)))
|
||||
(when gauge-fn (gauge-fn (.size queue)))))))
|
||||
|
||||
(defrecord AsyncIngestionQueue [^DelayQueue queue
|
||||
^AtomicLong pending
|
||||
^long delay-ms
|
||||
listener-name
|
||||
handler
|
||||
listener-opts
|
||||
gauge-fn]
|
||||
IngestionQueue
|
||||
(enqueue! [_ updates]
|
||||
(doseq [u updates]
|
||||
(.incrementAndGet pending)
|
||||
(queue/put-with-delay! queue delay-ms u))
|
||||
(when gauge-fn (gauge-fn (.size queue)))
|
||||
true)
|
||||
|
||||
(pending-count [_]
|
||||
;; The counter is bumped on enqueue and decremented only once a batch finishes processing,
|
||||
;; so it spans the gap between dequeue and index write that a raw `.size` read misses.
|
||||
(max 0 (.get pending)))
|
||||
|
||||
(await-idle! [this {:keys [timeout-ms poll-ms] :or {timeout-ms 30000 poll-ms 25}}]
|
||||
(let [deadline (+ (System/nanoTime) (* (long timeout-ms) 1000000))]
|
||||
(loop []
|
||||
(cond
|
||||
(idle? this) true
|
||||
(>= (System/nanoTime) deadline) (do (log/warnf "await-idle! timed out with %d messages still pending"
|
||||
(pending-count this))
|
||||
false)
|
||||
:else (do (Thread/sleep (long poll-ms)) (recur))))))
|
||||
|
||||
(clear! [_]
|
||||
(.clear queue)
|
||||
(.set pending 0)
|
||||
(when gauge-fn (gauge-fn 0)))
|
||||
|
||||
(running? [_]
|
||||
(queue/listener-exists? listener-name))
|
||||
|
||||
(start! [this]
|
||||
(queue/listen! listener-name queue (wrap-handler handler pending queue gauge-fn) listener-opts)
|
||||
this)
|
||||
|
||||
(stop! [this]
|
||||
(queue/stop-listening! listener-name)
|
||||
this))
|
||||
|
||||
(defrecord SyncIngestionQueue [handler]
|
||||
IngestionQueue
|
||||
(enqueue! [_ updates] (handler updates) true)
|
||||
(pending-count [_] 0)
|
||||
(await-idle! [_ _opts] true)
|
||||
(clear! [_] nil)
|
||||
(running? [_] true)
|
||||
(start! [this] this)
|
||||
(stop! [this] this))
|
||||
|
||||
(defn async-queue
|
||||
"Create a production async ingestion queue.
|
||||
|
||||
Options:
|
||||
- :delay-ms delay applied to each message before it becomes takeable (commit-settle window)
|
||||
- :listener-name unique listener name
|
||||
- :handler fn of a batch (sequence of messages) that performs the indexing
|
||||
- :listener-opts options passed to [[metabase.util.queue/listen!]] (success/err handlers, batching)
|
||||
- :gauge-fn optional fn of a number, called with the queue size to update a metric"
|
||||
[{:keys [delay-ms listener-name handler listener-opts gauge-fn]}]
|
||||
(->AsyncIngestionQueue (queue/delay-queue)
|
||||
(AtomicLong. 0)
|
||||
(long delay-ms)
|
||||
listener-name
|
||||
handler
|
||||
(or listener-opts {})
|
||||
gauge-fn))
|
||||
|
||||
(defn sync-queue
|
||||
"Create an ingestion queue that processes inline on the calling thread (no listener)."
|
||||
[handler]
|
||||
(->SyncIngestionQueue handler))
|
||||
@@ -7,23 +7,18 @@
|
||||
[metabase.app-db.core :as mdb]
|
||||
[metabase.lib-be.core :as lib-be]
|
||||
[metabase.search.engine :as search.engine]
|
||||
[metabase.search.index-queue :as index-queue]
|
||||
[metabase.search.spec :as search.spec]
|
||||
[metabase.tracing.core :as tracing]
|
||||
[metabase.util :as u]
|
||||
[metabase.util.json :as json]
|
||||
[metabase.util.log :as log]
|
||||
[metabase.util.performance :as perf]
|
||||
[metabase.util.queue :as queue]
|
||||
[toucan2.core :as t2]
|
||||
[toucan2.realize :as t2.realize])
|
||||
(:import
|
||||
(java.util.concurrent DelayQueue)))
|
||||
[toucan2.realize :as t2.realize]))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
;; Currently we use a single queue, even if multiple engines are enabled, but may want to revisit this.
|
||||
(defonce ^:private ^DelayQueue queue (queue/delay-queue))
|
||||
|
||||
;; Perhaps this config move up somewhere more visible? Conversely, we may want to specialize it per engine.
|
||||
|
||||
(def ^:private message-delay-ms
|
||||
@@ -352,72 +347,56 @@
|
||||
(update! documents to-delete))
|
||||
{}))))
|
||||
|
||||
(defn- track-queue-size! []
|
||||
(analytics/set-gauge! :metabase-search/queue-size (.size queue)))
|
||||
|
||||
(defn- index-worker-exists? []
|
||||
(queue/listener-exists? listener-name))
|
||||
|
||||
(defn ingest-maybe-async!
|
||||
"Update or create any search index entries related to the given updates.
|
||||
Will be async if the worker exists, otherwise it will be done synchronously on the calling thread.
|
||||
Can also be forced to run synchronously for testing."
|
||||
([updates]
|
||||
(ingest-maybe-async! updates (or *force-sync* (not (index-worker-exists?)))))
|
||||
([updates sync?]
|
||||
(when-not *disable-updates*
|
||||
(if sync?
|
||||
(bulk-ingest! updates)
|
||||
(do
|
||||
(doseq [update updates]
|
||||
(log/trace "Queuing update" update)
|
||||
(queue/put-with-delay! queue message-delay-ms update))
|
||||
(track-queue-size!)
|
||||
true)))))
|
||||
|
||||
(defn wait-for-idle!
|
||||
"Block until the ingestion queue has been drained and no more items arrive.
|
||||
Polls the queue and considers it idle when it has been empty for `settle-ms`
|
||||
(default 500ms). Gives up after `timeout-ms` (default 30s)."
|
||||
[& {:keys [settle-ms timeout-ms poll-ms]
|
||||
:or {settle-ms 500, timeout-ms 30000, poll-ms 50}}]
|
||||
(let [deadline-ns (+ (System/nanoTime) (* (long timeout-ms) 1000000))
|
||||
settle-ns (* (long settle-ms) 1000000)
|
||||
poll-ms (long poll-ms)
|
||||
empty-since (atom nil)]
|
||||
(loop []
|
||||
(let [now-ns (System/nanoTime)
|
||||
sz (.size queue)
|
||||
remaining-ns (- settle-ns (- now-ns (or @empty-since now-ns)))]
|
||||
(cond
|
||||
(>= now-ns deadline-ns)
|
||||
(log/warnf "wait-for-idle! timed out with %d items still in the queue" sz)
|
||||
|
||||
(pos? sz)
|
||||
(do (reset! empty-since nil)
|
||||
(Thread/sleep poll-ms)
|
||||
(recur))
|
||||
|
||||
(pos? remaining-ns)
|
||||
(do (when-not @empty-since (reset! empty-since now-ns))
|
||||
(Thread/sleep (long (max poll-ms (quot remaining-ns 1000000))))
|
||||
(recur)))))))
|
||||
|
||||
(defn start-listener!
|
||||
"Starts the ingestion listener on the queue"
|
||||
[]
|
||||
(when (seq (search.engine/active-engines))
|
||||
(queue/listen! listener-name queue bulk-ingest!
|
||||
{:success-handler (fn [_result _duration _]
|
||||
(track-queue-size!))
|
||||
:err-handler (fn [err _]
|
||||
(defonce ^:private index-queue
|
||||
;; A single queue, even if multiple engines are enabled, but we may want to revisit this.
|
||||
(index-queue/async-queue
|
||||
{:delay-ms message-delay-ms
|
||||
:listener-name listener-name
|
||||
;; resolve the var at call time so test redefs of `bulk-ingest!` take effect
|
||||
:handler (fn [batch] (bulk-ingest! batch))
|
||||
:gauge-fn (fn [n] (analytics/set-gauge! :metabase-search/queue-size n))
|
||||
:listener-opts {:err-handler (fn [err _listener-name]
|
||||
(log/error err "Error indexing search entries")
|
||||
(analytics/inc! :metabase-search/index-error)
|
||||
(track-queue-size!))
|
||||
(analytics/inc! :metabase-search/index-error))
|
||||
;; Note that each message can correspond to multiple documents,
|
||||
;; for example there would be 1 message for updating all
|
||||
;; the tables within a given database when it is renamed.
|
||||
;; Messages can also correspond to zero documents,
|
||||
;; such as when updating a table that is marked as not visible.
|
||||
:max-batch-messages 50
|
||||
:max-next-ms 100})))
|
||||
:max-next-ms 100}}))
|
||||
|
||||
(defn ingest-maybe-async!
|
||||
"Update or create any search index entries related to the given updates.
|
||||
Will be async if the worker exists, otherwise it will be done synchronously on the calling thread.
|
||||
Can also be forced to run synchronously for testing."
|
||||
([updates]
|
||||
(ingest-maybe-async! updates (or *force-sync* (not (index-queue/running? index-queue)))))
|
||||
([updates sync?]
|
||||
(when-not *disable-updates*
|
||||
(if sync?
|
||||
(bulk-ingest! updates)
|
||||
(index-queue/enqueue! index-queue updates)))))
|
||||
|
||||
(defn wait-for-idle!
|
||||
"Block until the ingestion queue has been fully drained — including any batch currently
|
||||
being processed. Accepts `:timeout-ms` (default 30s) and `:poll-ms`.
|
||||
Returns true if the queue became idle, false on timeout."
|
||||
[& {:as opts}]
|
||||
(index-queue/await-idle! index-queue opts))
|
||||
|
||||
(defn pending-count
|
||||
"Number of search-index updates that have been enqueued but not yet fully processed."
|
||||
[]
|
||||
(index-queue/pending-count index-queue))
|
||||
|
||||
(defn clear-queue!
|
||||
"Discard all not-yet-processed queued updates. Intended for test/restore reset."
|
||||
[]
|
||||
(index-queue/clear! index-queue))
|
||||
|
||||
(defn start-listener!
|
||||
"Starts the ingestion listener on the queue"
|
||||
[]
|
||||
(when (seq (search.engine/active-engines))
|
||||
(index-queue/start! index-queue)))
|
||||
|
||||
@@ -90,10 +90,22 @@
|
||||
:order-by [[[:max :updated_at] :desc]
|
||||
[[:max :id] :desc]]
|
||||
:limit 3})))]
|
||||
(t2/query-one {:delete-from [(t2/table-name :model/SearchIndexMetadata)]
|
||||
:where [:or
|
||||
[:not-in :version most-recent]
|
||||
;; Drop those older than 1 day, unless we are using them, or they are the most recent.
|
||||
[:and
|
||||
[:not-in :version (filter some? [our-version (first most-recent)])]
|
||||
[:< :updated_at (t/minus (t/zoned-date-time) pending-table-cut-off)]]]})))
|
||||
(let [;; Versions still serving an :active index. They must never be pruned just for being crowded out of the
|
||||
;; most-recent window: dropping a live version's metadata orphans its active table (the engine then
|
||||
;; sweeps it), breaking in-flight reads/writes on every node still pointing at it. Without this guard,
|
||||
;; transient/test-only versions can crowd the live version out of the window and delete it out from
|
||||
;; under search. Such a version is only retired by the age-based clause below. (Materialized rather than
|
||||
;; a NOT IN sub-select for portability — H2 mishandles NOT IN over an empty sub-select.)
|
||||
serving (seq (map :version (t2/query {:select [[[:distinct :version]]]
|
||||
:from [(t2/table-name :model/SearchIndexMetadata)]
|
||||
:where [:= :status "active"]})))
|
||||
out-of-window (if serving
|
||||
[:and [:not-in :version most-recent] [:not-in :version serving]]
|
||||
[:not-in :version most-recent])]
|
||||
(t2/query-one {:delete-from [(t2/table-name :model/SearchIndexMetadata)]
|
||||
:where [:or
|
||||
out-of-window
|
||||
;; Drop those older than 1 day, unless we are using them, or they are the most recent.
|
||||
[:and
|
||||
[:not-in :version (filter some? [our-version (first most-recent)])]
|
||||
[:< :updated_at (t/minus (t/zoned-date-time) pending-table-cut-off)]]]}))))
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
[clojurewerkz.quartzite.jobs :as jobs]
|
||||
[clojurewerkz.quartzite.schedule.simple :as simple]
|
||||
[clojurewerkz.quartzite.triggers :as triggers]
|
||||
[metabase.app-db.cluster-lock :as cluster-lock]
|
||||
[metabase.search.core :as search]
|
||||
[metabase.search.ingestion :as ingestion]
|
||||
[metabase.startup.core :as startup]
|
||||
@@ -19,7 +18,6 @@
|
||||
|
||||
(def ^:private init-stem "metabase.task.search-index.init")
|
||||
(def ^:private reindex-stem "metabase.task.search-index.reindex")
|
||||
(def ^:private cluster-lock-name ::search-index-lock)
|
||||
|
||||
(def init-job-key
|
||||
"Key used to define and trigger a job that ensures there is an active index."
|
||||
@@ -31,19 +29,21 @@
|
||||
|
||||
;; We define the job bodies outside the defrecord, so that we can redefine them live from the REPL
|
||||
|
||||
;; The cluster-wide lock that ensures only one node builds the index at a time lives at the appdb engine's
|
||||
;; init!/reindex! choke points (see metabase.search.appdb.core), so every trigger — this job, the locale
|
||||
;; change, the API force-reindex, late-init recovery — is covered, not just the scheduled paths here.
|
||||
|
||||
(defn init!
|
||||
"Create a new index, if necessary"
|
||||
[]
|
||||
(when (search/supports-index?)
|
||||
(tracing/with-span :search "search.task.init" {}
|
||||
(cluster-lock/with-cluster-lock cluster-lock-name
|
||||
(search/init-index! {:force-reset? false, :re-populate? false})))))
|
||||
(search/init-index! {:force-reset? false, :re-populate? false}))))
|
||||
|
||||
(task/defjob ^{DisallowConcurrentExecution true
|
||||
:doc "Populate a new Search Index"}
|
||||
SearchIndexReindex [_ctx]
|
||||
(cluster-lock/with-cluster-lock cluster-lock-name
|
||||
(search/reindex! {:async? false})))
|
||||
(search/reindex! {:async? false}))
|
||||
|
||||
;; Atom holding a promise that is delivered when the background init thread finishes.
|
||||
;; nil when no init has been started — [[wait-for-init!]] returns immediately in that case.
|
||||
|
||||
@@ -28,7 +28,6 @@
|
||||
[toucan2.core :as t2])
|
||||
(:import
|
||||
(com.mchange.v2.c3p0 PoolBackedDataSource)
|
||||
(java.util Queue)
|
||||
(java.util.concurrent.locks ReentrantReadWriteLock)))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
@@ -140,7 +139,7 @@
|
||||
[:name ms/NonBlankString]]]
|
||||
;; reset the system clock, in case `/set-time` was called without cleanup
|
||||
(alter-var-root #'java-time.clock/*clock* (constantly nil))
|
||||
(.clear ^Queue @#'search.ingestion/queue)
|
||||
(search.ingestion/clear-queue!)
|
||||
(restore-snapshot! snapshot-name)
|
||||
(search/sync-from-restored-db!)
|
||||
nil)
|
||||
|
||||
@@ -65,6 +65,36 @@
|
||||
(let [data-source (mdb.data-source/raw-connection-string->DataSource (str "jdbc:h2:mem:" (random-schema-migrations-test-db-name)))]
|
||||
(f data-source)))
|
||||
|
||||
(defn- do-with-isolated-search-state
|
||||
"Run `thunk` with the appdb search state store bound to a throwaway, isolated in-memory (mock) store, and with
|
||||
search ingestion forced synchronous.
|
||||
|
||||
This temp app DB swaps out the JVM-global `*application-db*`, but the search index's state store
|
||||
([[metabase.search.appdb.index/*state-store*]]) is a separate JVM-global atom that is NOT swapped. Code run
|
||||
against this temp DB can build/rotate a search index -- e.g. installing the audit DB triggers a serialization
|
||||
load that fires a `(search/reindex!)` -- which would otherwise point the global state store at a table that
|
||||
exists only in this temp DB. When the temp DB is torn down the table vanishes, but the global pointer persists
|
||||
into the real app DB, so later ingestion upserts hit a missing relation, abort the transaction, and poison
|
||||
pooled connections -- cascading into dozens of unrelated tests.
|
||||
|
||||
Binding an isolated *mock* store (rather than a db-backed one) keeps the rebuild off the global store AND off
|
||||
the cluster reindex lock: a reindex against a non-db-backed store skips `cluster-lock/with-cluster-lock`
|
||||
entirely. That matters because the cluster lock is a single JVM-global lock shared across every (H2) temp DB,
|
||||
so letting these throwaway rebuilds take it serializes them and can wedge the whole partition. Forcing
|
||||
`*force-sync*` keeps the rebuild synchronous so it finishes inside the temp DB's lifetime rather than as a
|
||||
future that outlives it (blocking on a torn-down connection pool). Both bindings are conveyed to any future
|
||||
spawned in the body.
|
||||
|
||||
No-op when the search index namespace isn't loaded (pure migration runs)."
|
||||
[thunk]
|
||||
(if-let [store-var (some-> (find-ns 'metabase.search.appdb.index) (ns-resolve '*state-store*))]
|
||||
(let [mock-store (ns-resolve (find-ns 'metabase.search.appdb.index-state) 'mock-store)
|
||||
force-sync-var (some-> (find-ns 'metabase.search.ingestion) (ns-resolve '*force-sync*))]
|
||||
(with-bindings (cond-> {store-var (mock-store)}
|
||||
force-sync-var (assoc force-sync-var true))
|
||||
(thunk)))
|
||||
(thunk)))
|
||||
|
||||
(defn do-with-temp-empty-app-db
|
||||
"The function invoked by [[with-temp-empty-app-db]] to execute function `f` in a temporary, empty app DB. Use the
|
||||
macro instead: [[with-temp-empty-app-db]]."
|
||||
@@ -78,7 +108,7 @@
|
||||
(with-open [conn (.getConnection data-source)]
|
||||
(binding [mdb.connection/*application-db* (mdb.connection/application-db driver data-source)
|
||||
custom-migrations.util/*allow-temp-scheduling* false]
|
||||
(f conn))))))
|
||||
(do-with-isolated-search-state #(f conn)))))))
|
||||
|
||||
(defmacro with-temp-empty-app-db
|
||||
"Create a new temporary application DB of `db-type` and execute `body` with `conn-binding` bound to a
|
||||
|
||||
@@ -1941,8 +1941,15 @@
|
||||
|
||||
(deftest ^:synchronized multiple-limits-test
|
||||
(when (search/supports-index?)
|
||||
;; This test is failing with "no index" for some reason, forcing the reindex
|
||||
(mt/user-real-request :crowberto :post 200 "search/force-reindex"))
|
||||
;; This test hits the real (non-temp) index via user-real-request, so it has to populate it itself.
|
||||
;; force-reindex rebuilds asynchronously (a Quartz job triggered by the endpoint in test mode), so we
|
||||
;; can't assert against the index immediately — we'd race the rebuild and hit a transient empty /
|
||||
;; "Search Index not found" (500) state. Poll until the index is actually queryable and populated.
|
||||
(mt/user-real-request :crowberto :post 200 "search/force-reindex")
|
||||
(u/poll {:thunk #(mt/user-real-request :crowberto :get "search?q=product")
|
||||
:done? #(seq (:data %))
|
||||
:timeout-ms 10000
|
||||
:interval-ms 200}))
|
||||
(testing "Multiple `limit` query args should be handled correctly (#45345)"
|
||||
(let [total-count (-> (mt/user-real-request :crowberto :get 200 "search?q=product")
|
||||
:data count)
|
||||
|
||||
52
test/metabase/search/appdb/core_test.clj
Normal file
52
test/metabase/search/appdb/core_test.clj
Normal file
@@ -0,0 +1,52 @@
|
||||
(ns metabase.search.appdb.core-test
|
||||
(:require
|
||||
[clojure.test :refer :all]
|
||||
[metabase.app-db.cluster-lock :as cluster-lock]
|
||||
[metabase.search.appdb.core :as appdb.core]
|
||||
[metabase.search.appdb.index :as search.index]
|
||||
[metabase.search.appdb.index-state :as index-state]
|
||||
[metabase.test :as mt]))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
;;; The cluster-wide reindex lock funnels every index-build trigger through one choke point. These tests
|
||||
;;; pin the try-acquire-skip contract of the helper without needing real cross-node concurrency, by
|
||||
;;; standing in for the underlying cluster-lock acquisition.
|
||||
|
||||
(deftest reindex-lock-runs-body-when-free-test
|
||||
(testing "when the lock is free, the body runs and its value is returned"
|
||||
(mt/with-dynamic-fn-redefs [cluster-lock/do-with-cluster-lock (fn [_opts thunk] (thunk))]
|
||||
(is (= :ran (#'appdb.core/do-with-reindex-lock (constantly :ran)))))))
|
||||
|
||||
(deftest reindex-lock-skips-when-held-test
|
||||
(testing "when another node/thread holds the lock, the body is skipped and nil is returned"
|
||||
(let [ran (atom false)]
|
||||
;; do-with-cluster-lock wraps a failed acquisition in an ex-info carrying :lock-names
|
||||
(mt/with-dynamic-fn-redefs [cluster-lock/do-with-cluster-lock
|
||||
(fn [_opts _thunk]
|
||||
(throw (ex-info "Failed to obtain cluster lock: search-index"
|
||||
{:lock-names ["search-index"] :retries 0})))]
|
||||
(is (nil? (#'appdb.core/do-with-reindex-lock (fn [] (reset! ran true) :ran))))
|
||||
(is (false? @ran) "the reindex body must not run while a build is already in progress")))))
|
||||
|
||||
(deftest reindex-lock-bypassed-for-non-db-backed-store-test
|
||||
(testing "an in-memory (mock) store bypasses the cluster lock entirely — parallel per-test reindexes have
|
||||
no shared resource to serialize and must not skip each other"
|
||||
(binding [search.index/*state-store* (index-state/mock-store {:active :some-table})]
|
||||
;; if the lock were used, this stand-in would make acquisition look 'already held' and skip the body
|
||||
(mt/with-dynamic-fn-redefs [cluster-lock/do-with-cluster-lock
|
||||
(fn [_opts _thunk]
|
||||
(throw (ex-info "cluster lock must not be taken for a non-db-backed store"
|
||||
{:lock-names ["search-index"]})))]
|
||||
(is (= :ran (#'appdb.core/do-with-reindex-lock (constantly :ran))))))))
|
||||
|
||||
(deftest reindex-lock-propagates-real-errors-test
|
||||
(testing "an error from the body itself is not mistaken for a held lock — it propagates"
|
||||
(mt/with-dynamic-fn-redefs [cluster-lock/do-with-cluster-lock (fn [_opts thunk] (thunk))]
|
||||
(is (thrown-with-msg? Exception #"boom"
|
||||
(#'appdb.core/do-with-reindex-lock (fn [] (throw (ex-info "boom" {}))))))))
|
||||
(testing "an unrelated cluster-lock ExceptionInfo (no :lock-names) also propagates"
|
||||
(mt/with-dynamic-fn-redefs [cluster-lock/do-with-cluster-lock
|
||||
(fn [_opts _thunk] (throw (ex-info "unexpected" {:something :else})))]
|
||||
(is (thrown-with-msg? Exception #"unexpected"
|
||||
(#'appdb.core/do-with-reindex-lock (constantly :ran)))))))
|
||||
74
test/metabase/search/appdb/document_test.clj
Normal file
74
test/metabase/search/appdb/document_test.clj
Normal file
@@ -0,0 +1,74 @@
|
||||
(ns metabase.search.appdb.document-test
|
||||
(:require
|
||||
[clojure.test :refer :all]
|
||||
[metabase.app-db.core :as mdb]
|
||||
[metabase.search.appdb.document :as document]
|
||||
[metabase.test.fixtures :as fixtures]))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
;; document->entry dispatches into specialization, which needs a configured app DB to know its db-type.
|
||||
(use-fixtures :once (fixtures/initialize :db))
|
||||
|
||||
(deftest strip-junk-chars-test
|
||||
;; Build inputs from explicit code points rather than embedding raw control bytes in the source file.
|
||||
(let [strip @#'document/strip-junk-chars
|
||||
c (fn [code] (str (char code)))]
|
||||
(testing "non-string values pass through unchanged"
|
||||
(is (= 42 (strip 42)))
|
||||
(is (= :a-kw (strip :a-kw)))
|
||||
(is (= nil (strip nil)))
|
||||
(is (= [:|| "x"] (strip [:|| "x"]))))
|
||||
(testing "NUL byte (0x00) is replaced with a space — Postgres rejects raw NUL in text columns"
|
||||
(is (= "a b" (strip (str "a" (c 0x00) "b")))))
|
||||
(testing "C0 controls (BEL, BS, VT, FF, US) are replaced with spaces"
|
||||
(is (= "a b c d e f"
|
||||
(strip (str "a" (c 0x07) "b" (c 0x08) "c" (c 0x0B) "d" (c 0x0C) "e" (c 0x1F) "f")))))
|
||||
(testing "tab/newline/CR also become spaces (already whitespace for tsvector tokenization)"
|
||||
(is (= "a b c d" (strip "a\tb\nc\rd"))))
|
||||
(testing "DEL (0x7F) and C1 controls (0x80-0x9F) are replaced"
|
||||
(is (= "a b c d" (strip (str "a" (c 0x7F) "b" (c 0x80) "c" (c 0x9F) "d")))))
|
||||
(testing "unpaired surrogate code points are replaced"
|
||||
(is (= "a b" (strip (str "a" (char 0xD800) "b")))))
|
||||
(testing "ordinary text — letters, digits, punctuation, CJK, emoji — is untouched"
|
||||
(is (= "Hello, world! 123 你好 🦄"
|
||||
(strip "Hello, world! 123 你好 🦄"))))
|
||||
(testing "BOM (U+FEFF) and zero-width joiner (U+200D — Cf class) are intentionally NOT stripped"
|
||||
(let [s (str "a" (char 0xFEFF) "b" (char 0x200D) "c")]
|
||||
(is (= s (strip s)))))
|
||||
(testing "control chars at start/end of string"
|
||||
(is (= " a " (strip (str (c 0x07) "a" (c 0x07))))))
|
||||
(testing "consecutive control chars each become a space"
|
||||
(is (= "a b" (strip (str "a" (c 0x07) (c 0x08) (c 0x0B) "b")))))))
|
||||
|
||||
(deftest document->entry-junk-chars-test
|
||||
;; document->entry calls (specialization/extra-entry-fields entity), which dispatches on (mdb/db-type)
|
||||
;; and is only implemented for :postgres and :h2 — running this on MySQL/MariaDB hits the missing-impl error.
|
||||
(when (#{:postgres :h2} (mdb/db-type))
|
||||
(let [->entry @#'document/document->entry
|
||||
FF (str (char 0x000C))
|
||||
NUL (str (char 0x0000))
|
||||
BEL (str (char 0x0007))
|
||||
DEL (str (char 0x007F))
|
||||
dirty-name (str "Title" FF "Sub" NUL "title" BEL)
|
||||
dirty-st (str "line1" DEL "line2")
|
||||
entity {:model "card"
|
||||
:name dirty-name
|
||||
:searchable_text dirty-st
|
||||
:display_data {:name dirty-name}
|
||||
:legacy_input {}
|
||||
:archived false}
|
||||
entry (->entry entity)]
|
||||
(testing "top-level :name column has control chars replaced with spaces"
|
||||
(is (= "Title Sub title " (:name entry))))
|
||||
(testing "search_vector / extra fields built from the stripped entity contain no raw control chars"
|
||||
(let [printed (pr-str (vals (select-keys entry [:search_vector :with_native_query_vector
|
||||
:search_terms :native_search_terms])))]
|
||||
(is (false? (.contains printed NUL)) "no NUL byte")
|
||||
(is (false? (.contains printed BEL)) "no BEL")
|
||||
(is (false? (.contains printed FF)) "no form feed")
|
||||
(is (false? (.contains printed DEL)) "no DEL")))
|
||||
(testing "display_data preserves original characters (value was a map at strip time, encoded after)"
|
||||
(let [^String json-str (:display_data entry)]
|
||||
(is (false? (.contains json-str ^CharSequence NUL)) "encoded JSON has no raw NUL")
|
||||
(is (false? (.contains json-str ^CharSequence FF)) "encoded JSON has no raw form feed"))))))
|
||||
112
test/metabase/search/appdb/index_state_test.clj
Normal file
112
test/metabase/search/appdb/index_state_test.clj
Normal file
@@ -0,0 +1,112 @@
|
||||
(ns metabase.search.appdb.index-state-test
|
||||
(:require
|
||||
[clojure.test :refer :all]
|
||||
[metabase.search.appdb.index-state :as index-state])
|
||||
(:import
|
||||
(java.util.concurrent CountDownLatch TimeUnit)))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
(defn- tracking-sync-fn
|
||||
"Returns {:fn f :call-count atom} where f is a sync-fn that records invocations."
|
||||
[result]
|
||||
(let [call-count (atom 0)]
|
||||
{:fn (fn [] (swap! call-count inc) result)
|
||||
:call-count call-count}))
|
||||
|
||||
;;; DbBackedStateStore
|
||||
|
||||
(deftest db-backed-initial-sync-test
|
||||
(let [{:keys [fn call-count]} (tracking-sync-fn {:active :table-a :pending nil})
|
||||
store (index-state/db-backed-store fn)]
|
||||
(testing "syncs from DB on first current-state call"
|
||||
(let [s (index-state/current-state store)]
|
||||
(is (= {:active :table-a :pending nil} s))
|
||||
(is (= 1 @call-count))))))
|
||||
|
||||
(deftest db-backed-ttl-caching-test
|
||||
(let [{:keys [fn call-count]} (tracking-sync-fn {:active :table-a :pending nil})
|
||||
store (index-state/db-backed-store fn)]
|
||||
(index-state/current-state store)
|
||||
(testing "does not call sync-fn again within the TTL window"
|
||||
(index-state/current-state store)
|
||||
(index-state/current-state store)
|
||||
(is (= 1 @call-count)))))
|
||||
|
||||
(deftest db-backed-force-refresh-test
|
||||
(let [{:keys [fn call-count]} (tracking-sync-fn {:active :table-a :pending nil})
|
||||
store (index-state/db-backed-store fn)]
|
||||
(index-state/current-state store)
|
||||
(testing "force-refresh! always calls sync-fn regardless of TTL"
|
||||
(index-state/force-refresh! store)
|
||||
(index-state/force-refresh! store)
|
||||
(is (= 3 @call-count)))))
|
||||
|
||||
(deftest db-backed-set-state-test
|
||||
(let [{:keys [fn call-count]} (tracking-sync-fn {:active :db-table :pending nil})
|
||||
store (index-state/db-backed-store fn)]
|
||||
(index-state/set-state! store {:active :new-table :pending :building})
|
||||
(testing "set-state! updates the cached state immediately without calling sync-fn"
|
||||
(is (= 0 @call-count)))
|
||||
(testing "current-state returns the value from set-state! within the TTL"
|
||||
(let [s (index-state/current-state store)]
|
||||
(is (= :new-table (:active s)))
|
||||
(is (= :building (:pending s)))
|
||||
(is (= 0 @call-count) "TTL was reset by set-state!, so sync-fn still not called")))))
|
||||
|
||||
(deftest db-backed-set-state-partial-test
|
||||
(let [{:keys [fn]} (tracking-sync-fn {:active :db-table :pending nil})
|
||||
store (index-state/db-backed-store fn)]
|
||||
(index-state/set-state! store {:active :first-table :pending :pend})
|
||||
(testing "set-state! with only :active clears :pending"
|
||||
(index-state/set-state! store {:active :second-table :pending nil})
|
||||
(is (= {:active :second-table :pending nil} (index-state/current-state store))))))
|
||||
|
||||
(deftest db-backed-concurrent-ttl-test
|
||||
(let [sync-count (atom 0)
|
||||
;; Gate so all threads enter sync-fn together before any completes
|
||||
enter-latch (CountDownLatch. 1)
|
||||
sync-fn (fn []
|
||||
(swap! sync-count inc)
|
||||
(.await ^CountDownLatch enter-latch 1 TimeUnit/SECONDS)
|
||||
{:active :table :pending nil})
|
||||
store (index-state/db-backed-store sync-fn)
|
||||
n 10
|
||||
threads (repeatedly n #(Thread. ^Runnable (fn [] (index-state/current-state store))))]
|
||||
(run! #(.start ^Thread %) threads)
|
||||
(Thread/sleep 50)
|
||||
(.countDown ^CountDownLatch enter-latch)
|
||||
(run! #(.join ^Thread % 2000) threads)
|
||||
(is (= 1 @sync-count) "sync-fn called exactly once under concurrent TTL expiry")))
|
||||
|
||||
;;; MockStateStore
|
||||
|
||||
(deftest mock-store-initial-state-test
|
||||
(let [store (index-state/mock-store {:active :my-table})]
|
||||
(is (= {:active :my-table :pending nil} (index-state/current-state store)))))
|
||||
|
||||
(deftest mock-store-set-state-test
|
||||
(let [store (index-state/mock-store {:active :my-table})]
|
||||
(index-state/set-state! store {:active :my-table :pending :building})
|
||||
(is (= {:active :my-table :pending :building} (index-state/current-state store)))))
|
||||
|
||||
(deftest mock-store-force-refresh-is-noop-test
|
||||
(let [store (index-state/mock-store {:active :my-table})]
|
||||
(index-state/set-state! store {:active :updated})
|
||||
(is (= {:active :updated :pending nil} (index-state/force-refresh! store))
|
||||
"force-refresh! on mock returns current in-memory state unchanged")))
|
||||
|
||||
(deftest mock-store-empty-test
|
||||
(let [store (index-state/mock-store)]
|
||||
(is (= {:active nil :pending nil} (index-state/current-state store)))))
|
||||
|
||||
;;; db-backed?
|
||||
|
||||
(deftest db-backed-store-is-db-backed-test
|
||||
(let [{:keys [fn]} (tracking-sync-fn {:active :t :pending nil})
|
||||
store (index-state/db-backed-store fn)]
|
||||
(is (true? (index-state/db-backed? store)))))
|
||||
|
||||
(deftest mock-store-is-not-db-backed-test
|
||||
(let [store (index-state/mock-store {:active :t})]
|
||||
(is (false? (index-state/db-backed? store)))))
|
||||
@@ -7,8 +7,12 @@
|
||||
[metabase.indexed-entities.models.model-index :as model-index]
|
||||
[metabase.lib.core :as lib]
|
||||
[metabase.lib.metadata :as lib.metadata]
|
||||
[metabase.search.appdb.core :as appdb.core]
|
||||
[metabase.search.appdb.index :as search.index]
|
||||
[metabase.search.appdb.index-state :as index-state]
|
||||
[metabase.search.appdb.specialization.api :as specialization]
|
||||
[metabase.search.appdb.table :as search.table]
|
||||
[metabase.search.appdb.writer :as search.writer]
|
||||
[metabase.search.core :as search]
|
||||
[metabase.search.engine :as search.engine]
|
||||
[metabase.search.ingestion :as search.ingestion]
|
||||
@@ -514,22 +518,25 @@
|
||||
(try
|
||||
(doseq [tn (cons related-table obsolete-tables)]
|
||||
(try
|
||||
(search.index/create-table! tn)
|
||||
(search.table/create-table! tn)
|
||||
;; They might already exist
|
||||
(catch Exception _)))
|
||||
(testing "Given various obsolete search indexes"
|
||||
(is (every? #'search.index/exists? (cons related-table obsolete-tables))))
|
||||
(is (every? #'search.table/exists? (cons related-table obsolete-tables))))
|
||||
;; Clear tracking so reset-index! activates a fresh table (nothing serving) rather than building a
|
||||
;; pending replacement; this test asserts the activate-immediately path.
|
||||
(search.engine/reset-tracking! :search.engine/appdb)
|
||||
(search.index/reset-index!)
|
||||
(testing "We can create new index"
|
||||
(is (#'search.index/exists? (search.index/active-table))))
|
||||
(is (#'search.table/exists? (search.index/active-table))))
|
||||
(testing "... without destroying any related non-index tables"
|
||||
(is (#'search.index/exists? related-table)))
|
||||
(is (#'search.table/exists? related-table)))
|
||||
(testing "... and we clear out all the obsolete tables"
|
||||
(is (every? (comp not #'search.index/exists?) obsolete-tables)))
|
||||
(is (every? (comp not #'search.table/exists?) obsolete-tables)))
|
||||
(testing "... and there is no more pending table"
|
||||
(is (not (#'search.index/exists? (#'search.index/pending-table)))))
|
||||
(is (not (#'search.table/exists? (#'search.index/pending-table)))))
|
||||
(finally
|
||||
(#'search.index/drop-table! related-table))))))
|
||||
(#'search.table/drop-table! related-table))))))
|
||||
|
||||
;; We don't currently track database deletes in realtime in the search index.
|
||||
;; To do so, we would make use of the following mechanism to find downstream elements deleted by a cascade.
|
||||
@@ -559,126 +566,141 @@
|
||||
[p ds])))
|
||||
table->descendants))))))
|
||||
|
||||
(defn- active-table-after [simulated-delay-ns]
|
||||
(mt/with-dynamic-fn-redefs [search.index/now (constantly (+ simulated-delay-ns (System/nanoTime)))]
|
||||
(search.index/active-table)))
|
||||
(deftest ^:synchronized reset-index!-keeps-serving-index-test
|
||||
;; reset-index! must never blank out search: when a complete index is already serving, it builds the
|
||||
;; replacement as :pending and leaves the old :active table answering queries until the caller populates
|
||||
;; and rotates. It only activates a fresh empty table immediately when nothing is serving.
|
||||
(search.tu/with-temp-real-index "reset-signal-test"
|
||||
(testing ":activated when nothing is serving — a fresh empty table becomes active immediately"
|
||||
(is (= :activated (search.index/reset-index!)))
|
||||
(is (search.table/exists? (search.index/active-table)) "fresh table is active and exists")
|
||||
(is (nil? (search.index/pending-table)) "no pending left behind"))
|
||||
(testing ":pending when a complete index is serving — old active kept, replacement is pending"
|
||||
(let [serving (search.index/active-table)]
|
||||
(is (= :pending (search.index/reset-index!)))
|
||||
(is (= serving (search.index/active-table)) "old active still serving queries")
|
||||
(is (some? (search.index/pending-table)) "replacement created as pending")
|
||||
(is (not= serving (search.index/pending-table)) "pending is a different, empty table")))))
|
||||
|
||||
(deftest auto-refresh-test
|
||||
(when (search/supports-index?)
|
||||
(binding [search.spec/*testing-only-index-version-hash* "auto-refresh-test"]
|
||||
(try
|
||||
(reset! @#'search.index/next-sync-at nil)
|
||||
(search.index/reset-index!)
|
||||
(let [active-before (search.index/active-table)
|
||||
active-after (search.index/gen-table-name)
|
||||
pending-after (search.index/gen-table-name)
|
||||
period @#'search.index/sync-tracking-period
|
||||
version (search.spec/index-version-hash)]
|
||||
(search-index-metadata/create-pending! :appdb version active-after)
|
||||
(search.index/create-table! active-after)
|
||||
(search-index-metadata/active-pending! :appdb version)
|
||||
(search-index-metadata/create-pending! :appdb version pending-after)
|
||||
(search.index/create-table! pending-after)
|
||||
(testing "We continue using our cached references for some time"
|
||||
(is (= active-before (active-table-after 100)))
|
||||
(is (= active-before (active-table-after (/ period 2)))))
|
||||
(testing "But eventually we refresh"
|
||||
(is (= active-after (active-table-after period)))))
|
||||
(finally
|
||||
(t2/delete! :model/SearchIndexMetadata :version "auto-refresh-test")
|
||||
(#'search.index/delete-obsolete-tables!))))))
|
||||
(deftest ^:synchronized pending-rebuild-keeps-serving-old-index-test
|
||||
;; The headline :pending win, end to end: when a complete index is already serving, a forced rebuild creates
|
||||
;; the replacement as a SEPARATE pending table and keeps the old active table answering queries until the
|
||||
;; build finishes, then rotates atomically — search is never blanked mid-rebuild. (writer-test covers the
|
||||
;; per-batch routing; reset-index!-keeps-serving-index-test covers reset-index!'s :pending signal; this
|
||||
;; covers the full init! populate-then-rotate flow with live queries.)
|
||||
(search.tu/with-temp-real-index "serving-continuity-test"
|
||||
(binding [search.ingestion/*force-sync* true]
|
||||
(mt/with-temp [:model/User {} (when-not (t2/exists? :model/User 1) {:id 1})
|
||||
:model/Collection {col-id :id} {:name "Collection"}
|
||||
:model/Card {} {:name "Serving Continuity Probe" :collection_id col-id}]
|
||||
;; Build the initial complete index (t1) containing the probe card.
|
||||
(search.engine/reindex! :search.engine/appdb {})
|
||||
(let [t1 (search.index/active-table)
|
||||
observed (atom nil)
|
||||
real-populate (deref #'appdb.core/populate-index!)]
|
||||
(is (pos? (count (search.index/search "Serving Continuity Probe")))
|
||||
"the original index serves the probe card")
|
||||
;; Force a rebuild. The pending replacement is created by reset-index! before populate runs, so we
|
||||
;; snapshot the world at the start of the populate step (mid-rebuild, pre-rotation), then let the
|
||||
;; real populate fill the pending table so activation has something to rotate in.
|
||||
(with-redefs [appdb.core/populate-index!
|
||||
(fn [mode]
|
||||
(reset! observed {:active (search.index/active-table)
|
||||
:pending (search.index/pending-table)
|
||||
:hits (count (search.index/search "Serving Continuity Probe"))})
|
||||
(real-populate mode))]
|
||||
(search.engine/init! :search.engine/appdb {:force-reset? true}))
|
||||
(testing "mid-rebuild the old index is still active and answering queries"
|
||||
(is (= t1 (:active @observed)) "old active table is unchanged during the build")
|
||||
(is (some? (:pending @observed)) "the replacement is built as a separate pending table")
|
||||
(is (not= t1 (:pending @observed)) "pending is a different table from the serving active one")
|
||||
(is (pos? (:hits @observed)) "search still returns results from the old index mid-rebuild"))
|
||||
(testing "after the build the replacement is rotated in atomically"
|
||||
(is (not= t1 (search.index/active-table)) "a new table is now active")
|
||||
(is (= (:pending @observed) (search.index/active-table)) "the pending we built became active")
|
||||
(is (pos? (count (search.index/search "Serving Continuity Probe")))
|
||||
"the rotated-in index serves the probe card")))))))
|
||||
|
||||
(deftest pending-table-expiry-test
|
||||
(when (search/supports-index?)
|
||||
(binding [search.spec/*testing-only-index-version-hash* "pending-timeout-test"]
|
||||
(try
|
||||
(reset! @#'search.index/next-sync-at nil)
|
||||
(search.index/reset-index!)
|
||||
(let [active-table (search.index/active-table)
|
||||
pending-old (search.index/gen-table-name)
|
||||
pending-new (search.index/gen-table-name)
|
||||
version (search.spec/index-version-hash)]
|
||||
;; Set up old pending table (more than a day old)
|
||||
(search.index/create-table! pending-old)
|
||||
(search-index-metadata/create-pending! :appdb version pending-old)
|
||||
(t2/update! :model/SearchIndexMetadata
|
||||
{:index_name (name pending-old)}
|
||||
{:created_at (t/minus (t/offset-date-time) (t/days 2))})
|
||||
(#'search.index/sync-tracking-atoms!)
|
||||
(testing "Active table is returned"
|
||||
(is (= active-table (search.index/active-table))))
|
||||
(testing "Old pending table is ignored (more than a day old)"
|
||||
(is (nil? (#'search.index/pending-table))))
|
||||
;; Create new pending table (less than a day old)
|
||||
(search.index/create-table! pending-new)
|
||||
(search-index-metadata/create-pending! :appdb version pending-new)
|
||||
(#'search.index/sync-tracking-atoms!)
|
||||
(testing "New pending table is included (less than a day old)"
|
||||
(is (= active-table (search.index/active-table)))
|
||||
(is (= pending-new (#'search.index/pending-table)))))
|
||||
(finally
|
||||
(t2/delete! :model/SearchIndexMetadata :version "pending-timeout-test")
|
||||
(#'search.index/delete-obsolete-tables!))))))
|
||||
(deftest ^:synchronized init!-forced-reindex-reenters-cluster-lock-test
|
||||
;; init! on a >3-day-old index calls reindex! while ALREADY holding the cluster reindex-lock, so correctness
|
||||
;; depends on the lock being reentrant (H2: ReentrantReadWriteLock; Postgres: the nested t2/with-transaction
|
||||
;; reuses the connection, re-acquiring the SELECT ... FOR UPDATE row lock within one transaction). Unlike the
|
||||
;; stubbed contract tests in core-test, this drives the REAL lock end to end: if reentrancy regressed, the
|
||||
;; inner reindex! would block, time out, and get swallowed as "already running elsewhere", so the forced
|
||||
;; reindex would silently no-op and the active table would NOT change.
|
||||
(search.tu/with-temp-real-index "reindex-reentrancy-test"
|
||||
(binding [search.ingestion/*force-sync* true]
|
||||
(is (true? (index-state/db-backed? search.index/*state-store*))
|
||||
"sanity: running against the real db-backed store, so the actual cluster lock is taken")
|
||||
(let [old-table (search.table/gen-table-name)
|
||||
version (search.spec/index-version-hash)]
|
||||
(search.table/create-table! old-table)
|
||||
(search-index-metadata/create-pending! :appdb version old-table)
|
||||
(search-index-metadata/active-pending! :appdb version)
|
||||
;; backdate so init! treats the existing index as stale (>3 days) and forces a reindex
|
||||
(t2/update! :model/SearchIndexMetadata :version version
|
||||
{:created_at (t/minus (t/offset-date-time) (t/days 4))})
|
||||
(search.index/sync-from-restored-db!)
|
||||
(is (= old-table (search.index/active-table)) "the stale table is serving before init!")
|
||||
(let [reindex-ran? (atom false)]
|
||||
;; Skip the heavy document populate; prepare!/finish! still build + rotate a fresh (empty) table,
|
||||
;; which is all we need to prove the inner reindex! body ran under the re-entered lock.
|
||||
(with-redefs [appdb.core/populate-index! (fn [_mode] (reset! reindex-ran? true) {})]
|
||||
(search.engine/init! :search.engine/appdb {:force-reset? false}))
|
||||
(is (true? @reindex-ran?)
|
||||
"the forced early reindex body executed — the inner reindex! re-acquired the held lock")
|
||||
(is (not= old-table (search.index/active-table))
|
||||
"a fresh index table was rotated in (init! did not silently skip the reindex)"))))))
|
||||
|
||||
(deftest strip-junk-chars-test
|
||||
(let [strip @#'search.index/strip-junk-chars]
|
||||
(testing "non-string values pass through unchanged"
|
||||
(is (= 42 (strip 42)))
|
||||
(is (= :a-kw (strip :a-kw)))
|
||||
(is (= nil (strip nil)))
|
||||
(is (= [:|| "x"] (strip [:|| "x"]))))
|
||||
(testing "NUL byte (0x00) is replaced with a space — Postgres rejects raw NUL in text columns"
|
||||
(is (= "a b" (strip "a | ||||