Compare commits

...

15 Commits

Author SHA1 Message Date
Nathan Voxland
0b5c4d664f Fix appdb search index test-isolation leaks and a reindex cluster-lock deadlock 2026-06-25 01:14:01 -05:00
Nathan Voxland
7e953fce2b Merge origin/search-code-cleanup into search-code-cleanup
Resolve conflicts in the appdb search files by keeping the local refinements (write-plan, assert-db-backed! guard, reentrancy + serving-continuity integration tests, index-docs! capture-once docs). The remote side only carried older versions of the same work via its origin/master merges; its sole net-new change is the SearchBar styled-component tweak, taken as-is.
2026-06-23 12:52:57 -05:00
Nathan Voxland
94daeccf1c Cleanup fixes 2026-06-23 12:09:36 -05:00
Nathan Voxland
cf73eed90b Fixed unneeded formatting changes 2026-06-23 10:06:33 -05:00
Nathan Voxland
fb632d12fd Merge remote-tracking branch 'origin/master' into search-code-cleanup 2026-06-23 10:01:07 -05:00
Nathan Voxland
477d839fb3 Merge remote-tracking branch 'origin/master' into search-code-cleanup
# Conflicts:
#	src/metabase/search/appdb/core.clj
#	src/metabase/search/appdb/index.clj
#	test/metabase/search/appdb/index_test.clj
2026-06-23 09:52:55 -05:00
Nathan Voxland
f1f97ef705 Merge remote-tracking branch 'origin/master' into search-code-cleanup 2026-06-22 16:06:18 -05:00
Nathan Voxland
2591a195ec merged from origin/master 2026-06-22 10:46:11 -05:00
Nathan Voxland
a1a441fb2c Merge origin/master into search-code-cleanup
Resolve conflicts in the appdb search index refactor: keep the namespace
split (index-state store / force-refresh model) while incorporating
master's analyze-on-activate and inline index-size gauge removal.

- Port analyze-on-activate into the new activate-table!
- Drop the inline appdb-index-size push gauge from writer.clj (now handled
  by the metrics.clj pull-collector estimate)
- Repoint metrics.clj from search.index/exists? to table/exists? after the
  exists? move to table.clj

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-19 12:42:44 -05:00
Nathan Voxland
48c874ff98 Encapsulate search ingestion queue; fix multiple-limits-test flake
Extract the asynchronous search-index ingestion queue into an
IndexingQueue component (metabase.search.index-queue) with a protocol and
sync/async implementations, mirroring the index-state store pattern. The
async impl owns the DelayQueue, listener lifecycle, and an in-flight
AtomicLong so idleness is detected exactly (queued + in-flight), removing
the old .size-polling race and settle heuristic. ingestion.clj now
delegates to the component and exposes pending-count/clear-queue!,
replacing the two private-var reach-ins in appdb/core.clj and
testing_api/api.clj.

Also fix the intermittent multiple-limits-test failure: it hits the real
(non-temp) index via user-real-request and triggers an async Quartz
reindex, then asserted immediately and raced the rebuild into a transient
empty / "Search Index not found" state. Poll until the index is queryable
and populated before asserting.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-12 14:33:38 -05:00
Nathan Voxland
339033aa5a Fix test failures 2026-06-09 15:22:47 -05:00
Nathan Voxland
423486b8b6 The cluster lock now covers every reindex trigger, not just the scheduled job
reset-index! no longer blanks out search while rebuilding.

reset-index! no longer blanks out search while rebuilding.
2026-06-09 13:43:26 -05:00
Nathan Voxland
7cefb72acd Split appdb search index into more focused namespaces.
No behavior changes, purely refactoring.
2026-06-09 09:51:34 -05:00
Nathan Voxland
e928fdbb0b Fix clj-kondo warnings after search index state machine refactor
- Add docstring to *state-store* defonce via ^{:doc ...} metadata
- Replace removed sync-tracking-atoms! private var call with index-state/force-refresh!
- Replace removed *indexes* atom reset with index-state/set-state!

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-08 15:57:06 -05:00
Nathan Voxland
436bb09ff9 Refactor appdb search index: extract state machine, fix multi-node race, decompose upsert logic 2026-06-08 13:59:17 -05:00
30 changed files with 1856 additions and 730 deletions

View File

@@ -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"

View File

@@ -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)

View 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)))))

View File

@@ -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#)))))))

View 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)}))))

View File

@@ -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

View 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)))

View 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 {}))))

View 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))

View File

@@ -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)))

View File

@@ -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)]]]}))))

View File

@@ -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.

View File

@@ -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)

View File

@@ -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

View File

@@ -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)

View 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)))))))

View 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"))))))

View 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)))))

View File

@@ -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 "ab"))))
(testing "C0 controls (BEL, BS, VT, FF, US) are replaced with spaces"
(is (= "a b c d e f"
(strip "abc d ef"))))
(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 "ab€cŸ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"
(is (= "ab‍c" (strip "ab‍c"))))
(testing "control chars at start/end of string"
(is (= " a " (strip "a"))))
(testing "consecutive control chars each become a space"
(is (= "a b" (strip "ab"))))))
(deftest ^:synchronized auto-refresh-test
;; TTL caching duration is tested in index-state-test. This integration test verifies
;; that the appdb index uses the cached state and that force-refresh! reads new DB state.
(search.tu/with-temp-real-index "auto-refresh-test"
;; Start from a clean slate so reset-index! activates a fresh table (nothing is serving yet).
(search.index/reset-index!)
(let [active-before (search.index/active-table)
active-after (search.table/gen-table-name)
pending-after (search.table/gen-table-name)
version (search.spec/index-version-hash)]
(search-index-metadata/create-pending! :appdb version active-after)
(search.table/create-table! active-after)
(search-index-metadata/active-pending! :appdb version)
(search-index-metadata/create-pending! :appdb version pending-after)
(search.table/create-table! pending-after)
(testing "We continue using our cached references without a force-refresh"
(is (= active-before (search.index/active-table))))
(testing "But we see the new active table after force-refresh!"
(index-state/force-refresh! search.index/*state-store*)
(is (= active-after (search.index/active-table)))))))
(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 @#'search.index/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"))))))
(deftest ^:synchronized pending-table-expiry-test
(search.tu/with-temp-real-index "pending-timeout-test"
;; Start clean so reset-index! activates a fresh table rather than leaving a pending replacement.
(search.index/reset-index!)
(let [active-table (search.index/active-table)
pending-old (search.table/gen-table-name)
pending-new (search.table/gen-table-name)
version (search.spec/index-version-hash)]
;; Set up old pending table (more than a day old)
(search.table/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))})
;; Force a re-read so we pick up the (expired) pending table
(index-state/force-refresh! search.index/*state-store*)
(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.table/create-table! pending-new)
(search-index-metadata/create-pending! :appdb version pending-new)
;; Force a re-read so we pick up the new pending table
(index-state/force-refresh! search.index/*state-store*)
(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)))))))
(deftest reindex-survives-duplicate-most-recent-revisions-test
;; Two `revision` rows with `most_recent = true` for the same card cause the spec's LEFT
@@ -707,7 +729,7 @@
(search.engine/reindex! :search.engine/appdb {:in-place? true})
(is (= 1 (t2/count (search.index/active-table) :model "card" :model_id (str card-id)))))))))
(deftest reindex-does-not-misdirect-writes-to-active-when-pending-tracking-lost-test
(deftest ^:synchronized reindex-does-not-misdirect-writes-to-active-when-pending-tracking-lost-test
;; Regression for the reindex write-misdirection race. A full reindex resolved its destination table
;; per batch from the tracking atom. If a concurrent TTL resync transiently blanked the :pending entry
;; mid-reindex, later batches fell through to the LIVE active table instead of the pending one -- so
@@ -716,16 +738,16 @@
(when (search/supports-index?)
(search.tu/with-temp-index-table
(let [active-tbl (search.index/active-table)
pending-tbl (search.index/gen-table-name)
pending-tbl (search.table/gen-table-name)
pending-lost? (atom false)
real-upsert specialization/batch-upsert!
docs (mapv (fn [i] {:model "card" :id (str i) :name (str "Reindex doc " i)
:display_data {} :legacy_input {} :archived false})
(range 1 6))]
(try
(search.index/create-table! pending-tbl)
(search.table/create-table! pending-tbl)
(is (zero? (t2/count active-tbl)) "active index starts empty")
(with-redefs [search.index/insert-batch-size 1
(with-redefs [search.writer/insert-batch-size 1
;; Real pending table until the first batch is written, then the tracking atom
;; "loses" it -- exactly what a background resync did mid-reindex in the incident.
search.index/pending-table (fn [] (when-not @pending-lost? pending-tbl))
@@ -734,15 +756,15 @@
(real-upsert t entries))]
;; *force-sync* false so we exercise the real reindex write path (not the dual-write path)
(binding [search.ingestion/*force-sync* false]
(#'search.index/index-docs! :search/reindexing docs)))
(search.writer/index-docs! (search.index/background-mode) docs)))
(testing "no batch is written to the live active table"
(is (zero? (t2/count active-tbl))))
(testing "every document lands in the pending table captured at the start of the reindex"
(is (= (count docs) (t2/count pending-tbl))))
(finally
(#'search.index/drop-table! pending-tbl)))))))
(#'search.table/drop-table! pending-tbl)))))))
(deftest reindex-without-pending-populates-active-table-test
(deftest ^:synchronized reindex-without-pending-populates-active-table-test
;; The initial-build path populates the freshly-activated table directly with :search/reindexing context
;; and NO pending table. That must keep writing to the active table (it is the captured destination), even
;; if the tracking atom is blanked mid-build -- it must NOT be treated as an error.
@@ -755,7 +777,7 @@
:display_data {} :legacy_input {} :archived false})
(range 1 4))]
(is (zero? (t2/count active-tbl)) "active index starts empty")
(with-redefs [search.index/insert-batch-size 1
(with-redefs [search.writer/insert-batch-size 1
;; No pending table at all; the active reference is "lost" after the first write to
;; prove the destination captured at the start is used for the whole build.
search.index/pending-table (constantly nil)
@@ -764,27 +786,43 @@
(reset! atom-blank? true)
(real-upsert t entries))]
(binding [search.ingestion/*force-sync* false]
(#'search.index/index-docs! :search/reindexing docs)))
(search.writer/index-docs! (search.index/background-mode) docs)))
(testing "every document lands in the active table that was captured at the start of the build"
(is (= (count docs) (t2/count active-tbl))))))))
(deftest when-index-created
(search.tu/with-temp-real-index "index-age-test"
(let [table-name (search.table/gen-table-name)
version (search.spec/index-version-hash)]
(testing "Nil age if no active table"
(is (nil? (#'search.index/when-index-created))))
(testing "Returns age of active table"
(let [update-time (t/truncate-to (t/minus (t/offset-date-time) (t/days 2)) :millis)]
(search.table/create-table! table-name)
(search-index-metadata/create-pending! :appdb version table-name)
(search-index-metadata/active-pending! :appdb version)
(t2/update! :model/SearchIndexMetadata
:index_name (name table-name)
{:created_at update-time})
(is (= update-time (t/truncate-to (#'search.index/when-index-created) :millis))))))))
(deftest ^:synchronized with-temp-resyncs-search-state-after-rollback-test
;; Regression: a top-level with-temp body that mutates the in-memory appdb search state store -- e.g. a
;; synchronous reindex builds + tracks a fresh table inside the (rolled-back) with-temp transaction -- must not
;; leave the GLOBAL store pointing at that now-vanished table. metabase.test.redefs re-syncs the store after the
;; transaction; without that, later tests' ingestion upserts into the missing relation, aborting and poisoning
;; the pooled connection ("current transaction is aborted") and cascading into unrelated tests.
(when (search/supports-index?)
(binding [search.spec/*testing-only-index-version-hash* "index-age-test"]
(try
(let [table-name (search.index/gen-table-name)
version (search.spec/index-version-hash)]
(testing "Nil age if no active table"
(is (nil? (#'search.index/when-index-created))))
(testing "Returns age of active table"
(let [update-time (t/truncate-to (t/minus (t/offset-date-time) (t/days 2)) :millis)]
(search.index/create-table! table-name)
(search-index-metadata/create-pending! :appdb version table-name)
(search-index-metadata/active-pending! :appdb version)
(t2/update! :model/SearchIndexMetadata
:index_name (name table-name)
{:created_at update-time})
(is (= update-time (t/truncate-to (#'search.index/when-index-created) :millis))))))
(finally
(t2/delete! :model/SearchIndexMetadata :version "index-age-test")
(#'search.index/delete-obsolete-tables!))))))
(index-state/force-refresh! search.index/*state-store*)
(let [committed (search.index/active-table)]
(mt/with-temp [:model/Card _ {:name "resync-probe"}]
;; mimic ensure-ready!/reset-index! creating + tracking a table inside the transaction
(let [t (search.table/gen-table-name)]
(search.table/create-table! t)
(index-state/set-state! search.index/*state-store* {:active t :pending nil})))
;; the with-temp transaction has rolled back: the in-txn table is gone
(let [after (search.index/active-table)]
(is (= committed after)
"the state store is re-synced to the committed active table, not the rolled-back one")
(is (or (nil? after) (search.table/exists? after))
"the state store never points at a non-existent table")))))

View File

@@ -3,6 +3,7 @@
[clojure.test :refer :all]
[metabase.app-db.core :as mdb]
[metabase.search.appdb.index :as search.index]
[metabase.search.appdb.index-state :as index-state]
[metabase.search.appdb.metrics :as metrics]
[metabase.search.engine :as search.engine]
[metabase.search.ingestion :as search.ingestion]
@@ -15,7 +16,11 @@
(use-fixtures :once (fixtures/initialize :db :test-users))
(deftest active-index-size-test
;; ^:synchronized: the rotation block drives a real reindex (taking the global cluster reindex-lock and sweeping
;; obsolete physical tables), so it stays serialized with other tests that touch the shared search index. The
;; rotation itself runs against an *isolated* state store (with-temp-real-index), so it never mutates the global
;; tracking state.
(deftest ^:synchronized active-index-size-test
(testing "estimated row count of the active index, for the metabase_search_appdb_index_size metric -- see #75064"
(search.tu/with-temp-index-table
(binding [search.ingestion/*force-sync* true]
@@ -34,18 +39,21 @@
(testing "returns an exact count on H2"
(is (= exact (#'metrics/active-index-size)))))))))))
(testing "is accurate immediately after a fresh index table is rotated in (no waiting for autoanalyze) -- see #75064"
(search.tu/with-temp-index-table
;; A full (background) reindex builds a fresh pending table, ANALYZEs it, then rotates it in. Rotation only
;; happens against a real (db-backed) state store, so we can't use with-temp-index-table (mock store).
;; with-temp-real-index gives a real but *isolated* store, so the rotation never touches the shared global
;; state (which would leak our throwaway table into other parallel tests -- see its docstring).
(search.tu/with-temp-real-index "metrics-rotation-test"
(binding [search.ingestion/*force-sync* true]
(mt/dataset test-data
(mt/with-temp [:model/User {} (when-not (t2/exists? :model/User 1) {:id 1})
:model/Collection {col-id :id} {:name "Collection"}
:model/Card {} {:name "Quarterly Forecast" :collection_id col-id}]
;; a full (not in-place) reindex populates a pending table, then activates it
(search.engine/reindex! :search.engine/appdb {})
(let [exact (t2/count (search.index/active-table))]
(is (pos? exact))
(is (= exact (#'metrics/active-index-size)))))))))
(testing "returns nil when this instance has no active index to serve from"
(search.tu/with-temp-index-table
(reset! @#'search.index/*indexes* {:active nil :pending nil})
(index-state/set-state! search.index/*state-store* {:active nil :pending nil})
(is (nil? (#'metrics/active-index-size))))))

View File

@@ -5,6 +5,7 @@
[clojure.string :as str]
[clojure.test :refer :all]
[metabase.app-db.core :as mdb]
[metabase.search.appdb.document :as search.document]
[metabase.search.appdb.index :as search.index]
[metabase.search.appdb.scoring :as scoring]
[metabase.search.appdb.specialization.api :as specialization]
@@ -30,7 +31,7 @@
[entities & body]
`(search.tu/with-temp-index-table
(#'specialization/batch-upsert! (search.index/active-table)
(map (comp #'search.index/document->entry
(map (comp #'search.document/document->entry
#'search.ingestion/->document)
~entities))
~@body))

View File

@@ -0,0 +1,55 @@
(ns metabase.search.appdb.table-test
(:require
[clojure.string :as str]
[clojure.test :refer :all]
[metabase.app-db.core :as mdb]
[metabase.search.appdb.table :as search.table]
[metabase.test.fixtures :as fixtures]
[toucan2.core :as t2])
(:import
(org.h2.jdbc JdbcSQLSyntaxErrorException)
(org.postgresql.util PSQLException PSQLState)))
(set! *warn-on-reflection* true)
(use-fixtures :once (fixtures/initialize :db))
(deftest gen-table-name-test
(testing "generates a unique, prefixed, dash-free table name"
(let [a (search.table/gen-table-name)
b (search.table/gen-table-name)]
(is (str/starts-with? (name a) "search_index__"))
(is (not= a b) "two calls produce distinct names")
(is (not (str/includes? (name a) "-")) "nano-id dashes are replaced with underscores")))
(testing "an optional suffix is appended (used to mark temp tables)"
(is (str/ends-with? (name (search.table/gen-table-name "_temp")) "_temp"))))
(deftest table-not-found-exception?-test
(testing "true when the cause is a driver missing-table error"
(is (true? (search.table/table-not-found-exception?
(ex-info "wrapped" {} (PSQLException. "relation does not exist" PSQLState/UNDEFINED_TABLE)))))
(is (true? (search.table/table-not-found-exception?
(ex-info "wrapped" {} (JdbcSQLSyntaxErrorException. "Table not found" "" "" 42102 nil ""))))))
(testing "false for unrelated errors"
(is (false? (search.table/table-not-found-exception? (ex-info "no cause" {}))))
(is (false? (search.table/table-not-found-exception? (ex-info "wrapped" {} (RuntimeException. "boom")))))))
(deftest create-exists-drop-roundtrip-test
;; create-table! dispatches into the appdb specialization, which is only implemented for the DBs that
;; support the appdb search index (postgres + h2); on mysql/mariadb there is no method.
(when (#{:postgres :h2} (mdb/db-type))
(testing "a created table is detected by exists? and gone after drop-table!"
(let [tbl (search.table/gen-table-name "_temp")]
(try
(is (false? (boolean (search.table/exists? tbl))) "absent before creation")
(search.table/create-table! tbl)
(is (true? (boolean (search.table/exists? tbl))) "present after creation")
;; the table is real and writable: it has the base columns
(is (= 0 (t2/count tbl)))
(finally
(#'search.table/drop-table! tbl)))
(is (false? (boolean (search.table/exists? tbl))) "absent after drop")))))
(deftest exists?-nil-test
(testing "exists? on nil is falsey without hitting the DB"
(is (not (search.table/exists? nil)))))

View File

@@ -0,0 +1,97 @@
(ns metabase.search.appdb.writer-test
(:require
[clojure.test :refer :all]
[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.ingestion :as search.ingestion]
[metabase.test.fixtures :as fixtures])
(:import
(org.postgresql.util PSQLException PSQLState)))
(set! *warn-on-reflection* true)
;; classify-upsert-error's :stale-table branch calls table/exists?, which needs a configured app DB.
(use-fixtures :once (fixtures/initialize :db))
;;; ---------------------------------- Reindex-mode write routing -----------------------------------
(deftest reindex-mode-write-routing-test
(let [write-plan #'search.writer/write-plan
;; Mirror how index-docs! captures the destination table ONCE at the start of a committing reindex.
captured (fn [mode] (when (#'search.index/commits-per-batch? mode)
(or (search.index/pending-table) (search.index/active-table))))
;; write-plan owns the whole per-batch decision: the [label table-fn] targets AND the commit flag.
plan (fn [mode] (write-plan mode (captured mode)))
labels (fn [mode] (map first (:targets (plan mode))))]
(testing "a committing background reindex pins ALL writes to the single table captured at the start"
;; the active table keeps serving queries and is swapped in by activate-table! when the build finishes;
;; pinning prevents a concurrent resync from redirecting writes mid-rebuild
(binding [search.index/*state-store* (index-state/mock-store {:active :active-t :pending :pending-t})
search.ingestion/*force-sync* false]
(is (= [:pending] (labels (search.index/background-mode))))
(is (true? (:commit? (plan (search.index/background-mode)))))))
(testing "an initial background build with no pending pins to the captured active table"
;; the captured destination is the active table (no pending yet); writes are still pinned to it
(binding [search.index/*state-store* (index-state/mock-store {:active :active-t :pending nil})
search.ingestion/*force-sync* false]
(is (= [:pending] (labels (search.index/background-mode))))
(is (true? (:commit? (plan (search.index/background-mode)))))))
(testing "*force-sync* disables per-batch commits and keeps active current via dual-write"
;; synchronous runs happen inside the caller's transaction, so we must not open a side connection
(binding [search.index/*state-store* (index-state/mock-store {:active :active-t :pending :pending-t})
search.ingestion/*force-sync* true]
(is (false? (:commit? (plan (search.index/background-mode)))))
(is (= [:active :pending] (labels (search.index/background-mode))))))
(testing "in-place and incremental modes always dual-write without committing"
;; dual-write keeps an in-flight background rebuild current with live edits
(binding [search.index/*state-store* (index-state/mock-store {:active :active-t :pending :pending-t})
search.ingestion/*force-sync* false]
(doseq [mode [(search.index/in-place-mode) (search.index/incremental-mode)]]
(is (false? (:commit? (plan mode))))
(is (= [:active :pending] (labels mode))))))))
;;; ---------------------------------- Batch upsert error recovery ----------------------------------
(defn- missing-table-ex
"An exception shaped like the one a driver throws when upserting into a dropped table."
[]
(ex-info "boom" {} (PSQLException. "relation does not exist" PSQLState/UNDEFINED_TABLE)))
(deftest classify-upsert-error-test
(let [classify #'search.writer/classify-upsert-error]
(testing "interruption is recognised so the thread can be re-interrupted"
(is (= :interrupted (classify (InterruptedException.) (search.table/gen-table-name)))))
(testing "an error with no missing-table cause is :unknown-error (short-circuits before touching the DB)"
(is (= :unknown-error (classify (ex-info "plain" {}) (search.table/gen-table-name)))))
(testing "a missing-table error against a table that really is gone is :stale-table"
;; gen-table-name is never created, so exists? is false
(is (= :stale-table (classify (missing-table-ex) (search.table/gen-table-name)))))))
(deftest safe-batch-upsert-recovers-from-stale-table-test
(testing "a stale tracked table triggers one state refresh and the batch is retried against the new table"
(let [missing (search.table/gen-table-name)
good (search.table/gen-table-name)
writes (atom [])
;; First read of state points at the (missing) table; force-refresh! then reveals the new one.
states (atom [{:active missing :pending nil} {:active good :pending nil}])
sync-fn (fn [] (let [[head & tail] @states]
(when tail (reset! states (vec tail)))
head))]
(with-redefs [specialization/batch-upsert!
(fn [table entries]
(if (= (keyword table) missing)
(throw (missing-table-ex))
(swap! writes conj [(keyword table) (count entries)])))]
(binding [search.index/*state-store* (index-state/db-backed-store sync-fn)]
(is (= good (#'search.writer/safe-batch-upsert! :active search.index/active-table [{:x 1}])))
(is (= [[good 1]] @writes) "the batch landed in the refreshed table exactly once"))))))
(deftest safe-batch-upsert-skips-unknown-errors-test
(testing "a non-stale upsert error is logged and skipped (returns nil) so the rest of the reindex continues"
(let [t (search.table/gen-table-name)]
(with-redefs [specialization/batch-upsert! (fn [_ _] (throw (ex-info "kaboom" {})))]
(binding [search.index/*state-store* (index-state/mock-store {:active t})]
(is (nil? (#'search.writer/safe-batch-upsert! :active search.index/active-table [{:x 1}]))))))))

View File

@@ -9,6 +9,7 @@
[metabase.config.core :as config]
[metabase.queries-rest.api.card :as api.card]
[metabase.search.appdb.index :as search.index]
[metabase.search.appdb.index-state :as index-state]
[metabase.search.config :as search.config]
[metabase.search.core :as search]
[metabase.search.impl :as search.impl]
@@ -284,7 +285,7 @@
(deftest old-values-removed-from-index
(when (search/supports-index?)
(#'search.index/sync-tracking-atoms!)
(index-state/force-refresh! search.index/*state-store*)
(let [search-term (str (random-uuid))]
(binding [search.ingestion/*force-sync* true]
(mt/with-temp

View File

@@ -0,0 +1,92 @@
(ns metabase.search.index-queue-test
(:require
[clojure.test :refer :all]
[metabase.search.index-queue :as iq]))
(defn- recording-handler
"Returns [handler results-atom] where handler appends each batch to results-atom."
[]
(let [results (atom [])]
[(fn [batch] (swap! results conj (vec batch))) results]))
(defn- async-test-queue
[name handler & {:as opts}]
(iq/async-queue (merge {:delay-ms 0
:listener-name name
:handler handler
:listener-opts {:max-batch-messages 50 :max-next-ms 10}}
opts)))
(deftest async-enqueue-processes-and-becomes-idle-test
(let [[handler results] (recording-handler)
q (async-test-queue "iq-test-basic" handler)]
(try
(iq/start! q)
(iq/enqueue! q [[:card 1] [:card 2]])
(is (true? (iq/await-idle! q {:timeout-ms 2000})) "queue drains")
(is (= #{[:card 1] [:card 2]} (set (apply concat @results))))
(finally (iq/stop! q)))))
(deftest in-flight-counted-until-handler-completes-test
(testing "a dequeued-but-still-processing batch keeps the queue non-idle even though the queue itself is empty"
(let [release (promise)
started (promise)
handler (fn [_batch] (deliver started true) @release)
q (async-test-queue "iq-test-inflight" handler)]
(try
(iq/start! q)
(iq/enqueue! q [[:card 1]])
(is (true? (deref started 2000 false)) "handler picked up the batch")
;; The single message is now off the queue (queue size == 0) but the handler has
;; not returned. The queue MUST still report itself as busy.
(is (false? (iq/idle? q))
"must not be idle while a batch is in flight")
(is (pos? (iq/pending-count q)))
(deliver release :done)
(is (true? (iq/await-idle! q {:timeout-ms 2000}))
"idle once the in-flight batch completes")
(finally
(deliver release :done)
(iq/stop! q))))))
(deftest await-idle-times-out-test
(let [release (promise)
handler (fn [_batch] @release)
q (async-test-queue "iq-test-timeout" handler)]
(try
(iq/start! q)
(iq/enqueue! q [[:card 1]])
(is (false? (iq/await-idle! q {:timeout-ms 200 :poll-ms 10}))
"returns false when work never completes")
(finally
(deliver release :done)
(iq/stop! q)))))
(deftest lifecycle-test
(let [[handler _] (recording-handler)
q (async-test-queue "iq-test-lifecycle" handler)]
(is (false? (boolean (iq/running? q))) "not running before start")
(iq/start! q)
(is (true? (iq/running? q)) "running after start")
(iq/stop! q)
(is (false? (boolean (iq/running? q))) "not running after stop")))
(deftest clear-resets-pending-test
(let [[handler _] (recording-handler)
;; large delay so messages never become takeable during the test
q (async-test-queue "iq-test-clear" handler :delay-ms 100000)]
(iq/enqueue! q [[:card 1] [:card 2] [:card 3]])
(is (pos? (iq/pending-count q)) "pending after enqueue")
(iq/clear! q)
(is (zero? (iq/pending-count q)) "cleared")
(is (true? (iq/idle? q)))))
(deftest sync-queue-runs-inline-and-is-always-idle-test
(let [[handler results] (recording-handler)
q (iq/sync-queue handler)]
(is (true? (iq/idle? q)))
(iq/enqueue! q [[:card 1] [:card 2]])
(is (= [[[:card 1] [:card 2]]] @results) "handler invoked inline with the whole batch")
(is (zero? (iq/pending-count q)))
(is (true? (iq/await-idle! q {})))
(is (true? (iq/running? q)))))

View File

@@ -49,7 +49,11 @@
versions (map str (repeatedly n random-uuid))
;; to make things interesting, we're not using the latest one
our-version (nth versions 3)
index-count #(t2/count :model/SearchIndexMetadata)
;; Start from a clean slate (rolled back at the end of the transaction): the app DB may already hold a
;; real (appdb) serving index whose metadata delete-obsolete! now deliberately protects, which would
;; otherwise skew the most-recent ranking and these counts.
_ (t2/delete! :model/SearchIndexMetadata)
index-count #(t2/count :model/SearchIndexMetadata :engine engine)
initial-count (index-count)]
;; create a bunch of indexes, more than we will want to keep
(doseq [v versions]
@@ -64,12 +68,31 @@
(is (= kept (index-count))))
(testing "It keeps the latest versions"
(is (= (set (take-last 3 versions))
(t2/select-fn-set :version :model/SearchIndexMetadata))))
(t2/select-fn-set :version :model/SearchIndexMetadata :engine engine))))
(testing "After 1 day, it deletes version which are neither the latest, nor used by this instance"
(mt/with-dynamic-fn-redefs [t/zoned-date-time (constantly (t/plus (t/zoned-date-time) (t/days 1) (t/minutes 1)))]
(search-index-metadata/delete-obsolete! our-version)
(is (= #{our-version (last versions)}
(t2/select-fn-set :version :model/SearchIndexMetadata))))))))
(t2/select-fn-set :version :model/SearchIndexMetadata :engine engine))))))))
(deftest delete-obsolete!-keeps-serving-version-test
(testing "a version still serving an ACTIVE index is never pruned by the most-recent window, even when newer
(e.g. transient/test-only) versions crowd it out -- dropping its metadata would orphan its live table,
which then gets swept and breaks in-flight reads/writes"
(t2/with-transaction [_ t2.connection/*current-connectable* {:rollback-only true}]
(let [engine :something-futureproof
serving (str (random-uuid))
;; four *newer* versions, created after `serving`, so `serving` (oldest updated_at + lowest id) falls
;; outside the 3-most-recent window.
newer (mapv (fn [_] (str (random-uuid))) (range 4))]
(search-index-metadata/create-pending! engine serving serving)
(search-index-metadata/active-pending! engine serving) ; `serving` is now :active
(doseq [v newer]
(search-index-metadata/create-pending! engine v v))
;; our-version is one of the throwaway newer versions, NOT `serving`
(search-index-metadata/delete-obsolete! (first newer))
(is (contains? (t2/select-fn-set :version :model/SearchIndexMetadata :engine engine) serving)
"the serving version's metadata must survive being crowded out of the most-recent window")))))
(comment
(t2/delete! :model/SearchIndexMetadata :engine :something-futureproof))

View File

@@ -1,41 +1,53 @@
(ns metabase.search.task.search-index-test
(:require
[clojure.test :refer :all]
[honey.sql.helpers :as sql.helpers]
[metabase.search.appdb.index :as search.index]
[metabase.search.core :as search]
;; loaded for its side effect: registers ALL search engines' defmethods (appdb + semantic + in-place) so
;; the search facade's supported-engine?/active-engines work — supported-engine? :default throws rather
;; than returning false, so every candidate engine must be registered.
[metabase.search.init]
[metabase.search.task.search-index :as task]
[metabase.search.test-util :as search.tu]
[metabase.test :as mt]
[metabase.test.fixtures :as fixtures]
[toucan2.core :as t2]))
(set! *warn-on-reflection* true)
(use-fixtures :once (fixtures/initialize :db))
;; TODO this is coupled to appdb engines at the moment
(defn- index-size []
(t2/count (search.index/active-table)))
(deftest index!-test
(search.tu/with-temp-index-table
;; TODO this is coupled to appdb engines at the moment
(t2/query (sql.helpers/drop-table (search.index/active-table)))
(testing "It can recreate the index from scratch"
;; May return falsey if there is nothing to index.
(is (task/init!))
(is (pos? (index-size))))
(testing "It will reuse an existing index"
(is (not (task/init!))))))
;; These tests have init!/reindex! build, track, and rotate GENUINE tables — so they need a real (db-backed)
;; store, not with-temp-index-table's MockStateStore (whose force-refresh! is a no-op, so the rotations can't be
;; tracked). with-temp-real-index gives exactly that: a fresh, isolated db-backed store under a throwaway version,
;; so the rotations neither read nor clobber the shared production index tracking (which would leak a dropped
;; table into other parallel tests). See its docstring.
(deftest reindex!-test
(search.tu/with-temp-index-table
;; TODO this is coupled to appdb engines at the moment
(t2/query (sql.helpers/drop-table (search.index/active-table)))
(testing "It can recreate the index from scratch"
(is (search/reindex! {:async? false}))
(let [initial-size (index-size)
table-name (search.index/active-table)]
(is (pos? initial-size))
(t2/delete! table-name (t2/select-one-pk table-name))
(is (= (dec initial-size) (index-size)))
(testing "It can cycle the index gracefully"
(is (search/reindex! {:async? false}))
(is (= initial-size (index-size))))))))
(deftest ^:synchronized index!-test
;; A temp card guarantees there is at least one searchable document to index, independent of whatever the
;; test app DB happens to contain.
(mt/with-temp [:model/Card _ {:name (mt/random-name)}]
(search.tu/with-temp-real-index "task-init-test"
(testing "It can recreate the index from scratch"
(is (task/init!))
(is (pos? (index-size))))
(testing "It will reuse an existing index"
(is (not (task/init!)))))))
(deftest ^:synchronized reindex!-test
(mt/with-temp [:model/Card _ {:name (mt/random-name)}]
(search.tu/with-temp-real-index "task-reindex-test"
(testing "It can recreate the index from scratch"
(is (search/reindex! {:async? false}))
(let [initial-size (index-size)
table-name (search.index/active-table)]
(is (pos? initial-size))
(t2/delete! table-name (t2/select-one-pk table-name))
(is (= (dec initial-size) (index-size)))
(testing "It can cycle the index gracefully"
(is (search/reindex! {:async? false}))
(is (= initial-size (index-size)))))))))

View File

@@ -4,10 +4,13 @@
[metabase.permissions.util :as perms-util]
[metabase.request.core :as request] ;; For now, this is specialized to the appdb engine, but we should be able to generalize it to all engines.
[metabase.search.appdb.index :as search.index]
[metabase.search.appdb.index-state :as index-state]
[metabase.search.appdb.table :as search.table]
[metabase.search.config :as search.config]
[metabase.search.core :as search]
[metabase.search.engine :as search.engine]
[metabase.search.impl :as search.impl]
[metabase.search.spec :as search.spec]
[metabase.test :as mt]
[toucan2.core :as t2]))
@@ -30,6 +33,30 @@
(with-sync-search-indexing
~@body))))
#_{:clj-kondo/ignore [:metabase/test-helpers-use-non-thread-safe-functions]}
(defmacro with-temp-real-index
"Run BODY against a FRESH, isolated db-backed state store plus a throwaway index `version`, so real rotation /
TTL-cache behavior can be exercised end to end.
Unlike [[with-temp-index-table]] (which uses a mock store and so cannot rotate), this keeps a *real* db-backed
store — but a brand-new instance rather than the shared global `*state-store*`. That isolation matters: a
rotation caches its new `:active` table for the full TTL, so rotating the *global* store would leave a
throwaway table cached on the node, and the cleanup below then drops it — every other (parallel) test would
keep reading the now-dropped table (`search_index__… does not exist`, which aborts its transaction and
cascades). With its own store + version, the whole rebuild is invisible to the rest of the suite.
Drops the throwaway tables + metadata on the way out."
[version & body]
`(when (search/supports-index?)
(binding [search.index/*state-store* (index-state/db-backed-store
(:sync-fn search.index/*state-store*))
search.spec/*testing-only-index-version-hash* ~version]
(try
~@body
(finally
(t2/delete! :model/SearchIndexMetadata :version ~version)
(#'search.table/delete-obsolete-tables!))))))
#_{:clj-kondo/ignore [:metabase/test-helpers-use-non-thread-safe-functions]}
(defmacro with-new-search-if-available*
"Create a temporary index table for the duration of the body."

View File

@@ -13,6 +13,33 @@
"Used to detect whether we're in a nested [[with-temp]]. Default is false."
false)
(defn- do-with-appdb-search-state-resync
"Run `thunk`; if it changed the in-memory appdb search index state store, re-sync that store to committed
metadata afterward.
A top-level [[with-temp]] body runs inside a transaction that is then rolled back. If that body (re)builds or
rotates the appdb search index -- e.g. synchronous ingestion triggering `ensure-ready!`/`reset-index!` -- the
table DDL is rolled back with the transaction, but the search state store is an in-memory atom that survives
the rollback. It would then keep pointing at a table that no longer exists, so every later test's ingestion
upserts into a missing relation, aborting the statement and poisoning the pooled connection (the failure then
surfaces in some unrelated test as \"current transaction is aborted\"). Re-syncing to committed metadata drops
the stale pointer.
No-op, and no DB access, unless the search index namespace is loaded, the store is db-backed (i.e. not a
with-temp-index-table mock), and the body actually mutated the in-memory state."
[thunk]
(let [store (some-> (find-ns 'metabase.search.appdb.index) (ns-resolve '*state-store*) deref)
db-backed? (some-> (find-ns 'metabase.search.appdb.index-state) (ns-resolve 'db-backed?))
snapshot (some-> (find-ns 'metabase.search.appdb.index) (ns-resolve 'state-snapshot))
force-refresh (some-> (find-ns 'metabase.search.appdb.index-state) (ns-resolve 'force-refresh!))
track? (boolean (and store db-backed? snapshot force-refresh (db-backed? store)))
before (when track? (snapshot))]
(try
(thunk)
(finally
(when (and track? (not= before (snapshot)))
(force-refresh store))))))
(methodical/defmethod toucan2.tools.with-temp/do-with-temp* :around :default
"Initialize the DB before doing the other with-temp stuff.
Make sure metabase.test.util is loaded.
@@ -25,8 +52,10 @@
;; run `f` in a transaction if it's the top-level with-temp
(if (and tu.thread-local/*thread-local* (not *in-with-temp*))
(binding [*in-with-temp* true]
(t2.connection/with-transaction [_ t2.connection/*current-connectable* {:rollback-only true}]
(next-method model attributes f)))
(do-with-appdb-search-state-resync
(fn []
(t2.connection/with-transaction [_ t2.connection/*current-connectable* {:rollback-only true}]
(next-method model attributes f)))))
(next-method model attributes f)))
;;; wrap `with-redefs-fn` (used by `with-redefs`) so it calls `assert-test-is-not-parallel`

View File

@@ -7,6 +7,7 @@
[java-time.clock]
[metabase.app-db.core :as mdb]
[metabase.search.appdb.index :as search.index]
[metabase.search.appdb.index-state :as index-state]
[metabase.search.core :as search]
[metabase.test :as mt]
[metabase.testing-api.api :as testing]
@@ -72,7 +73,7 @@
;; Snapshot with a valid search index in place
(mt/user-http-request :rasta :post 204 (format "testing/snapshot/%s" snapshot-name))
;; Clear the tracking atoms so we can verify they get restored
(reset! @#'search.index/*indexes* {:active nil, :pending nil})
(index-state/set-state! search.index/*state-store* {:active nil :pending nil})
(is (nil? (search.index/active-table))
"Precondition: tracking atoms should be cleared before restore")
;; Restore should call sync-from-restored-db! and refresh the atoms