Give vector-only search hits the same timestamps the keyword arm returns

Both engines rebuild a result row from its JSON legacy_input, where every
timestamp has been flattened to a string, and the appdb engine parsed three of
them back into java.time values on the way out. The Lucene vector arm did not,
so a hit only it found reached callers with strings: POST /api/agent/v1/search
rejected its own response with "value must be a valid date/time/datetime". Only
vector-only hits were affected, because a fused row prefers the appdb body.

Moves that parsing into metabase.search.util as rehydrate-timestamps, exported
through metabase.search.core, and runs it in both arms over the same keys, so
the two engines cannot drift apart again.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

.clj-kondo/config.edn

@@ -486,6 +486,7 @@
   metabase-enterprise.mfa.enrollment-flow-test/with-fresh-user-session!                  clojure.core/fn
   metabase-enterprise.mfa.login-flow-test/with-enrolled-rasta!                           clojure.core/fn
   metabase-enterprise.mfa.recovery-codes-test/with-confirmed-enrollment!                 clojure.core/fn
+  metabase-enterprise.semantic-search.lucene.index/with-searcher                        clojure.core/fn
   metabase-enterprise.serialization.metadata-file-import.wire-format-test/with-tiny-fixture! clojure.core/fn
   metabase-enterprise.serialization.test-util/with-dbs                                  clojure.core/fn
   metabase-enterprise.serialization.test-util/with-random-dump-dir                      clojure.core/let

.clj-kondo/config/modules/config.edn

@@ -4191,11 +4191,13 @@
            metabot.llm
            models
            permissions
+           plugins
            premium-features
            request
            search
            search.embeddings
            settings
+           startup
            task
            tracing
            util

deps.edn

@@ -127,6 +127,7 @@
   {:mvn/version "2.26.1"}             ; allows the custom json logging format
   org.apache.logging.log4j/log4j-slf4j2-impl
   {:mvn/version "2.26.1"}             ; allows the slf4j2 API to work with log4j 2
+  org.apache.lucene/lucene-core             {:mvn/version "10.5.1"}             ; local HNSW vector index for semantic search
   org.apache.pdfbox/pdfbox                  {:mvn/version "3.0.8"}              ; PDF generation (backend dashboard PDF prototype)
   org.apache.poi/poi                        {:mvn/version "5.5.1"}              ; Work with Office documents (e.g. Excel spreadsheets) -- newer version than one specified by Docjure
   org.apache.poi/poi-ooxml                  {:mvn/version "5.5.1"

enterprise/backend/src/metabase_enterprise/semantic_search/api.clj

@@ -2,19 +2,37 @@
   "/api/ee/semantic-search endpoints"
   (:require
    [clojure.core.memoize :as memoize]
+   [metabase-enterprise.semantic-search.db :as semantic-search.db]
    [metabase-enterprise.semantic-search.db.datasource :as semantic.db.datasource]
    [metabase-enterprise.semantic-search.env :as semantic.env]
    [metabase-enterprise.semantic-search.index :as semantic.index]
    [metabase-enterprise.semantic-search.index-metadata :as semantic.index-metadata]
+   [metabase-enterprise.semantic-search.lucene.store :as lucene.store]
+   [metabase-enterprise.semantic-search.util :as semantic.util]
    [metabase.api.macros :as api.macros]
    [metabase.api.routes.common :refer [+auth]]
    [metabase.permissions.core :as perms]
-   [metabase.search.ingestion :as search.ingestion]))
+   [metabase.search.ingestion :as search.ingestion]
+   [metabase.util.log :as log]))
 
 (def ^:private indexible-items-count
   (memoize/ttl search.ingestion/search-items-count
                :ttl/threshold (* 5 60 1000))) ; 5 minutes
 
+(defn- lucene-status
+  "Indexing progress for the Lucene backend: how many documents of the current embedding space are embedded."
+  []
+  ;; Resolving the space needs the configured embedding provider to be registered, which it need not be.
+  (try
+    (let [indexed (semantic-search.db/count-embeddings (lucene.store/space-id))]
+      (if (pos? indexed)
+        {:indexed_count indexed
+         :total_est     (indexible-items-count)}
+        {}))
+    (catch Throwable t
+      (log/debugf "No semantic search status to report: %s" (ex-message t))
+      {})))
+
 (defn- active-index-document-count
   [pgvector index-metadata]
   (when-let [table-name (-> index-metadata :index :table-name)]
@@ -31,11 +49,18 @@
    :indexed_count <number of indexed items>
    :total_est     <estimated total number of items to index>
 
-  If no index is active, or no pgvector database is configured, returns an empty map."
+  If nothing is indexed yet, or the pgvector backend is selected with no pgvector database configured, returns an
+  empty map."
   []
   (perms/check-has-application-permission :setting)
-  (if-not (semantic.db.datasource/pgvector-configured?)
+  (cond
+    (semantic.util/lucene-backend?)
+    (lucene-status)
+
+    (not (semantic.db.datasource/pgvector-configured?))
     {}
+
+    :else
     (let [pgvector       (semantic.env/get-pgvector-datasource!)
           index-metadata (semantic.env/get-index-metadata)
           active-index   (when (and pgvector index-metadata)

enterprise/backend/src/metabase_enterprise/semantic_search/core.clj

@@ -8,6 +8,8 @@
    [metabase-enterprise.semantic-search.embedding :as semantic.embedding]
    [metabase-enterprise.semantic-search.env :as semantic.env]
    [metabase-enterprise.semantic-search.index-metadata :as semantic.index-metadata]
+   [metabase-enterprise.semantic-search.lucene.core :as lucene.core]
+   [metabase-enterprise.semantic-search.lucene.query :as lucene.query]
    [metabase-enterprise.semantic-search.pgvector-api :as semantic.pgvector-api]
    [metabase-enterprise.semantic-search.repair :as semantic.repair]
    [metabase-enterprise.semantic-search.settings :as semantic.settings]
@@ -61,7 +63,8 @@
   No-ops when semantic search isn't active or this process already has a build running. Database-level
   coordination in [[semantic.pgvector-api/ensure-active-hnsw-index!]] serializes builds across instances."
   []
-  (when (and (semantic.util/semantic-search-active?)
+  (when (and (not (semantic.util/lucene-backend?))
+             (semantic.util/semantic-search-active?)
              (compare-and-set! hnsw-index-build-running? false true))
     (future
       (try
@@ -97,10 +100,12 @@
   [search-ctx]
   (tracing/with-span :search "search.semantic.execute" {:search/query-length (count (:search-string search-ctx))}
     (try
-      (let [{:keys [results raw-count]}
-            (semantic.pgvector-api/query (semantic.env/get-pgvector-datasource!)
-                                         (semantic.env/get-index-metadata)
-                                         search-ctx)
+      (let [{:keys [results raw-count keyword-results]}
+            (if (semantic.util/lucene-backend?)
+              (lucene.query/query search-ctx)
+              (semantic.pgvector-api/query (semantic.env/get-pgvector-datasource!)
+                                           (semantic.env/get-index-metadata)
+                                           search-ctx))
             final-count (count results)
             threshold (semantic.settings/semantic-search-min-results-threshold)]
         (if (or (>= final-count threshold)
@@ -120,7 +125,13 @@
               (log/warn "Using an offset with semantic search will produce strange results, e.g. missing expected results, or duplicating them across pages"))
             (let [total-limit      (semantic.settings/semantic-search-results-limit)
                   fallback-results (try
-                                     (cond->> (search.engine/results (assoc search-ctx :search-engine fallback))
+                                     ;; The Lucene arm already ran the appdb query to fuse with; running it again
+                                     ;; here would double every search's keyword cost to rediscover the same rows.
+                                     ;; `some?`, not `or`: an empty vector means that arm ran and matched nothing,
+                                     ;; while nil means it never ran and this fallback is the only keyword search.
+                                     (cond->> (if (some? keyword-results)
+                                                keyword-results
+                                                (search.engine/results (assoc search-ctx :search-engine fallback)))
                                        ;; The in-place engine returns a reducible (but not seqable) result that needs to
                                        ;; be realized before we concat and dedup with the semantic engine results.
                                        (= :search.engine/in-place fallback)
@@ -146,38 +157,44 @@
   "Enterprise implementation of semantic index updating."
   :feature :semantic-search
   [document-reducible]
-  (let [pgvector       (semantic.env/get-pgvector-datasource!)
-        index-metadata (semantic.env/get-index-metadata)]
-    (if-not (index-active? pgvector index-metadata)
-      (log/debug "update-index! called prior to init!")
-      (semantic.pgvector-api/gate-updates!
-       pgvector
-       index-metadata
-       document-reducible))))
+  (if (semantic.util/lucene-backend?)
+    (lucene.core/update! document-reducible)
+    (let [pgvector       (semantic.env/get-pgvector-datasource!)
+          index-metadata (semantic.env/get-index-metadata)]
+      (if-not (index-active? pgvector index-metadata)
+        (log/debug "update-index! called prior to init!")
+        (semantic.pgvector-api/gate-updates!
+         pgvector
+         index-metadata
+         document-reducible)))))
 
 (defenterprise delete-from-index!
   "Enterprise implementation of semantic index deletion."
   :feature :semantic-search
   [model ids]
-  (let [pgvector       (semantic.env/get-pgvector-datasource!)
-        index-metadata (semantic.env/get-index-metadata)]
-    (if-not (index-active? pgvector index-metadata)
-      (log/debug "delete-from-index! called prior to init!")
-      (semantic.pgvector-api/gate-deletes!
-       pgvector
-       index-metadata
-       model
-       ids))))
+  (if (semantic.util/lucene-backend?)
+    (lucene.core/delete! model ids)
+    (let [pgvector       (semantic.env/get-pgvector-datasource!)
+          index-metadata (semantic.env/get-index-metadata)]
+      (if-not (index-active? pgvector index-metadata)
+        (log/debug "delete-from-index! called prior to init!")
+        (semantic.pgvector-api/gate-deletes!
+         pgvector
+         index-metadata
+         model
+         ids)))))
 
 (defenterprise diagnose
   "Enterprise implementation of the semantic search engine-owned diagnostic stages."
   :feature :semantic-search
   [search-ctx expected-model expected-id]
-  (let [pgvector       (semantic.env/get-pgvector-datasource!)
-        index-metadata (semantic.env/get-index-metadata)]
-    (if-not (index-active? pgvector index-metadata)
-      {:type :missing-from-index :details {:reason :no-active-index}}
-      (semantic.pgvector-api/diagnose pgvector index-metadata search-ctx expected-model expected-id))))
+  (if (semantic.util/lucene-backend?)
+    (lucene.query/diagnose search-ctx expected-model expected-id)
+    (let [pgvector       (semantic.env/get-pgvector-datasource!)
+          index-metadata (semantic.env/get-index-metadata)]
+      (if-not (index-active? pgvector index-metadata)
+        {:type :missing-from-index :details {:reason :no-active-index}}
+        (semantic.pgvector-api/diagnose pgvector index-metadata search-ctx expected-model expected-id)))))
 
 ;; NOTE:
 ;; we're currently not returning stats from `init!` as the async nature means
@@ -187,19 +204,17 @@
   "Initialize the semantic search table and populate it with initial data."
   :feature :semantic-search
   [searchable-documents opts]
-  (let [pgvector        (semantic.env/get-pgvector-datasource!)
-        index-metadata  (semantic.env/get-index-metadata)
-        embedding-model (semantic.env/get-configured-embedding-model)]
-    (semantic.pgvector-api/init-semantic-search! pgvector index-metadata embedding-model opts)
-    (semantic.pgvector-api/gate-updates! pgvector index-metadata searchable-documents)
-    nil))
+  (if (semantic.util/lucene-backend?)
+    (lucene.core/init! searchable-documents opts)
+    (let [pgvector        (semantic.env/get-pgvector-datasource!)
+          index-metadata  (semantic.env/get-index-metadata)
+          embedding-model (semantic.env/get-configured-embedding-model)]
+      (semantic.pgvector-api/init-semantic-search! pgvector index-metadata embedding-model opts)
+      (semantic.pgvector-api/gate-updates! pgvector index-metadata searchable-documents)
+      nil)))
 
-(defenterprise repair-index!
-  "Brings the semantic search index into consistency with the provided document set.
-  Does not fully reinitialize the index, but will add missing documents and remove stale ones.
-  Returns the repaired index ID, its stale-orphan count, and the pgvector timestamp captured before reading
-  the canonical stream."
-  :feature :semantic-search
+(defn- pgvector-repair-index!
+  "Repair the pgvector semantic index: backfill missing documents, count stale orphans, replay lost deletes."
   [searchable-documents]
   (let [pgvector       (semantic.env/get-pgvector-datasource!)
         index-metadata (semantic.env/get-index-metadata)
@@ -252,6 +267,17 @@
              :orphans        orphans
              :snapshot-at    snapshot-at}))))))
 
+(defenterprise repair-index!
+  "Brings the semantic search index into consistency with the provided document set.
+  Does not fully reinitialize the index, but will add missing documents and remove stale ones.
+  Returns the repaired index ID, its stale-orphan count, and the pgvector timestamp captured before reading
+  the canonical stream."
+  :feature :semantic-search
+  [searchable-documents]
+  (if (semantic.util/lucene-backend?)
+    (lucene.core/repair! searchable-documents)
+    (pgvector-repair-index! searchable-documents)))
+
 (comment
   (update-index! [{:model "card"
                    :id "1"

enterprise/backend/src/metabase_enterprise/semantic_search/db.clj

@@ -113,3 +113,95 @@
   "Delete the SemanticSearchTokenTracking rows created before `cutoff`."
   [cutoff :- ms/TemporalInstant]
   (t2/delete! :model/SemanticSearchTokenTracking {:where [:< :created_at cutoff]}))
+
+;;;; semantic_search_embedding -- the Lucene backend's vector store
+
+(def ^:private max-ids-per-query
+  "How many model IDs to list in one statement. Each is a bound parameter and Postgres caps a statement at 65535 of
+  them, so a caller passing a very large set is split across several queries rather than failing outright."
+  5000)
+
+(mu/defn embeddings-by-content-hash
+  "The `content_hash` and `embedding` of the rows in `space` whose text hashes to one of `hashes`."
+  [space  :- :string
+   hashes :- [:set :string]]
+  (t2/select [:model/SemanticSearchEmbedding :content_hash :embedding]
+             {:where [:and
+                      [:= :embedding_space_id space]
+                      [:in :content_hash hashes]]}))
+
+(mu/defn delete-embeddings!
+  "Delete the rows of `space` for `model` and `model-ids`, returning how many were deleted."
+  [space     :- :string
+   model     :- :string
+   model-ids :- [:sequential :string]]
+  (transduce (map (fn [chunk]
+                    (t2/delete! :model/SemanticSearchEmbedding
+                                :embedding_space_id space
+                                :model model
+                                :model_id [:in chunk])))
+             +
+             (partition-all max-ids-per-query model-ids)))
+
+(mu/defn insert-embeddings!
+  "Insert embedding `rows`."
+  [rows :- [:sequential ::semantic-search.schema/semantic-search-embedding.insert]]
+  (t2/insert! :model/SemanticSearchEmbedding rows))
+
+(mu/defn delete-embedding-space!
+  "Delete every row of `space`, returning how many were deleted."
+  [space :- :string]
+  (t2/delete! :model/SemanticSearchEmbedding :embedding_space_id space))
+
+(mu/defn delete-embeddings-outside-space!
+  "Delete every row that is not in `space`, returning how many were deleted."
+  [space :- :string]
+  (t2/delete! :model/SemanticSearchEmbedding :embedding_space_id [:not= space]))
+
+(mu/defn count-embeddings
+  "How many rows `space` has."
+  [space :- :string]
+  (t2/count :model/SemanticSearchEmbedding :embedding_space_id space))
+
+(mu/defn embedding-space-stats
+  "The row count and latest `updated_at` of `space`, as `{:n … :mx …}`."
+  [space :- :string]
+  (t2/query-one {:select [[[:count :*] :n] [[:max :updated_at] :mx]]
+                 :from   [:semantic_search_embedding]
+                 :where  [:= :embedding_space_id space]}))
+
+(mu/defn embeddings-after
+  "Up to `limit` rows of `space` with `id` above `after-id`, lowest id first, optionally written at or after `since`."
+  [space    :- :string
+   after-id :- ms/Int
+   since    :- [:maybe [:fn {:error/message "a timestamp the JDBC driver accepts as a parameter"}
+                        #(or (inst? %) (instance? java.time.temporal.Temporal %))]]
+   limit    :- ms/PositiveInt]
+  (t2/select :model/SemanticSearchEmbedding
+             {:where    (cond-> [:and
+                                 [:= :embedding_space_id space]
+                                 [:> :id after-id]]
+                          since (conj [:>= :updated_at since]))
+              :order-by [[:id :asc]]
+              :limit    limit}))
+
+(mu/defn embedding-model-ids
+  "The `{:model … :model_id …}` of every row in `space`."
+  [space :- :string]
+  (t2/query {:select [:model :model_id]
+             :from   [:semantic_search_embedding]
+             :where  [:= :embedding_space_id space]}))
+
+(mu/defn embeddings-for-model
+  "The rows of `space` for `model` and `model-ids`."
+  [space     :- :string
+   model     :- :string
+   model-ids :- [:sequential :string]]
+  (into []
+        (mapcat (fn [chunk]
+                  (t2/select :model/SemanticSearchEmbedding
+                             {:where [:and
+                                      [:= :embedding_space_id space]
+                                      [:= :model model]
+                                      [:in :model_id chunk]]})))
+        (partition-all max-ids-per-query model-ids)))

enterprise/backend/src/metabase_enterprise/semantic_search/health.clj

@@ -61,7 +61,8 @@
   ;; Active, not merely available: an available-but-inactive engine (another engine selected) has no index
   ;; by design, and must not read as a standing "No active semantic search index" incident. Active folds in
   ;; the license and kill switch, so a disabled instance neither records runs nor probes the embedder.
-  (when (semantic.util/semantic-search-active?)
+  (when (and (not (semantic.util/lucene-backend?))
+             (semantic.util/semantic-search-active?))
     ;; Acquire the datasource inside the try, so a malformed MB_PGVECTOR_DB_URL reads as degraded instead of
     ;; throwing out of the check.
     (let [active (try
@@ -149,7 +150,8 @@
 (defn- active-index*
   "The active semantic index and pgvector datasource, or nil when there is nothing applicable to measure."
   []
-  (when (semantic.util/semantic-search-active?)
+  (when (and (not (semantic.util/lucene-backend?))
+             (semantic.util/semantic-search-active?))
     (try
       (let [pgvector (semantic.env/get-pgvector-datasource!)
             state    (semantic.index-metadata/get-active-index-state pgvector (semantic.env/get-index-metadata))]

enterprise/backend/src/metabase_enterprise/semantic_search/index.clj

@@ -130,7 +130,7 @@
     (= 1 b) true
     :else (throw (ex-info "Unexpected boolean value" {:v b}))))
 
-(defn- batch-resolve-personal-owner-ids
+(defn batch-resolve-personal-owner-ids
   "Given a seq of collection-ids, return a map of collection-id -> personal_owner_id.
    Collections not in any personal tree will be absent from the map.
    Uses at most 2 queries regardless of the number of collection-ids."
@@ -977,7 +977,7 @@
   Wrapped in `delay` so the registry can populate before first read."
   (delay (compute-collection-id-only-search-models)))
 
-(defn- filter-read-permitted
+(defn filter-read-permitted
   "Returns the subset of `docs` whose t2 instances pass `mi/can-read?` for the current user."
   [docs]
   (let [timer (u/start-timer)
@@ -1027,7 +1027,7 @@
                          (str/starts-with? (:location collection) (str "/" collection-id "/")))))))
              docs)))
 
-(defn- apply-collection-id-filter
+(defn apply-collection-id-filter
   "Apply collection ID filtering with logging."
   [search-context docs]
   (let [collection-id (:collection search-context)]

enterprise/backend/src/metabase_enterprise/semantic_search/init.clj

@@ -4,6 +4,7 @@
    [metabase-enterprise.semantic-search.embedding-health]
    [metabase-enterprise.semantic-search.events]
    [metabase-enterprise.semantic-search.health]
+   [metabase-enterprise.semantic-search.lucene.sync]
    [metabase-enterprise.semantic-search.settings]
    [metabase-enterprise.semantic-search.task.index-cleanup]
    [metabase-enterprise.semantic-search.task.index-repair]

enterprise/backend/src/metabase_enterprise/semantic_search/lucene/core.clj

@@ -0,0 +1,108 @@
+(ns metabase-enterprise.semantic-search.lucene.core
+  "Write path for the Lucene semantic search backend.
+
+  Every ingestion batch lands in `semantic_search_embedding` first and in this node's Lucene index second, so a
+  write is searchable here immediately and on every other node once its sync catches up."
+  (:require
+   [java-time.api :as t]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.lucene.store :as lucene.store]
+   [metabase.util.log :as log]))
+
+(set! *warn-on-reflection* true)
+
+(def ^:private batch-size
+  "Documents per embedding request and per transaction. Matches the batch size ingestion already uses."
+  150)
+
+(defn- write-batch!
+  "Persist one batch of documents and index the rows that came back, returning its `{model n}` report."
+  [documents]
+  (let [{:keys [rows report]} (lucene.store/upsert-documents! documents)]
+    (lucene.index/upsert-rows! rows)
+    report))
+
+(defn update!
+  "Write a stream of ingestion documents to the table and this node's index, returning `{model n}` counts.
+
+  Documents whose embedding could not be produced are left out of the counts; the periodic repair backfills them."
+  [document-reducible]
+  (lucene.index/ensure-open!)
+  (or (transduce (comp (partition-all batch-size) (map write-batch!))
+                 (partial merge-with +)
+                 document-reducible)
+      {}))
+
+(defn delete!
+  "Remove `ids` of search `model` from the table and this node's index, returning `{model n}`."
+  [model ids]
+  (if (seq ids)
+    (do
+      (lucene.index/ensure-open!)
+      (let [deleted (lucene.store/delete-documents! model ids)]
+        (lucene.index/delete-ids! (map #(lucene.index/document-id model %) ids))
+        {model deleted}))
+    {}))
+
+(defn init!
+  "Make sure this instance has semantic search embeddings, populating them from `searchable-documents` when it does not.
+
+  `:force-reset?` drops this embedding space's rows first; `:re-populate?` re-runs the whole corpus through the
+  write path even when rows are already there. Other nodes pick the result up through their own sync."
+  [searchable-documents {:keys [force-reset? re-populate?]}]
+  (lucene.index/ensure-open!)
+  (let [space (lucene.store/space-id)]
+    (when force-reset?
+      (log/info "Resetting semantic search embeddings")
+      (lucene.store/delete-space! space)
+      (lucene.store/delete-other-spaces! space)
+      (lucene.index/delete-all!))
+    (if (or force-reset? re-populate? (zero? (:n (lucene.store/space-stats space))))
+      (update! searchable-documents)
+      (do
+        (log/debug "Semantic search embeddings are already populated, skipping initial population")
+        {}))))
+
+(defn- model-id-of [document]
+  [(:model document) (str (:id document))])
+
+(defn- backfill-batch!
+  "Persist only the documents of this batch that `stored` does not already have, recording every id in `seen`."
+  [stored seen documents]
+  (swap! seen into (map model-id-of) documents)
+  (let [absent (remove (comp stored model-id-of) documents)]
+    (if (seq absent)
+      (write-batch! absent)
+      {})))
+
+(defn repair!
+  "Bring the embedding table and this node's index back in line with `searchable-documents`.
+
+  Backfills documents the table is missing -- whatever an unavailable embedder skipped -- and drops rows for
+  documents that no longer exist, or that belong to an embedding space this instance has moved off. Rows that are
+  already there are left alone: rewriting them would move every one of their `updated_at` timestamps and make every
+  node in the cluster re-import the whole corpus on its next sync.
+
+  Returns `{:index-id … :orphans … :snapshot-at …}`, the shape
+  `metabase-enterprise.semantic-search.task.index-repair` reports on."
+  [searchable-documents]
+  (lucene.index/ensure-open!)
+  (let [space       (lucene.store/space-id)
+        snapshot-at (t/offset-date-time)
+        stored      (lucene.store/stored-model-ids space)
+        seen        (atom #{})]
+    (transduce (comp (partition-all batch-size)
+                     (map (partial backfill-batch! stored seen)))
+               (partial merge-with +)
+               searchable-documents)
+    (let [stale     (into [] (remove @seen) stored)
+          abandoned (lucene.store/delete-other-spaces! space)]
+      (doseq [[model pairs] (group-by first stale)]
+        (delete! model (map second pairs)))
+      (when (seq stale)
+        (log/infof "Dropped %d stale semantic search embeddings" (count stale)))
+      (when (pos? abandoned)
+        (log/infof "Dropped %d semantic search embeddings from abandoned embedding spaces" abandoned))
+      {:index-id    nil
+       :orphans     (count stale)
+       :snapshot-at snapshot-at})))

enterprise/backend/src/metabase_enterprise/semantic_search/lucene/index.clj

@@ -0,0 +1,356 @@
+(ns metabase-enterprise.semantic-search.lucene.index
+  "This node's Apache Lucene index: the vector store behind the `:lucene` semantic search backend.
+
+  The app-DB table `semantic_search_embedding` is the source of truth. This index is a disposable local
+  materialization of it — one directory per embedding space, under the plugins dir — that any node can rebuild
+  from the table.
+
+  The document schema below is the contract with [[metabase-enterprise.semantic-search.lucene.query]]:
+
+  | field                                                     | type                            | value |
+  |-----------------------------------------------------------|---------------------------------|-------|
+  | `id`                                                      | StringField (stored) + SortedDocValuesField | `\"<model>_<model_id>\"` |
+  | `model`, `model_id`                                       | StringField (stored)            | raw string |
+  | `display_type`                                            | StringField                     | raw string |
+  | `archived`, `verified`, `curated`                         | StringField                     | `\"true\"`/`\"false\"` |
+  | `collection_id`, `creator_id`, `last_editor_id`, `database_id` | StringField                | decimal string, absent when nil |
+  | `personal_owner_id`                                       | StringField                     | decimal string, or `\"__null__\"` |
+  | `model_created_at`, `model_updated_at`                    | LongPoint                       | epoch millis, absent when nil |
+  | `embedding`                                               | KnnFloatVectorField (COSINE)    | absent for degenerate vectors |
+  | `legacy_input`                                            | StoredField                     | the JSON string ingestion produced |"
+  (:require
+   [clojure.string :as str]
+   [java-time.api :as t]
+   [metabase-enterprise.semantic-search.embedding :as semantic.embedding]
+   [metabase-enterprise.semantic-search.models.embedding :as semantic.models.embedding]
+   [metabase.plugins.core :as plugins]
+   [metabase.util.date-2 :as u.date]
+   [metabase.util.files :as u.files]
+   [metabase.util.log :as log])
+  (:import
+   (java.nio.file FileVisitOption Files Path)
+   (java.time.temporal Temporal)
+   (org.apache.lucene.document Document Field$Store KnnFloatVectorField LongPoint SortedDocValuesField StoredField
+                               StringField)
+   (org.apache.lucene.index CorruptIndexException DocValues IndexNotFoundException IndexWriter IndexWriterConfig
+                            IndexWriterConfig$OpenMode LeafReader LeafReaderContext SortedDocValues Term
+                            VectorSimilarityFunction)
+   (org.apache.lucene.search DocIdSetIterator IndexSearcher SearcherFactory SearcherManager)
+   (org.apache.lucene.store Directory FSDirectory LockObtainFailedException)
+   (org.apache.lucene.util BytesRef)))
+
+(set! *warn-on-reflection* true)
+
+(def ^:const max-dimensions
+  "Widest vector Lucene's default codec will index. `Lucene99HnswVectorsFormat#getMaxDimensions` returns 1024;
+  going wider needs a custom `Codec`.
+  https://lucene.apache.org/core/10_5_1/core/org/apache/lucene/codecs/KnnVectorsFormat.html"
+  1024)
+
+(def ^:const null-owner
+  "Sentinel indexed in `personal_owner_id` for documents outside any personal collection.
+
+  Lucene's `FieldExistsQuery` does not cover `StringField`, so \"has no personal owner\" needs a real term."
+  "__null__")
+
+(def ^:dynamic *index-root*
+  "Directory holding one subdirectory per embedding space. Defaults to `<plugins dir>/semantic-search`."
+  nil)
+
+(defonce ^:private state
+  ;; {:space-id :dims :root :path :directory :writer :manager}, or nil when no index is open on this node.
+  (atom nil))
+
+(defonce ^:private lock (Object.))
+
+(defn document-id
+  "Lucene document id for `model`/`model-id`, the term used to update and delete a single search document."
+  ^String [model model-id]
+  (str model "_" model-id))
+
+(defn- index-root ^Path []
+  (if *index-root*
+    (u.files/get-path (str *index-root*))
+    (u.files/append-to-path (plugins/plugins-dir) "semantic-search")))
+
+(defn- space-file-name ^String [space-id]
+  (str/replace (str space-id) #"[^A-Za-z0-9._-]" "_"))
+
+(defn- space-dir ^Path [space-id]
+  (u.files/append-to-path (index-root) (space-file-name space-id)))
+
+(defn sidecar-path
+  "Path of a file sitting next to `space-id`'s index directory, for state Lucene's own directory cannot hold.
+
+  Beside rather than inside: `IndexWriter` owns the files in its directory."
+  ^Path [space-id ^String suffix]
+  (u.files/append-to-path (index-root) (str (space-file-name space-id) suffix)))
+
+(defn- delete-recursive! [^Path path]
+  (when (u.files/exists? path)
+    (with-open [stream (Files/walk path (into-array FileVisitOption []))]
+      (doseq [^Path p (rseq (vec (iterator-seq (.iterator stream))))]
+        (try (Files/delete p)
+             (catch Exception e
+               (log/warnf "Failed to delete %s: %s" p (ex-message e))))))))
+
+(defn- open-directory! [^Path path space-id dims]
+  (u.files/create-dir-if-not-exists! path)
+  (let [directory (FSDirectory/open path)
+        config    (doto (IndexWriterConfig.)
+                    (.setOpenMode IndexWriterConfig$OpenMode/CREATE_OR_APPEND))
+        writer    (IndexWriter. directory config)]
+    {:space-id  space-id
+     :dims      dims
+     :root      (index-root)
+     :path      path
+     :directory directory
+     :writer    writer
+     :manager   (SearcherManager. writer (SearcherFactory.))}))
+
+(defn- private-dir
+  "A directory of this JVM's own, for when another process already holds the lock on the shared one."
+  ^Path [space-id]
+  (u.files/append-to-path (u.files/get-path (System/getProperty "java.io.tmpdir"))
+                          "metabase-semantic-search"
+                          (space-file-name space-id)))
+
+(defn- open-index!
+  "Open the index directory for `space-id`, wiping and recreating it once if the existing one cannot be opened."
+  [space-id dims]
+  (when (> dims max-dimensions)
+    (throw (ex-info (str "Embedding model is too wide for the Lucene semantic search index: "
+                         dims " dimensions, maximum " max-dimensions)
+                    {:dims dims :max-dimensions max-dimensions})))
+  (let [path (space-dir space-id)]
+    (try
+      (open-directory! path space-id dims)
+      (catch LockObtainFailedException e
+        ;; Another live JVM owns this directory -- two containers sharing MB_PLUGINS_DIR. Deleting it would
+        ;; unlink the segments out from under that process, so take a private copy instead; the app-DB table
+        ;; refills it within one sync tick.
+        (log/warnf "Another process holds the semantic search Lucene index at %s (%s); using a private copy under %s"
+                   path (ex-message e) (private-dir space-id))
+        (open-directory! (private-dir space-id) space-id dims))
+      (catch Exception e
+        ;; A half-written directory is recoverable: the app-DB table can refill it.
+        (if (some #(instance? % e) [CorruptIndexException IndexNotFoundException])
+          (do
+            (log/warnf "Recreating unreadable semantic search Lucene index at %s: %s" path (ex-message e))
+            (delete-recursive! path)
+            (open-directory! path space-id dims))
+          (throw e))))))
+
+(defn- close-state! [{:keys [^SearcherManager manager ^IndexWriter writer ^Directory directory path]}]
+  (doseq [[what ^java.io.Closeable closeable] [["searcher manager" manager] ["writer" writer] ["directory" directory]]
+          :when closeable]
+    (try (.close closeable)
+         (catch Exception e
+           (log/warnf "Failed to close semantic search Lucene %s at %s: %s" what path (ex-message e))))))
+
+(defn close!
+  "Close this node's Lucene index. A no-op when none is open."
+  []
+  (locking lock
+    (when-let [s @state]
+      (close-state! s)
+      (reset! state nil)
+      (log/infof "Closed semantic search Lucene index at %s" (:path s))))
+  nil)
+
+(defn- configured-space
+  "Return `{:space-id … :dims …}` for the embedding model this instance is configured to use.
+
+  Throws when the configured embedding provider is not installed."
+  []
+  (let [resolved (semantic.embedding/resolve-model (semantic.embedding/get-configured-model))]
+    {:space-id (:embedding-space-id resolved)
+     :dims     (:vector-dimensions resolved)}))
+
+(defn ensure-open!
+  "Open this node's Lucene index for `space-id`, returning its state map.
+
+  Reopens when the embedding space (or [[*index-root*]]) changed since the last call, so a model switch lands in
+  a fresh directory. Throws when `dims` exceeds [[max-dimensions]] or the directory cannot be created."
+  ([]
+   (let [{:keys [space-id dims]} (configured-space)]
+     (ensure-open! space-id dims)))
+  ([space-id dims]
+   (let [open-for? (fn [s] (and s (= space-id (:space-id s)) (= (index-root) (:root s))))]
+     (if (open-for? @state)
+       @state
+       (locking lock
+         (if (open-for? @state)
+           @state
+           (do
+             (some-> @state close-state!)
+             (let [opened (open-index! space-id dims)]
+               (reset! state opened)
+               (log/infof "Opened semantic search Lucene index at %s" (:path opened))
+               opened))))))))
+
+(defn- current-state []
+  (or @state
+      (throw (ex-info "Semantic search Lucene index is not open" {}))))
+
+(defn do-with-searcher
+  "Call `f` with an `IndexSearcher` over this node's index, releasing it afterwards. See [[with-searcher]]."
+  [f]
+  (let [^SearcherManager manager (:manager (current-state))
+        searcher                 (.acquire manager)]
+    (try
+      (f searcher)
+      (finally
+        (.release manager searcher)))))
+
+(defmacro with-searcher
+  "Evaluate `body` with `binding` bound to an `IndexSearcher` over this node's index.
+
+  The searcher is released on the way out, so do not let it — or anything reading through it — escape `body`."
+  {:style/indent 1}
+  [[binding] & body]
+  `(do-with-searcher (fn [~binding] ~@body)))
+
+;;;; Writing
+
+(defn- epoch-millis
+  "Epoch millis for a timestamp that has been through JSON, or nil when absent or unparsable.
+
+  Timestamps without an offset are read as UTC, which is how the app DB stores them."
+  [v]
+  (try
+    (some-> (cond
+              (string? v)                  (u.date/parse v "UTC")
+              (instance? Temporal v)       v
+              (instance? java.util.Date v) (t/instant v))
+            t/instant
+            inst-ms)
+    (catch Exception e
+      (log/debugf "Ignoring unparsable search timestamp %s: %s" (pr-str v) (ex-message e))
+      nil)))
+
+(defn- boolean-term
+  "The indexed form of a boolean attribute, or nil when it is absent.
+
+  MySQL returns booleans as 0/1 rather than false/true, so the raw value cannot go straight into a term."
+  ^String [v]
+  (cond
+    (nil? v)     nil
+    (boolean? v) (str v)
+    (= 0 v)      "false"
+    (= 1 v)      "true"
+    :else        (str (boolean v))))
+
+(defn- add-term! [^Document doc ^String field value stored?]
+  (when (some? value)
+    (.add doc (StringField. field (str value) (if stored? Field$Store/YES Field$Store/NO)))))
+
+(defn- add-long-point! [^Document doc ^String field v]
+  (when-let [millis (epoch-millis v)]
+    (.add doc (LongPoint. field (long-array [millis])))))
+
+(defn- add-vector!
+  "Add the kNN field unless `bytes` decodes to a vector Lucene cannot score under cosine similarity."
+  [^Document doc ^bytes embedding dims]
+  (let [^floats v (semantic.models.embedding/bytes->floats embedding)]
+    (cond
+      (not= (alength v) (int dims))
+      (log/warnf "Skipping embedding of width %d in an index of width %d" (alength v) dims)
+
+      ;; COSINE throws on the zero vector; index the document without it so the id still tracks the table row.
+      (every? zero? v)
+      (log/warn "Skipping all-zero embedding, document will not be reachable by vector search")
+
+      :else
+      (.add doc (KnnFloatVectorField. "embedding" v VectorSimilarityFunction/COSINE)))))
+
+(defn- row->document
+  "Build the Lucene document for one `semantic_search_embedding` row. See the schema table in this namespace."
+  ^Document [{:keys [model model_id embedding document]} dims]
+  (let [doc (Document.)
+        id  (document-id model model_id)]
+    (add-term! doc "id" id true)
+    (.add doc (SortedDocValuesField. "id" (BytesRef. id)))
+    (add-term! doc "model" model true)
+    (add-term! doc "model_id" model_id true)
+    (add-term! doc "display_type" (:display_type document) false)
+    (add-term! doc "archived" (or (boolean-term (:archived document)) "false") false)
+    (add-term! doc "verified" (boolean-term (:verified document)) false)
+    (add-term! doc "curated" (boolean-term (:curated document)) false)
+    (add-term! doc "collection_id" (:collection_id document) false)
+    (add-term! doc "creator_id" (:creator_id document) false)
+    (add-term! doc "last_editor_id" (:last_editor_id document) false)
+    (add-term! doc "database_id" (:database_id document) false)
+    (add-term! doc "personal_owner_id" (or (:personal_owner_id document) null-owner) false)
+    (add-long-point! doc "model_created_at" (:created_at document))
+    (add-long-point! doc "model_updated_at" (:updated_at document))
+    (when embedding
+      (add-vector! doc embedding dims))
+    (.add doc (StoredField. "legacy_input" ^String (str (:legacy_input document))))
+    doc))
+
+(defn- refresh! [{:keys [^IndexWriter writer ^SearcherManager manager]}]
+  (.commit writer)
+  (.maybeRefresh manager))
+
+(defn upsert-rows!
+  "Index `rows` read from `semantic_search_embedding`, replacing any document with the same [[document-id]].
+
+  Commits and refreshes the searcher before returning the number of documents written."
+  [rows]
+  (let [{:keys [^IndexWriter writer dims] :as s} (current-state)
+        written (reduce (fn [n {:keys [model model_id] :as row}]
+                          (.updateDocument writer
+                                           (Term. "id" (document-id model model_id))
+                                           (row->document row dims))
+                          (inc n))
+                        0
+                        rows)]
+    (when (pos? written)
+      (refresh! s))
+    written))
+
+(defn delete-ids!
+  "Remove the documents with the given [[document-id]]s, returning the number of ids submitted."
+  [ids]
+  (let [{:keys [^IndexWriter writer] :as s} (current-state)
+        terms (mapv #(Term. "id" ^String %) ids)]
+    (when (seq terms)
+      (let [^"[Lorg.apache.lucene.index.Term;" arr (into-array Term terms)]
+        (.deleteDocuments writer arr))
+      (refresh! s))
+    (count terms)))
+
+(defn delete-all!
+  "Remove every document from this node's index."
+  []
+  (let [{:keys [^IndexWriter writer] :as s} (current-state)]
+    (.deleteAll writer)
+    (refresh! s)
+    nil))
+
+;;;; Reading
+
+(defn- leaf-ids [^LeafReaderContext ctx]
+  (let [^LeafReader reader   (.reader ctx)
+        live                 (.getLiveDocs reader)
+        ^SortedDocValues dvs (DocValues/getSorted reader "id")]
+    (loop [acc (transient [])]
+      (let [doc (.nextDoc dvs)]
+        (if (= doc DocIdSetIterator/NO_MORE_DOCS)
+          (persistent! acc)
+          (recur (if (or (nil? live) (.get live doc))
+                   (conj! acc (.utf8ToString (.lookupOrd dvs (.ordValue dvs))))
+                   acc)))))))
+
+(defn live-ids
+  "Set of [[document-id]]s currently visible in this node's index."
+  []
+  (with-searcher [^IndexSearcher searcher]
+    (into #{} (mapcat leaf-ids) (.leaves (.getIndexReader searcher)))))
+
+(defn live-count
+  "Number of documents currently visible in this node's index."
+  []
+  (with-searcher [^IndexSearcher searcher]
+    (.numDocs (.getIndexReader searcher))))

enterprise/backend/src/metabase_enterprise/semantic_search/lucene/query.clj

@@ -0,0 +1,329 @@
+(ns metabase-enterprise.semantic-search.lucene.query
+  "Read path for the Lucene semantic search backend.
+
+  Two arms answer every query: kNN over this node's Lucene index, and the OSS appdb keyword engine. Their
+  rankings are fused with reciprocal rank fusion — the same constants the pgvector arm applies in SQL — so a
+  document that only one arm finds still places well, and one both arms like places best."
+  (:require
+   [clojure.string :as str]
+   [metabase-enterprise.semantic-search.appdb-scoring :as appdb-scoring]
+   [metabase-enterprise.semantic-search.embedding :as semantic.embedding]
+   [metabase-enterprise.semantic-search.index :as semantic.index]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.scoring :as semantic.scoring]
+   [metabase-enterprise.semantic-search.settings :as semantic.settings]
+   [metabase.search.config :as search.config]
+   [metabase.search.core :as search]
+   [metabase.search.engine :as search.engine]
+   [metabase.search.scoring :as search.scoring]
+   [metabase.util :as u]
+   [metabase.util.json :as json]
+   [metabase.util.log :as log])
+  (:import
+   (org.apache.lucene.index StoredFields Term)
+   (org.apache.lucene.search BooleanClause$Occur BooleanQuery$Builder IndexSearcher KnnFloatVectorQuery
+                             MatchAllDocsQuery MatchNoDocsQuery Query ScoreDoc TermInSetQuery TermQuery TopDocs)
+   (org.apache.lucene.util BytesRef)))
+
+(set! *warn-on-reflection* true)
+
+(def ^:private score-floor
+  "Lucene cosine score below which a hit does not count as a match.
+
+  Lucene scores cosine similarity as `(1 + cos) / 2`; the pgvector arm cuts at cosine distance `1 - cos <= 0.7`,
+  which is the same line.
+  https://lucene.apache.org/core/10_5_1/core/org/apache/lucene/index/VectorSimilarityFunction.html"
+  0.65)
+
+(def ^:private rrf-k
+  "Reciprocal-rank-fusion smoothing constant, matching [[metabase-enterprise.semantic-search.scoring]]."
+  60)
+
+(def ^:private keyword-weight 0.51)
+(def ^:private semantic-weight 0.49)
+
+;;;; Filters
+;;;;
+;;;; These mirror the pgvector arm's `filter-conditions`, key for key, so the search debug endpoint reports the
+;;;; same filter names whichever backend answered.
+
+(defn- term-query ^Query [^String field value]
+  (TermQuery. (Term. field (str value))))
+
+(defn- terms-query ^Query [^String field values]
+  (TermInSetQuery. field (mapv #(BytesRef. (str %)) values)))
+
+(defn- any-of ^Query [queries]
+  (let [builder (BooleanQuery$Builder.)]
+    (doseq [^Query q queries]
+      (.add builder q BooleanClause$Occur/SHOULD))
+    (.setMinimumNumberShouldMatch builder 1)
+    (.build builder)))
+
+(defn- none-of ^Query [^Query q]
+  ;; Lucene needs a positive clause: a BooleanQuery of MUST_NOT alone matches nothing.
+  (-> (BooleanQuery$Builder.)
+      (.add (MatchAllDocsQuery.) BooleanClause$Occur/FILTER)
+      (.add q BooleanClause$Occur/MUST_NOT)
+      (.build)))
+
+(defn- personal-collection-query
+  "The `filter-items-in-personal-collection` clause. See the table in
+  [[metabase-enterprise.semantic-search.index]] for what each mode admits."
+  ^Query [{:keys [filter-items-in-personal-collection current-user-id]}]
+  (let [no-owner (term-query "personal_owner_id" lucene.index/null-owner)]
+    (case (or filter-items-in-personal-collection "all")
+      "all"            nil
+      "only-mine"      (term-query "personal_owner_id" current-user-id)
+      "only"           (none-of no-owner)
+      "exclude"        no-owner
+      "exclude-others" (any-of [no-owner (term-query "personal_owner_id" current-user-id)]))))
+
+(defn- filter-clauses
+  "Ordered `[filter-key query]` pairs for the structural filters `search-context` implies."
+  [{:keys [archived? verified curated? models created-by last-edited-by table-db-id ids display-type]
+    :as search-context}]
+  (keep (fn [[k q]] (when q [k q]))
+        [[:personal-collection (personal-collection-query search-context)]
+         [:archived?           (when (some? archived?) (term-query "archived" (boolean archived?)))]
+         [:verified            (when (some? verified) (term-query "verified" verified))]
+         [:curated             (when (some? curated?) (term-query "curated" curated?))]
+         ;; An empty but present set of models means the other filters left no applicable model, so match
+         ;; nothing rather than everything.
+         [:models              (cond
+                                 (seq models)   (terms-query "model" models)
+                                 (some? models) (MatchNoDocsQuery.))]
+         [:created-by          (when (seq created-by) (terms-query "creator_id" created-by))]
+         [:last-edited-by      (when (seq last-edited-by) (terms-query "last_editor_id" last-edited-by))]
+         [:table-db-id         (when table-db-id (term-query "database_id" table-db-id))]
+         [:ids                 (when (seq ids) (terms-query "model_id" ids))]
+         [:display-type        (when (seq display-type) (terms-query "display_type" display-type))]]))
+
+(defn filter-query
+  "The Lucene filter for `search-context`, or nil when it filters nothing."
+  ^Query [search-context]
+  (when-let [clauses (seq (filter-clauses search-context))]
+    (let [builder (BooleanQuery$Builder.)]
+      (doseq [[_ ^Query q] clauses]
+        (.add builder q BooleanClause$Occur/FILTER))
+      (.build builder))))
+
+(defn- date-filtered?
+  "Whether `search-context` carries a date filter, which the vector arm cannot express yet.
+
+  The pgvector arm silently ignores these (it reads `:start` off a string), so neither backend narrows by date
+  today; here the query falls back to the keyword arm alone, which does apply them."
+  [{:keys [created-at last-edited-at]}]
+  (boolean (or created-at last-edited-at)))
+
+;;;; The two arms
+
+(defn- hit->row
+  "One kNN hit as `{:model :model_id :legacy_input :semantic-rank :semantic-score}`."
+  [^StoredFields stored rank ^ScoreDoc hit]
+  (let [doc (.document stored (.-doc hit))]
+    {:model          (.get doc "model")
+     :model_id       (.get doc "model_id")
+     :legacy_input   (json/decode+kw (.get doc "legacy_input"))
+     :semantic-rank  (inc rank)
+     ;; Lucene's cosine score is already the [0 1] "how close is this" score the pgvector arm derives from
+     ;; distance as `1 - d/2`.
+     :semantic-score (.-score hit)}))
+
+(defn- vector-hits
+  "kNN hits for `embedding`, nearest first, cut off at [[score-floor]]."
+  [embedding search-context limit]
+  (lucene.index/with-searcher [^IndexSearcher searcher]
+    (let [query         (KnnFloatVectorQuery. "embedding" embedding (int limit) (filter-query search-context))
+          ^TopDocs hits (.search searcher query (int limit))
+          stored        (.storedFields searcher)
+          matching      (filter #(<= score-floor (.-score ^ScoreDoc %)) (.scoreDocs hits))]
+      (vec (map-indexed (partial hit->row stored) matching)))))
+
+(defn- keyword-hits
+  "The appdb keyword engine's results for `search-context`, best first, or nil when that arm did not run.
+
+  Degrades rather than failing the search: the vector arm can still answer. The nil is load-bearing -- an empty
+  vector means the arm ran and matched nothing, while nil means the caller still owes the user a keyword search
+  and should fall back to another engine."
+  [search-context limit]
+  (if-not (search.engine/supported-engine? :search.engine/appdb)
+    (log/debug "Skipping the keyword arm of semantic search: this app DB cannot hold a search index")
+    (try
+      (into [] (take limit) (search.engine/results (assoc search-context :search-engine :search.engine/appdb)))
+      (catch Throwable t
+        (log/warnf "Keyword arm of semantic search failed, continuing with vector results only: %s" (ex-message t))
+        nil))))
+
+;;;; Fusion
+
+(defn- rrf-score
+  "Reciprocal rank fusion of one document's two ranks; a rank of nil contributes nothing."
+  [semantic-rank keyword-rank]
+  (+ (if semantic-rank (/ semantic-weight (+ rrf-k (double semantic-rank))) 0.0)
+     (if keyword-rank (/ keyword-weight (+ rrf-k (double keyword-rank))) 0.0)))
+
+(defn- result-key [row]
+  [(:model row) (:id row)])
+
+(defn- vector-entries [rows]
+  (into {}
+        (map (fn [{:keys [legacy_input semantic-rank semantic-score]}]
+               ;; `:is_published` is an internal permission signal and never leaves the engine. The timestamps
+               ;; came back out of JSON as strings, and callers -- `POST /api/agent/v1/search`'s response schema
+               ;; among them -- require the same real temporal values the appdb arm hands back.
+               (let [body (-> legacy_input
+                              (dissoc :is_published)
+                              search/collapse-id
+                              search/rehydrate-timestamps)]
+                 [(result-key body) {:body           body
+                                     :semantic-rank  semantic-rank
+                                     :semantic-score semantic-score}])))
+        rows))
+
+(defn- keyword-entries [rows]
+  (into {}
+        (map-indexed (fn [i row]
+                       [(result-key row) {:body         (dissoc row :score :all-scores)
+                                          :keyword-rank (inc i)}]))
+        rows))
+
+(defn- scored-result
+  [weights {:keys [body semantic-rank semantic-score keyword-rank]}]
+  (let [scores     {:rrf               (rrf-score semantic-rank keyword-rank)
+                    ;; A keyword-only hit never went through vector search, so it scores 0 here -- the
+                    ;; least-relevant end, not the perfect-match end a zero *distance* would mean.
+                    :semantic-distance (or semantic-score 0.0)}
+        all-scores (search.scoring/all-scores weights [:rrf :semantic-distance] scores)]
+    (assoc body
+           :score      (reduce + (map :contribution all-scores))
+           :all-scores all-scores)))
+
+(defn fuse
+  "Fuse the two arms' rows into one ranked result list, best first.
+
+  Deliberately does not fold the keyword arm's own total score in: reciprocal rank fusion spreads over about a
+  point while the appdb base scorers spread over several, so adding it would rank every vector-only hit below
+  every keyword hit and undo the point of the vector arm."
+  [weights vector-rows keyword-rows]
+  (->> (merge-with merge (vector-entries vector-rows) (keyword-entries keyword-rows))
+       vals
+       (map (partial scored-result weights))
+       (sort-by :score >)
+       vec))
+
+;;;; Query
+
+(defn- embed-query
+  "The query embedding, prefixed the way the configured model expects."
+  [search-string]
+  (let [model (semantic.embedding/get-configured-model)]
+    (float-array
+     (semantic.embedding/get-embedding model
+                                       (semantic.embedding/prefix-search-query model search-string)
+                                       {:type :query :record-tokens? true}))))
+
+(defn- vector-rows
+  "The vector arm's rows, or `[]` when it cannot contribute to this query."
+  [search-context limit]
+  (cond
+    (date-filtered? search-context)
+    (do
+      (log/debug "Skipping the vector arm of semantic search: date filters are keyword-only")
+      [])
+
+    (zero? (lucene.index/live-count))
+    (do
+      (log/debug "Semantic search Lucene index is empty, answering from the keyword arm alone")
+      [])
+
+    :else
+    (vector-hits (embed-query (:search-string search-context)) search-context limit)))
+
+(defn query
+  "Answer `search-context` from the Lucene vector arm fused with the appdb keyword arm.
+
+  Returns `{:results … :raw-count … :keyword-results …}`. `:raw-count` counts the fused set before permission
+  filtering -- the signal [[metabase-enterprise.semantic-search.core/results]] uses to decide whether to
+  supplement -- and `:keyword-results` are the appdb rows this query already paid for, so that supplement does not
+  run the same query a second time. `:keyword-results` is nil when the keyword arm did not run at all, which
+  leaves that caller to fall back to another engine as it would have without this backend.
+  Throws when the query cannot be embedded, which that caller turns into a keyword-only fallback."
+  [search-context]
+  (let [search-string (:search-string search-context)]
+    ;; A blank search string is the native-editor table picker, which semantic search does not serve.
+    (if (str/blank? search-string)
+      {:results [] :raw-count 0}
+      (let [timer     (u/start-timer)
+            limit     (semantic.settings/semantic-search-results-limit)
+            weights   (search.config/weights search-context)
+            _         (lucene.index/ensure-open!)
+            keyword-rows (keyword-hits search-context limit)
+            fused     (fuse weights (vector-rows search-context limit) keyword-rows)
+            permitted (->> fused
+                           semantic.index/filter-read-permitted
+                           (semantic.index/apply-collection-id-filter search-context))
+            results   (semantic.scoring/with-appdb-scores search-context
+                        (appdb-scoring/appdb-scorers search-context)
+                        weights
+                        permitted)]
+        (log/debug "Semantic search (Lucene)"
+                   {:search-string-length (count search-string)
+                    :fused-count          (count fused)
+                    :final-count          (count results)
+                    :total-time-ms        (u/since-ms timer)})
+        {:results         results
+         :raw-count       (count fused)
+         :keyword-results keyword-rows}))))
+
+;;;; Diagnostics
+
+(defn- indexed-legacy-input
+  "The decoded `legacy_input` this node has indexed for `document-id`, or nil when it has none."
+  [^String document-id]
+  (lucene.index/with-searcher [^IndexSearcher searcher]
+    (let [^TopDocs hits (.search searcher (TermQuery. (Term. "id" document-id)) (int 1))]
+      (when-let [^ScoreDoc hit (first (.scoreDocs hits))]
+        (-> (.document (.storedFields searcher) (.-doc hit))
+            (.get "legacy_input")
+            json/decode+kw)))))
+
+(defn- excluding-filter
+  "The first filter key whose clause excludes `document-id`, or nil when every clause admits it."
+  [search-context ^String document-id]
+  (lucene.index/with-searcher [^IndexSearcher searcher]
+    (let [id-query (TermQuery. (Term. "id" document-id))]
+      (some (fn [[k ^Query clause]]
+              (let [builder (doto (BooleanQuery$Builder.)
+                              (.add id-query BooleanClause$Occur/FILTER)
+                              (.add clause BooleanClause$Occur/FILTER))]
+                (when (zero? (.count searcher (.build builder)))
+                  k)))
+            (filter-clauses search-context)))))
+
+(defn diagnose
+  "Explain why `expected-model`/`expected-id` does not turn up for `search-context`.
+
+  Returns the first engine-owned stage that drops it (`:missing-from-index`, `:filtered`, `:not-matching`) or
+  `:candidate` when it survives all of them. See [[metabase.search.debug/diagnose]]."
+  [search-context expected-model expected-id]
+  (lucene.index/ensure-open!)
+  (let [document-id (lucene.index/document-id expected-model expected-id)
+        legacy      (indexed-legacy-input document-id)]
+    (cond
+      (nil? legacy)
+      {:type :missing-from-index :details {:document-id document-id}}
+
+      ;; Access control before the structural filters, so a not-permitted row reads as such even when a query
+      ;; filter would also drop it. Matches the pgvector arm and the appdb engine.
+      (empty? (semantic.index/filter-read-permitted [legacy]))
+      {:type :filtered :details {:excluded-by :permissions}}
+
+      :else
+      (if-let [excluded-by (excluding-filter search-context document-id)]
+        {:type :filtered :details {:excluded-by excluded-by}}
+        (if-let [hit (first (vector-hits (embed-query (:search-string search-context))
+                                         (assoc search-context :ids [expected-id] :models #{expected-model})
+                                         1))]
+          {:type :candidate :details {:semantic-distance (- 2.0 (* 2.0 (:semantic-score hit)))}}
+          {:type :not-matching :details {:score-floor score-floor}})))))

enterprise/backend/src/metabase_enterprise/semantic_search/lucene/store.clj

@@ -0,0 +1,183 @@
+(ns metabase-enterprise.semantic-search.lucene.store
+  "Reads and writes `semantic_search_embedding`, the app-DB table the Lucene semantic search backend indexes from.
+
+  Every node writes here; every node syncs its own Lucene index from here. Rows are scoped by embedding space, so
+  switching embedding model starts a fresh corpus without disturbing the old one."
+  (:require
+   [metabase-enterprise.semantic-search.db :as semantic-search.db]
+   [metabase-enterprise.semantic-search.embedding :as semantic.embedding]
+   [metabase-enterprise.semantic-search.index :as semantic.index]
+   [metabase-enterprise.semantic-search.models.embedding :as semantic.models.embedding]
+   [metabase.app-db.core :as mdb]
+   [metabase.util.log :as log]
+   [toucan2.core :as t2]))
+
+(set! *warn-on-reflection* true)
+
+(defn space-id
+  "Identifier of the embedding space the configured model writes into. Rows in other spaces are not ours.
+
+  Throws when the configured embedding provider is not installed."
+  []
+  (:embedding-space-id (semantic.embedding/resolve-model (semantic.embedding/get-configured-model))))
+
+;;;; Writing
+
+(defn- cached-embeddings
+  "Map of content hash → stored embedding bytes, for those of `hashes` already embedded in `space`."
+  [space hashes]
+  (when-let [hashes (not-empty (set hashes))]
+    (into {}
+          (map (juxt :content_hash :embedding))
+          (semantic-search.db/embeddings-by-content-hash space hashes))))
+
+(defn- embed-texts!
+  "Map of content hash → embedding bytes for `texts`, or `{}` when the embedder cannot be called right now.
+
+  Never throws: an unavailable embedding service must not stop metadata updates, and the periodic repair
+  re-embeds whatever was skipped."
+  [model texts]
+  (if (semantic.embedding/embedder-circuit-untrusted?)
+    (do
+      (log/warn "Skipping semantic search embeddings: the embedding service circuit breaker is not closed")
+      {})
+    (let [collected (atom {})]
+      (try
+        (semantic.embedding/process-embeddings-streaming
+         model
+         (vec texts)
+         (fn [text->embedding] (swap! collected merge text->embedding) nil)
+         {:type :index :record-tokens? true})
+        (into {}
+              (map (fn [[text embedding]]
+                     [(semantic.models.embedding/content-hash text)
+                      (semantic.models.embedding/floats->bytes embedding)]))
+              @collected)
+        (catch Throwable t
+          (log/warnf "Failed to generate semantic search embeddings, skipping this batch: %s" (ex-message t))
+          {})))))
+
+(def ^:private stored-document-keys
+  "The document keys a stored row keeps: exactly what `row->document` in
+  [[metabase-enterprise.semantic-search.lucene.index]] reads back, plus `model`/`id` for diagnostics.
+
+  An ingestion document also carries `searchable_text`, `embeddable_text` and `display_data`, which together are
+  two to three times the size of the rest and which nothing reads out of this table."
+  [:model :id :legacy_input :display_type :archived :verified :curated
+   :collection_id :creator_id :last_editor_id :database_id :created_at :updated_at])
+
+(defn- document-row
+  "The table row for one ingestion document, or nil when we have no embedding for it."
+  [{:keys [space dims owner-ids embeddings]} {::keys [content-hash] :as document}]
+  (when-let [embedding (get embeddings content-hash)]
+    {:embedding_space_id space
+     :model              (:model document)
+     :model_id           (str (:id document))
+     :name               (:name document)
+     :archived           (boolean (:archived document))
+     :content_hash       content-hash
+     :dims               dims
+     :embedding          embedding
+     ;; Enough of the document travels with the row that a node can rebuild its Lucene index from this table alone.
+     :document           (-> document
+                             (select-keys stored-document-keys)
+                             (assoc :personal_owner_id (get owner-ids (:collection_id document))))}))
+
+(defn- write-rows!
+  "Replace `rows` in one transaction. Toucan has no portable upsert, so delete the ids then insert them."
+  [space rows]
+  (mdb/with-conflict-retry
+    (t2/with-transaction [_conn]
+      (doseq [[model model-rows] (group-by :model rows)]
+        (semantic-search.db/delete-embeddings! space model (mapv :model_id model-rows)))
+      (semantic-search.db/insert-embeddings! rows))))
+
+(defn upsert-documents!
+  "Persist one batch of ingestion `documents` into `semantic_search_embedding`.
+
+  Text that is already embedded in this space reuses the stored vector, so metadata-only changes — an archive, a
+  view-count bump — never call the embedding provider. Documents whose embedding is missing and cannot be produced
+  right now are skipped, and the periodic repair backfills them.
+
+  Returns `{:rows <the rows written> :report {model n}}`."
+  [documents]
+  (let [documents (vec documents)]
+    (if (empty? documents)
+      {:rows [] :report {}}
+      (let [model      (semantic.embedding/get-configured-model)
+            space      (:embedding-space-id (semantic.embedding/resolve-model model))
+            hashed     (mapv #(assoc % ::content-hash (semantic.models.embedding/content-hash (:embeddable_text %)))
+                             documents)
+            cached     (cached-embeddings space (map ::content-hash hashed))
+            to-embed   (into #{}
+                             (comp (remove #(contains? cached (::content-hash %)))
+                                   (map :embeddable_text))
+                             hashed)
+            embeddings (merge cached (when (seq to-embed) (embed-texts! model to-embed)))
+            context    {:space      space
+                        :dims       (:vector-dimensions model)
+                        :owner-ids  (semantic.index/batch-resolve-personal-owner-ids (map :collection_id documents))
+                        :embeddings embeddings}
+            rows       (into [] (keep (partial document-row context)) hashed)]
+        (when (seq rows)
+          (write-rows! space rows))
+        {:rows rows :report (frequencies (map :model rows))}))))
+
+(defn delete-documents!
+  "Delete the rows for `model` and `ids` in the current embedding space, returning the number deleted."
+  [model ids]
+  (if (seq ids)
+    (semantic-search.db/delete-embeddings! (space-id) model (mapv str ids))
+    0))
+
+(defn delete-space!
+  "Delete every row of `space`, returning the number deleted."
+  [space]
+  (semantic-search.db/delete-embedding-space! space))
+
+(defn delete-other-spaces!
+  "Delete the rows of every embedding space but `space`, returning the number deleted.
+
+  Switching embedding model abandons a whole space, and nothing else reclaims it under this backend: the pgvector
+  index-cleanup job does not run here."
+  [space]
+  ;; This assumes one configured embedding model per deployment. Mid-rollout, nodes on the old and the new model
+  ;; each see the other's rows as abandoned and delete them, so the corpus is re-embedded once per flip until they
+  ;; converge -- costly but self-correcting, since embeddings are derived data. Scoping the delete by age or by
+  ;; node would trade that for rows that are never reclaimed at all.
+  ;;
+  ;; The per-node index directory of an abandoned space is left on disk; evicting sibling directories is a
+  ;; deliberate follow-up (PLAN cut list item 1), and a restart is the workaround.
+  (semantic-search.db/delete-embeddings-outside-space! space))
+
+(defn stored-model-ids
+  "Set of `[model model_id]` pairs already stored in `space`."
+  [space]
+  (into #{} (map (juxt :model :model_id)) (semantic-search.db/embedding-model-ids space)))
+
+;;;; Reading
+
+(defn space-stats
+  "`{:n <row count> :mx <latest updated_at>}` for `space` — the watermark a node syncs its Lucene index against."
+  [space]
+  (semantic-search.db/embedding-space-stats space))
+
+(defn rows-after
+  "Up to `limit` rows of `space` whose `id` is above `after-id`, lowest id first.
+
+  `since` bounds the scan to rows written at or after it; pass nil to walk the whole space."
+  [space after-id since limit]
+  (semantic-search.db/embeddings-after space after-id since limit))
+
+(defn model-ids
+  "Every `{:model … :model_id …}` in `space`, for reconciling a local index against the table."
+  [space]
+  (semantic-search.db/embedding-model-ids space))
+
+(defn rows-for-model-ids
+  "Rows of `space` for the given `[model model-id]` pairs."
+  [space model+ids]
+  (into []
+        (mapcat (fn [[model pairs]]
+                  (semantic-search.db/embeddings-for-model space model (mapv second pairs))))
+        (group-by first model+ids)))

enterprise/backend/src/metabase_enterprise/semantic_search/lucene/sync.clj

@@ -0,0 +1,200 @@
+(ns metabase-enterprise.semantic-search.lucene.sync
+  "Keeps this node's Lucene index in step with `semantic_search_embedding`.
+
+  Quartz jobs are clustered, so they fire on one node — but every node has its own Lucene index to refresh. This
+  runs on a plain per-JVM daemon timer instead. A write made on this node is searchable here immediately (the
+  write path indexes it directly) and on every other node within one tick."
+  (:require
+   [clojure.edn :as edn]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.lucene.store :as lucene.store]
+   [metabase-enterprise.semantic-search.util :as semantic.util]
+   [metabase.startup.core :as startup]
+   [metabase.util :as u]
+   [metabase.util.files :as u.files]
+   [metabase.util.log :as log])
+  (:import
+   (java.nio.file Path)
+   (java.sql Timestamp)
+   (java.time Instant LocalDateTime OffsetDateTime ZonedDateTime)
+   (java.util.concurrent Executors ScheduledExecutorService ThreadFactory TimeUnit)))
+
+(set! *warn-on-reflection* true)
+
+(def ^:private tick-seconds 10)
+
+(def ^:private page-size
+  "Rows per keyset page while importing. Bounds memory on the first tick, which imports the whole space."
+  2000)
+
+(def ^:private watermark-overlap-seconds
+  "How far back of the last watermark to re-read. Covers a row that committed late with an older `updated_at`
+  than one this node has already seen; re-importing a row it already has is idempotent."
+  60)
+
+(def ^:private id-diff-interval-ms
+  "How long to go without a full id comparison. Deletes normally show up through the row count, and this is the
+  backstop for the case where an insert and a delete land in the same tick."
+  (* 5 60 1000))
+
+(defonce ^:private executor (atom nil))
+
+(defonce ^:private state
+  ;; {:space :n :mx-ms :diff-timer}, describing what this node last synced. nil until the first tick.
+  (atom nil))
+
+(defn- watermark-ms
+  "Epoch millis of the `max(updated_at)` the app DB reported, whatever JDBC type it came back as, or nil.
+
+  Millis rather than the value itself: it is compared across a restart, where it has been through EDN, and the
+  drivers do not agree on a type. Local timestamps are read in the JVM default zone, exactly as
+  `java.sql.Timestamp` converts them, so the value round-trips back into a query bound unchanged."
+  [v]
+  (cond
+    (nil? v)                        nil
+    (instance? Timestamp v)         (.getTime ^Timestamp v)
+    (instance? LocalDateTime v)     (.getTime (Timestamp/valueOf ^LocalDateTime v))
+    (instance? OffsetDateTime v)    (.toEpochMilli (.toInstant ^OffsetDateTime v))
+    (instance? ZonedDateTime v)     (.toEpochMilli (.toInstant ^ZonedDateTime v))
+    (instance? Instant v)           (.toEpochMilli ^Instant v)
+    (instance? java.util.Date v)    (.getTime ^java.util.Date v)
+    :else                           nil))
+
+(defn- watermark-file ^Path [space]
+  (lucene.index/sidecar-path space ".sync.edn"))
+
+(defn- read-watermark
+  "What this node had synced when it last shut down, or nil when there is no usable record.
+
+  Without this a restart walks the whole space again to refill an index that is already on disk."
+  [space]
+  (let [file (watermark-file space)]
+    (when (u.files/exists? file)
+      (try
+        (let [{:keys [n mx-ms]} (edn/read-string (slurp (.toFile file)))]
+          {:space      space
+           :n          n
+           :mx-ms      mx-ms
+           :diff-timer (u/start-timer)})
+        (catch Throwable t
+          (log/warnf "Ignoring unreadable semantic search sync watermark at %s: %s" file (ex-message t))
+          nil)))))
+
+(defn- write-watermark!
+  "Record `{:n … :mx-ms …}` beside the index. Never throws: an unwritable plugins dir costs a full import, not a sync."
+  [space {:keys [n mx-ms]}]
+  (try
+    (spit (.toFile (watermark-file space)) (pr-str {:n n :mx-ms mx-ms}))
+    (catch Throwable t
+      (log/warnf "Could not persist the semantic search sync watermark: %s" (ex-message t)))))
+
+(defn reset-state!
+  "Forget what this node last synced, so the next tick does a full import and id comparison."
+  []
+  (reset! state nil))
+
+(defn- import-rows!
+  "Index every row of `space` written at or after `since`, in keyset pages. Returns how many rows were indexed."
+  [space since]
+  (loop [after-id 0, total 0]
+    (let [rows (lucene.store/rows-after space after-id since page-size)]
+      (if (empty? rows)
+        total
+        (do
+          (lucene.index/upsert-rows! rows)
+          (recur (long (:id (last rows))) (long (+ total (count rows)))))))))
+
+(defn- reconcile-ids!
+  "Drop local documents the table no longer has, and import any it has that this node is missing."
+  [space]
+  (let [by-id   (into {}
+                      (map (fn [{:keys [model model_id]}]
+                             [(lucene.index/document-id model model_id) [model model_id]]))
+                      (lucene.store/model-ids space))
+        live    (lucene.index/live-ids)
+        stale   (remove by-id live)
+        missing (remove live (keys by-id))]
+    (when (seq stale)
+      (lucene.index/delete-ids! stale))
+    (when (seq missing)
+      (lucene.index/upsert-rows! (lucene.store/rows-for-model-ids space (map by-id missing))))
+    {:stale (count stale) :missing (count missing)}))
+
+(defn- since-bound
+  "The `updated_at` lower bound for the next import: the last watermark less the overlap window."
+  ^Timestamp [mx-ms]
+  (when mx-ms
+    (Timestamp. (- (long mx-ms) (* 1000 watermark-overlap-seconds)))))
+
+(defn sync-tick!
+  "Refresh this node's Lucene index from the embedding table once, returning what the tick did.
+
+  Returns nil when the Lucene backend is not selected or the semantic engine is not active — both are re-read
+  every tick, so a licence or engine change needs no restart."
+  []
+  (when (and (semantic.util/lucene-backend?)
+             (semantic.util/semantic-search-active?))
+    (lucene.index/ensure-open!)
+    (let [space          (lucene.store/space-id)
+          {:keys [n mx]} (lucene.store/space-stats space)
+          mx-ms          (watermark-ms mx)
+          previous       (or (when (= space (:space @state)) @state)
+                             (read-watermark space))
+          diff-due?      (or (nil? previous)
+                             (not= n (lucene.index/live-count))
+                             (< id-diff-interval-ms (u/since-ms (:diff-timer previous))))]
+      (if (and previous (= n (:n previous)) (= mx-ms (:mx-ms previous)) (not diff-due?))
+        {:space space :skipped true}
+        (let [indexed (import-rows! space (since-bound (:mx-ms previous)))
+              diff    (when diff-due? (reconcile-ids! space))]
+          (reset! state {:space      space
+                         :n          n
+                         :mx-ms      mx-ms
+                         :diff-timer (if diff-due? (u/start-timer) (:diff-timer previous))})
+          (write-watermark! space {:n n :mx-ms mx-ms})
+          (cond-> {:space space :indexed indexed}
+            diff (merge diff)))))))
+
+(defn- tick! []
+  (try
+    (sync-tick!)
+    (catch Throwable t
+      ;; scheduleWithFixedDelay stops rescheduling a task that throws, which would freeze this node's index.
+      (log/warnf "Semantic search Lucene sync failed, retrying next tick: %s" (ex-message t)))))
+
+(defn start!
+  "Start this node's Lucene sync timer. Idempotent; a no-op unless the Lucene backend is selected."
+  []
+  (when (semantic.util/lucene-backend?)
+    (let [new-executor (Executors/newSingleThreadScheduledExecutor
+                        (reify ThreadFactory
+                          (newThread [_ r]
+                            (doto (Thread. ^Runnable r "semantic-search-lucene-sync")
+                              (.setDaemon true)))))]
+      (if (compare-and-set! executor nil new-executor)
+        (do
+          (.scheduleWithFixedDelay new-executor
+                                   (reify Runnable (run [_] (tick!)))
+                                   tick-seconds tick-seconds TimeUnit/SECONDS)
+          (log/infof "Started the semantic search Lucene sync, every %ds" tick-seconds)
+          true)
+        (do
+          (.shutdown new-executor)
+          false)))))
+
+(defn stop!
+  "Stop this node's Lucene sync timer. Idempotent."
+  []
+  (when-let [^ScheduledExecutorService running (first (reset-vals! executor nil))]
+    (.shutdownNow running)
+    (log/info "Stopped the semantic search Lucene sync"))
+  (reset-state!))
+
+(defmethod startup/def-startup-logic! ::semantic-lucene-sync [_]
+  (start!))
+
+(defmethod startup/def-shutdown-logic! ::semantic-lucene-sync [_]
+  ;; Order matters, and `run-shutdown-logic!` does not order its methods: stop the timer before closing the
+  ;; index out from under it.
+  (stop!)
+  (lucene.index/close!))

enterprise/backend/src/metabase_enterprise/semantic_search/models/embedding.clj

@@ -0,0 +1,65 @@
+(ns metabase-enterprise.semantic-search.models.embedding
+  "`:model/SemanticSearchEmbedding` and the codec for the embedding vectors it stores in the app DB.
+
+  One row per search document per embedding space. The Lucene semantic-search backend treats this table as the
+  source of truth and each node's Lucene index as a disposable materialization of it."
+  (:require
+   [buddy.core.codecs :as buddy-codecs]
+   [buddy.core.hash :as buddy-hash]
+   [metabase.models.interface :as mi]
+   [methodical.core :as methodical]
+   [toucan2.core :as t2])
+  (:import
+   (java.nio ByteBuffer ByteOrder)
+   (java.nio.charset StandardCharsets)
+   (java.sql Blob)))
+
+(set! *warn-on-reflection* true)
+
+(def ^:const bytes-per-float
+  "Width of one encoded vector component. Vectors are float32, so a `dims`-dimensional vector is `4 * dims` bytes."
+  4)
+
+(defn floats->bytes
+  "Encode `floats` (anything `float-array` accepts) as little-endian float32 bytes."
+  ^bytes [floats]
+  (let [^floats fs (float-array floats)
+        buf        (.order (ByteBuffer/allocate (* bytes-per-float (alength fs))) ByteOrder/LITTLE_ENDIAN)]
+    (.put (.asFloatBuffer buf) fs)
+    (.array buf)))
+
+(defn bytes->floats
+  "Decode bytes written by [[floats->bytes]] back into a `float[]`.
+
+  Throws when `bs` is not a whole number of float32 components."
+  ^floats [^bytes bs]
+  (when-not (zero? (mod (alength bs) bytes-per-float))
+    (throw (ex-info "Embedding blob is not a whole number of float32 components"
+                    {:byte-count (alength bs)})))
+  (let [fb  (.asFloatBuffer (.order (ByteBuffer/wrap bs) ByteOrder/LITTLE_ENDIAN))
+        out (float-array (.remaining fb))]
+    (.get fb out)
+    out))
+
+(defn content-hash
+  "Return the hex SHA-256 of `text`, the key under which its embedding is cached and reused."
+  ^String [^String text]
+  (-> (.getBytes (or text "") StandardCharsets/UTF_8)
+      buddy-hash/sha256
+      buddy-codecs/bytes->hex))
+
+;;;; Model
+
+(derive :model/SemanticSearchEmbedding :metabase/model)
+
+(methodical/defmethod t2/table-name :model/SemanticSearchEmbedding [_model] :semantic_search_embedding)
+
+(defn- blob->bytes ^bytes [v]
+  ;; H2 hands back a java.sql.Blob where Postgres and MySQL hand back a byte array.
+  (if (instance? Blob v)
+    (let [^Blob b v] (.getBytes b 1 (int (.length b))))
+    v))
+
+(t2/deftransforms :model/SemanticSearchEmbedding
+  {:embedding {:in identity :out blob->bytes}
+   :document  mi/transform-json})

enterprise/backend/src/metabase_enterprise/semantic_search/schema.clj

@@ -18,3 +18,17 @@
    [:request_type {:optional true} [:maybe [:or :keyword :string]]]
    [:created_at   {:optional true} [:maybe ms/TemporalInstant]]
    [:total_tokens {:optional true} [:maybe :int]]])
+
+(mr/def ::semantic-search-embedding.insert
+  "What an insert into `:semantic_search_embedding` accepts: every column except the generated `id` and the
+  `created_at`/`updated_at` the app DB defaults."
+  [:map {:closed true}
+   [:embedding_space_id :string]
+   [:model              :string]
+   [:model_id           :string]
+   [:name               {:optional true} [:maybe :string]]
+   [:archived           :boolean]
+   [:content_hash       :string]
+   [:dims               ms/PositiveInt]
+   [:embedding          [:fn {:error/message "byte array"} bytes?]]
+   [:document           [:fn {:error/message "search document map"} map?]]])

enterprise/backend/src/metabase_enterprise/semantic_search/settings.clj

@@ -132,6 +132,18 @@
   :visibility :internal
   :doc false)
 
+(defsetting semantic-search-backend
+  (deferred-tru
+   (str "Where semantic search keeps its vectors: `lucene` (an app-DB table materialized into a per-node "
+        "Apache Lucene index, which works on every app-DB type) or `pgvector` (a Postgres store, either "
+        "MB_PGVECTOR_DB_URL or a `semantic_search` schema in a Postgres app DB)."))
+  :type :keyword
+  :default :lucene
+  :encryption :no
+  :export? false
+  :visibility :internal
+  :doc false)
+
 (defsetting semantic-search-results-limit
   (deferred-tru "Maximum number of results to return from a single semantic search query.")
   :type :integer

enterprise/backend/src/metabase_enterprise/semantic_search/task/index_repair.clj

@@ -26,15 +26,19 @@
     (lib-be/with-metadata-provider-cache
       (let [{:keys [index-id orphans snapshot-at]}
             (semantic-search.core/repair-index! (search.ingestion/searchable-documents))]
-        (semantic.health/report-repair-metrics! index-id {:orphan-count orphans
-                                                          :snapshot-at  snapshot-at})))
+        ;; The repair metrics read pgvector's index metadata, which the Lucene backend has none of.
+        (when-not (semantic.u/lucene-backend?)
+          (semantic.health/report-repair-metrics! index-id {:orphan-count orphans
+                                                            :snapshot-at  snapshot-at}))))
     (log/info "Completed semantic search index repair")
     (catch InterruptedException e
-      (semantic.health/report-repair-metrics! nil)
+      (when-not (semantic.u/lucene-backend?)
+        (semantic.health/report-repair-metrics! nil))
       (throw e))
     (catch Exception e
       ;; Invalidate the shared snapshot when the producing job fails.
-      (semantic.health/report-repair-metrics! nil)
+      (when-not (semantic.u/lucene-backend?)
+        (semantic.health/report-repair-metrics! nil))
       (log/errorf "Failed to complete semantic search index repair: %s" (ex-message e)))))
 
 (task/defjob ^{DisallowConcurrentExecution true
@@ -45,7 +49,8 @@
       (repair-index!))))
 
 (defmethod task/init! ::SemanticIndexRepair [_]
-  (when (semantic.u/semantic-search-configured?)
+  ;; Available, not configured: repair is the backfill for both backends, so it must schedule under Lucene too.
+  (when (semantic.u/semantic-search-available?)
     (let [job (jobs/build
                (jobs/of-type SemanticIndexRepair)
                (jobs/with-identity repair-job-key))

enterprise/backend/src/metabase_enterprise/semantic_search/task/metric_collector.clj

@@ -70,7 +70,8 @@
   (try
     ;; Active, not merely available: on an available-but-inactive instance the index tables never exist,
     ;; and the collectors would warn about them on every run.
-    (when (semantic.u/semantic-search-active?)
+    (when (and (not (semantic.u/lucene-backend?))
+               (semantic.u/semantic-search-active?))
       (let [pgvector (semantic.env/get-pgvector-datasource!)
             index-metadata (semantic.env/get-index-metadata)]
         (collect-gate-size! pgvector)

enterprise/backend/src/metabase_enterprise/semantic_search/task/usage_trimmer.clj

@@ -33,7 +33,8 @@
 
 (defmethod task/init! ::SemanticSearchUsageTrimmer
   [_]
-  (when (semantic.u/semantic-search-configured?)
+  ;; Available, not configured: the token-tracking table is in the app DB, so trimming applies to both backends.
+  (when (semantic.u/semantic-search-available?)
     (let [job (jobs/build
                (jobs/of-type SemanticSearchUsageTrimmer)
                (jobs/with-identity trimmer-job-key))

enterprise/backend/src/metabase_enterprise/semantic_search/util.clj

@@ -2,6 +2,7 @@
   (:require
    [clojure.string :as str]
    [metabase-enterprise.semantic-search.db.datasource :as semantic.db.datasource]
+   [metabase-enterprise.semantic-search.settings :as semantic.settings]
    [metabase.app-db.core :as mdb]
    [metabase.premium-features.core :as premium-features]
    [metabase.search.engine :as search.engine]
@@ -109,28 +110,39 @@
   [pgvector index-name]
   (contains? #{nil :invalid} (index-state pgvector index-name)))
 
+(defn lucene-backend?
+  "Whether semantic search stores its vectors in the app DB plus a per-node Lucene index, rather than pgvector.
+  Everything pgvector-only -- its Quartz jobs, health checks and store probes -- must stay off when this is true."
+  []
+  (= :lucene (semantic.settings/semantic-search-backend)))
+
 (defn semantic-search-configured?
-  "Whether to schedule the semantic-search Quartz jobs at startup.
-  True when the `:semantic-search` feature is present and a pgvector store might exist: a dedicated
-  MB_PGVECTOR_DB_URL, or a Postgres app DB that [[semantic-search-available?]] can probe to answer for sure.
-  Cheap and infallible by contract -- it runs at boot and never queries the DB."
+  "Whether to schedule the pgvector semantic-search Quartz jobs at startup.
+  True when the `:semantic-search` feature is present, the pgvector backend is selected, and a pgvector store
+  might exist: a dedicated MB_PGVECTOR_DB_URL, or a Postgres app DB that [[semantic-search-available?]] can
+  probe to answer for sure.
+  Cheap and infallible by contract -- it runs at boot, and the only read it makes is of the
+  `semantic-search-backend` setting."
   []
   ;; The license is in this boot gate, not only the per-execution gates, so an unlicensed instance's
   ;; scheduler stays free of no-op jobs. The asymmetry is deliberate: removing the feature at runtime lets
   ;; the scheduled jobs no-op via semantic-search-active?, but adding it needs a restart before they
   ;; schedule. Engine activity stays per-execution so it never needs one.
   (and (premium-features/has-feature? :semantic-search)
+       (not (lucene-backend?))
        (or (semantic.db.datasource/dedicated-url-configured?)
            (= :postgres (mdb/db-type)))))
 
 (defn semantic-search-available?
-  "Does this instance have the infrastructure for semantic search: the premium feature and a pgvector DB.
+  "Does this instance have the infrastructure for semantic search: the premium feature and somewhere to keep
+  vectors -- the app DB under the Lucene backend, a pgvector DB otherwise.
   Engine selection and hygiene tasks key off this."
   []
   ;; Feature first: the pgvector check may probe the app DB, and instances that can't use the answer
   ;; must never probe.
   (and (premium-features/has-feature? :semantic-search)
-       (semantic.db.datasource/pgvector-configured?)))
+       (or (lucene-backend?)
+           (semantic.db.datasource/pgvector-configured?))))
 
 (defn semantic-search-active?
   "Is the semantic index being maintained on this instance?

enterprise/backend/src/metabase_enterprise/serialization/v2/models.clj

@@ -152,6 +152,7 @@
    "SearchIndexMetadata"
    "Secret"
    "SecurityAdvisory"
+   "SemanticSearchEmbedding"
    "SemanticSearchTokenTracking"
    "Session"
    "SourceDimensionDaily"

enterprise/backend/test/metabase_enterprise/semantic_search/health_test.clj

@@ -53,7 +53,9 @@
   each test only has to override the signal it cares about."
   [thunk]
   (mt/with-dynamic-fn-redefs
-    [semantic.util/semantic-search-active? (constantly true)
+    ;; These probes are pgvector's; the instance default backend is :lucene, which turns them off.
+    [semantic.util/lucene-backend?         (constantly false)
+     semantic.util/semantic-search-active? (constantly true)
      semantic.env/get-pgvector-datasource! (constantly ::pgvector)
      semantic.env/get-index-metadata       (constantly ::index-metadata)
      semantic.env/get-configured-embedding-model (constantly embedding-model)

enterprise/backend/test/metabase_enterprise/semantic_search/lucene/core_test.clj

@@ -0,0 +1,82 @@
+(ns metabase-enterprise.semantic-search.lucene.core-test
+  (:require
+   [clojure.test :refer :all]
+   [metabase-enterprise.semantic-search.lucene.core :as lucene.core]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.lucene.store :as lucene.store]
+   [metabase-enterprise.semantic-search.lucene.test-util :as lucene.tu]
+   [metabase-enterprise.semantic-search.models.embedding :as semantic.models.embedding]
+   [toucan2.core :as t2]))
+
+(set! *warn-on-reflection* true)
+
+(defn- updated-at [model-id]
+  (:updated_at (t2/select-one :model/SemanticSearchEmbedding :model_id model-id)))
+
+(deftest update-reports-an-empty-stream-test
+  (lucene.tu/with-lucene-store [4]
+    (testing "an empty document stream reports no writes rather than nil"
+      (is (= {} (lucene.core/update! []))))))
+
+(deftest update-indexes-what-it-persists-test
+  (lucene.tu/with-lucene-store [4]
+    (is (= {"card" 2} (lucene.core/update! [(lucene.tu/document "card" 1)
+                                            (lucene.tu/document "card" 2)])))
+    (is (= #{"card_1" "card_2"} (lucene.index/live-ids)))
+    (testing "deleting removes the row and the local document together"
+      (is (= {"card" 1} (lucene.core/delete! "card" [2])))
+      (is (= #{"card_1"} (lucene.index/live-ids)))
+      (is (= 1 (t2/count :model/SemanticSearchEmbedding))))))
+
+(deftest init-populates-once-test
+  (lucene.tu/with-lucene-store [4]
+    (is (= {"card" 1} (lucene.core/init! [(lucene.tu/document "card" 1)] {})))
+    (testing "a second init! leaves the corpus alone"
+      (lucene.tu/reset-embedded-texts!)
+      (is (= {} (lucene.core/init! [(lucene.tu/document "card" 1)] {})))
+      (is (= [] (lucene.tu/embedded-texts))))
+    (testing "force-reset? wipes and repopulates"
+      (is (= {"card" 1} (lucene.core/init! [(lucene.tu/document "card" 1)] {:force-reset? true})))
+      (is (= 1 (t2/count :model/SemanticSearchEmbedding))))))
+
+(deftest repair-only-fills-gaps-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.core/update! [(lucene.tu/document "card" 1)])
+    (let [before (updated-at "1")]
+      (lucene.tu/reset-embedded-texts!)
+      (testing "a document the table is missing is backfilled"
+        (is (= 0 (:orphans (lucene.core/repair! [(lucene.tu/document "card" 1)
+                                                 (lucene.tu/document "card" 2)]))))
+        (is (= 2 (t2/count :model/SemanticSearchEmbedding)))
+        (is (= ["[card]\nname: doc 2"] (lucene.tu/embedded-texts))
+            "only the missing document reaches the embedder"))
+      (testing "a document the table already has is left untouched"
+        ;; Rewriting it would move its updated_at and make every node re-import the whole space next tick.
+        (is (= before (updated-at "1")))))))
+
+(deftest repair-drops-what-no-longer-exists-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.core/update! [(lucene.tu/document "card" 1)
+                          (lucene.tu/document "card" 2)])
+    (testing "a document that has left the corpus is dropped from the table and the index"
+      (is (= 1 (:orphans (lucene.core/repair! [(lucene.tu/document "card" 1)]))))
+      (is (= #{"card_1"} (lucene.index/live-ids)))
+      (is (= 1 (t2/count :model/SemanticSearchEmbedding))))))
+
+(deftest repair-reclaims-abandoned-embedding-spaces-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.core/update! [(lucene.tu/document "card" 1)])
+    (t2/insert! :model/SemanticSearchEmbedding
+                {:embedding_space_id "emb:v1:sha256:a-model-we-no-longer-use"
+                 :model              "card"
+                 :model_id           "99"
+                 :archived           false
+                 :content_hash       "deadbeef"
+                 :dims               4
+                 :embedding          (semantic.models.embedding/floats->bytes [1.0 0.0 0.0 0.0])
+                 :document           {:model "card" :id 99}})
+    (is (= 2 (t2/count :model/SemanticSearchEmbedding)))
+    (testing "rows left behind by an embedding-model change are reclaimed, since nothing else collects them"
+      (lucene.core/repair! [(lucene.tu/document "card" 1)])
+      (is (= [(lucene.store/space-id)]
+             (map :embedding_space_id (t2/select :model/SemanticSearchEmbedding)))))))

enterprise/backend/test/metabase_enterprise/semantic_search/lucene/index_test.clj

@@ -0,0 +1,162 @@
+(ns metabase-enterprise.semantic-search.lucene.index-test
+  (:require
+   [clojure.test :refer :all]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.models.embedding :as semantic.models.embedding]
+   [metabase.test :as mt]
+   [metabase.util.json :as json])
+  (:import
+   (org.apache.lucene.document Document)
+   (org.apache.lucene.index Term)
+   (org.apache.lucene.search IndexSearcher KnnFloatVectorQuery ScoreDoc TermQuery TopDocs)))
+
+(set! *warn-on-reflection* true)
+
+(defmacro ^:private with-index
+  "Open a Lucene index of width `dims` under a fresh temp directory for the body."
+  [[dims] & body]
+  `(mt/with-temp-dir [dir# nil]
+     (binding [lucene.index/*index-root* (str dir#)]
+       (try
+         (lucene.index/ensure-open! "test-space" ~dims)
+         ~@body
+         (finally
+           (lucene.index/close!))))))
+
+(defn- row
+  "A `semantic_search_embedding` row for `model`/`id` carrying `embedding`, merged with extra `document` keys."
+  [model id embedding & [document]]
+  {:model     model
+   :model_id  (str id)
+   :embedding (semantic.models.embedding/floats->bytes embedding)
+   :document  (merge {:id           id
+                      :model        model
+                      :legacy_input (json/encode {:id id :model model :name (str model " " id)})}
+                     document)})
+
+(defn- knn-hits
+  "Run a kNN search for `embedding` and return `[{:id … :score … :legacy_input …}]`, nearest first."
+  [embedding k]
+  (lucene.index/with-searcher [^IndexSearcher searcher]
+    (let [query           (KnnFloatVectorQuery. "embedding" (float-array embedding) k)
+          ^TopDocs hits   (.search searcher query (int k))
+          stored          (.storedFields searcher)]
+      (vec (for [^ScoreDoc hit (.scoreDocs hits)
+                 :let [^Document doc (.document stored (.-doc hit))]]
+             {:id           (.get doc "id")
+              :model        (.get doc "model")
+              :model_id     (.get doc "model_id")
+              :legacy_input (.get doc "legacy_input")
+              :score        (.-score hit)})))))
+
+(deftest floats-round-trip-test
+  (testing "a 4-dimensional vector survives the little-endian float32 codec"
+    (let [v [0.25 -0.5 0.0 1.0]]
+      (is (= v (vec (semantic.models.embedding/bytes->floats
+                     (semantic.models.embedding/floats->bytes v)))))))
+  (testing "a 1024-dimensional vector survives, at 4 bytes per component"
+    (let [v     (mapv #(float (/ % 1024.0)) (range 1024))
+          bytes (semantic.models.embedding/floats->bytes v)]
+      (is (= 4096 (alength bytes)))
+      (is (= v (vec (semantic.models.embedding/bytes->floats bytes))))))
+  (testing "a blob that is not a whole number of components is rejected"
+    (is (thrown-with-msg? clojure.lang.ExceptionInfo #"whole number of float32"
+                          (semantic.models.embedding/bytes->floats (byte-array 3))))))
+
+(deftest document-id-test
+  (is (= "card_12" (lucene.index/document-id "card" 12)))
+  (is (= "card_12" (lucene.index/document-id "card" "12"))))
+
+(deftest upsert-and-knn-round-trip-test
+  (with-index [4]
+    (testing "upserting reports how many documents it wrote and makes them visible"
+      (is (= 2 (lucene.index/upsert-rows! [(row "card" 1 [1.0 0.0 0.0 0.0])
+                                           (row "dashboard" 2 [0.0 1.0 0.0 0.0])])))
+      (is (= 2 (lucene.index/live-count)))
+      (is (= #{"card_1" "dashboard_2"} (lucene.index/live-ids))))
+    (testing "kNN returns the nearest document first, with its stored fields"
+      (let [[nearest :as hits] (knn-hits [0.9 0.1 0.0 0.0] 2)]
+        (is (= 2 (count hits)))
+        (is (= "card_1" (:id nearest)))
+        (is (= "card" (:model nearest)))
+        (is (= "1" (:model_id nearest)))
+        (is (= {:id 1 :model "card" :name "card 1"}
+               (json/decode+kw (:legacy_input nearest))))
+        (testing "cosine scores are in [0 1] and rank the nearer document higher"
+          (is (< 0.0 (:score (second hits)) (:score nearest) 1.0001)))))
+    (testing "upserting the same id again replaces the document rather than duplicating it"
+      (is (= 1 (lucene.index/upsert-rows! [(row "card" 1 [0.0 0.0 1.0 0.0] {:legacy_input "{\"id\":1}"})])))
+      (is (= 2 (lucene.index/live-count)))
+      (is (= "{\"id\":1}" (:legacy_input (first (knn-hits [0.0 0.0 1.0 0.0] 1))))))))
+
+(deftest delete-test
+  (with-index [4]
+    (lucene.index/upsert-rows! [(row "card" 1 [1.0 0.0 0.0 0.0])
+                                (row "card" 2 [0.0 1.0 0.0 0.0])
+                                (row "card" 3 [0.0 0.0 1.0 0.0])])
+    (testing "deleting by document id removes exactly those documents"
+      (is (= 1 (lucene.index/delete-ids! ["card_2"])))
+      (is (= #{"card_1" "card_3"} (lucene.index/live-ids))))
+    (testing "deleting an id that is not indexed is harmless"
+      (is (= 1 (lucene.index/delete-ids! ["card_999"])))
+      (is (= 2 (lucene.index/live-count))))
+    (testing "delete-all! empties the index"
+      (lucene.index/delete-all!)
+      (is (= 0 (lucene.index/live-count)))
+      (is (= #{} (lucene.index/live-ids))))))
+
+(deftest indexed-filter-fields-test
+  (with-index [4]
+    (lucene.index/upsert-rows! [(row "card" 1 [1.0 0.0 0.0 0.0]
+                                     {:archived false :collection_id 7 :personal_owner_id nil})
+                                (row "card" 2 [0.0 1.0 0.0 0.0]
+                                     {:archived true :collection_id 8 :personal_owner_id 3})])
+    (lucene.index/with-searcher [^IndexSearcher searcher]
+      (testing "booleans index as their string form"
+        (is (= 1 (.count searcher (TermQuery. (Term. "archived" "true")))))
+        (is (= 1 (.count searcher (TermQuery. (Term. "archived" "false"))))))
+      (testing "ids index as decimal strings"
+        (is (= 1 (.count searcher (TermQuery. (Term. "collection_id" "7"))))))
+      (testing "a missing personal owner indexes as the null sentinel"
+        (is (= 1 (.count searcher (TermQuery. (Term. "personal_owner_id" lucene.index/null-owner)))))
+        (is (= 1 (.count searcher (TermQuery. (Term. "personal_owner_id" "3")))))))))
+
+(deftest zero-vector-is-not-searchable-test
+  (with-index [4]
+    (testing "an all-zero embedding is indexed without its vector field, since cosine cannot score it"
+      (is (= 1 (lucene.index/upsert-rows! [(row "card" 1 [0.0 0.0 0.0 0.0])])))
+      (is (= #{"card_1"} (lucene.index/live-ids)))
+      (is (= [] (knn-hits [1.0 0.0 0.0 0.0] 5))))))
+
+(deftest reopen-on-space-change-test
+  (mt/with-temp-dir [dir nil]
+    (binding [lucene.index/*index-root* (str dir)]
+      (try
+        (let [four (lucene.index/ensure-open! "space-4dim" 4)]
+          (lucene.index/upsert-rows! [(row "card" 1 [1.0 0.0 0.0 0.0])])
+          (is (= 1 (lucene.index/live-count)))
+          (testing "reopening the same space keeps the same index"
+            (is (= (:path four) (:path (lucene.index/ensure-open! "space-4dim" 4))))
+            (is (= 1 (lucene.index/live-count))))
+          (testing "a different embedding space opens an empty index in its own directory"
+            (let [eight (lucene.index/ensure-open! "space-8dim" 8)]
+              (is (not= (:path four) (:path eight)))
+              (is (= 0 (lucene.index/live-count))))))
+        (finally
+          (lucene.index/close!))))))
+
+(deftest too-many-dimensions-test
+  (mt/with-temp-dir [dir nil]
+    (binding [lucene.index/*index-root* (str dir)]
+      (testing "a model wider than Lucene's default codec allows is refused up front"
+        (is (thrown-with-msg? clojure.lang.ExceptionInfo #"too wide"
+                              (lucene.index/ensure-open! "space-too-wide" (inc lucene.index/max-dimensions))))))))
+
+(deftest space-id-is-sanitized-into-a-directory-name-test
+  (mt/with-temp-dir [dir nil]
+    (binding [lucene.index/*index-root* (str dir)]
+      (try
+        (let [{:keys [path]} (lucene.index/ensure-open! "emb:v1:sha256:abc" 4)]
+          (is (= "emb_v1_sha256_abc" (str (.getFileName ^java.nio.file.Path path)))))
+        (finally
+          (lucene.index/close!))))))

enterprise/backend/test/metabase_enterprise/semantic_search/lucene/query_test.clj

@@ -0,0 +1,284 @@
+(ns metabase-enterprise.semantic-search.lucene.query-test
+  (:require
+   [clojure.string :as str]
+   [clojure.test :refer :all]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.lucene.query :as lucene.query]
+   [metabase-enterprise.semantic-search.lucene.test-util :as lucene.tu]
+   [metabase.search.config :as search.config]
+   [metabase.search.core :as search]
+   [metabase.search.engine :as search.engine]
+   [metabase.search.test-util :as search.tu]
+   [metabase.test :as mt])
+  (:import
+   (java.time OffsetDateTime)))
+
+(set! *warn-on-reflection* true)
+
+;;;; Fusion
+
+(def ^:private test-weights {:rrf 500 :semantic-distance 10})
+
+(defn- score-named [result nm]
+  (first (filter (comp #{nm} :name) (:all-scores result))))
+
+(defn- vector-row [model id rank score]
+  {:legacy_input   {:model model :id id :name (str model " " id)}
+   :semantic-rank  rank
+   :semantic-score score})
+
+(defn- keyword-row [model id]
+  {:model model :id id :name (str model " " id) :score 7.0 :all-scores [{:name :text}]})
+
+(deftest fuse-combines-both-ranks-test
+  (let [[result] (lucene.query/fuse test-weights
+                                    [(vector-row "card" 1 1 0.9)]
+                                    [(keyword-row "card" 1)])]
+    (testing "a document both arms rank first gets both reciprocal-rank contributions"
+      (is (= (+ (/ 0.49 61) (/ 0.51 61)) (:score (score-named result :rrf)))))
+    (testing "its vector score rides along as the semantic-distance score"
+      (is (= 0.9 (:score (score-named result :semantic-distance)))))
+    (testing "the total is the weighted sum of the two"
+      (is (= (+ (* 500 (+ (/ 0.49 61) (/ 0.51 61))) (* 10 0.9))
+             (:score result))))
+    (testing "the keyword arm's own total score is not folded in"
+      (is (not= 7.0 (:score result)))
+      (is (nil? (score-named result :text))))))
+
+(deftest fuse-scores-single-arm-hits-test
+  (testing "a keyword-only hit at rank 2 gets only the keyword contribution and no semantic distance"
+    (let [[result] (lucene.query/fuse test-weights [] [(keyword-row "card" 1) (keyword-row "card" 2)])
+          second-result (second (lucene.query/fuse test-weights [] [(keyword-row "card" 1) (keyword-row "card" 2)]))]
+      (is (= (/ 0.51 61) (:score (score-named result :rrf))))
+      (is (= (/ 0.51 62) (:score (score-named second-result :rrf))))
+      (is (= 0.0 (:score (score-named result :semantic-distance))))))
+  (testing "a vector-only hit gets only the semantic contribution"
+    (let [[result] (lucene.query/fuse test-weights [(vector-row "card" 9 1 0.8)] [])]
+      (is (= (/ 0.49 61) (:score (score-named result :rrf))))
+      (is (= 0.8 (:score (score-named result :semantic-distance)))))))
+
+(deftest fuse-ranks-agreed-documents-above-one-arm-hits-test
+  (let [results (lucene.query/fuse test-weights
+                                   [(vector-row "card" 1 2 0.7)
+                                    (vector-row "card" 2 1 0.7)]
+                                   [(keyword-row "card" 1)
+                                    (keyword-row "card" 3)])]
+    (is (= [["card" 1] ["card" 2] ["card" 3]]
+           (map (juxt :model :id) results))
+        "card 1 places first on both arms agreeing, ahead of each arm's own top hit")))
+
+(deftest fuse-prefers-the-keyword-arm-body-test
+  (testing "when both arms find a document, its row comes from the keyword arm, which carries the bookmark flag"
+    (let [[result] (lucene.query/fuse test-weights
+                                      [(vector-row "card" 1 1 0.9)]
+                                      [(assoc (keyword-row "card" 1) :bookmark true)])]
+      (is (true? (:bookmark result))))))
+
+;;;; Filters
+
+(defn- filter-string [search-context]
+  (some-> (lucene.query/filter-query search-context) str))
+
+(deftest filter-query-is-nil-when-nothing-is-filtered-test
+  (is (nil? (filter-string {}))))
+
+(deftest filter-query-per-context-key-test
+  (are [expected ctx] (= expected (filter-string ctx))
+    "#archived:true"                  {:archived? true}
+    "#archived:false"                 {:archived? false}
+    "#verified:true"                  {:verified true}
+    "#curated:true"                   {:curated? true}
+    "#creator_id:(12)"                {:created-by #{12}}
+    "#last_editor_id:(3)"             {:last-edited-by #{3}}
+    "#database_id:7"                  {:table-db-id 7}
+    "#model_id:(42)"                  {:ids [42]}
+    "#display_type:(table)"           {:display-type ["table"]}))
+
+(deftest filter-query-models-test
+  (testing "a non-empty model set filters to those models"
+    (is (= "#model:(card)" (filter-string {:models #{"card"}}))))
+  (testing "an empty but present model set matches nothing, rather than everything"
+    (is (= "#MatchNoDocsQuery(\"\")" (filter-string {:models #{}})))))
+
+(deftest filter-query-personal-collection-modes-test
+  (are [expected mode] (= expected (filter-string {:filter-items-in-personal-collection mode
+                                                   :current-user-id 5}))
+    nil                                                     "all"
+    nil                                                     nil
+    "#personal_owner_id:5"                                  "only-mine"
+    "#(#*:* -personal_owner_id:__null__)"                   "only"
+    "#personal_owner_id:__null__"                           "exclude"
+    "#((personal_owner_id:__null__ personal_owner_id:5)~1)" "exclude-others"))
+
+;;;; The keyword arm
+
+(deftest keyword-arm-distinguishes-empty-from-not-run-test
+  (testing "an arm that ran and matched nothing is empty, so the caller need not search again"
+    (with-redefs [search.engine/supported-engine? (constantly true)
+                  search.engine/results           (constantly [])]
+      (is (= [] (#'lucene.query/keyword-hits {:search-string "puppy"} 10)))))
+  (testing "an app DB the appdb engine cannot serve leaves the arm nil, so the caller still falls back"
+    (with-redefs [search.engine/supported-engine? (constantly false)]
+      (is (nil? (#'lucene.query/keyword-hits {:search-string "puppy"} 10)))))
+  (testing "an arm that threw is nil for the same reason"
+    (with-redefs [search.engine/supported-engine? (constantly true)
+                  search.engine/results           (fn [_] (throw (ex-info "no index" {})))]
+      (is (nil? (#'lucene.query/keyword-hits {:search-string "puppy"} 10))))))
+
+;;;; End to end, through the search API
+
+(def ^:private dog-card "Dog Training Guide")
+(def ^:private revenue-card "Quarterly Revenue")
+
+(defn- embed-by-topic
+  "Put dog-ish text on one axis and revenue-ish text on another, so `puppy` is near one card and far from the other."
+  [text]
+  (cond
+    (str/includes? text dog-card)     [1.0 0.0 0.0 0.0]
+    (str/includes? text revenue-card) [0.0 0.0 1.0 0.0]
+    (str/includes? text "puppy")      [0.98 0.2 0.0 0.0]
+    :else                             [0.0 1.0 0.0 0.0]))
+
+(defmacro ^:private with-semantic-search
+  "Run `body` on an instance where the Lucene semantic engine serves search, over a temporary appdb keyword index."
+  [& body]
+  `(mt/with-premium-features #{:semantic-search}
+     (lucene.tu/with-lucene-store [4]
+       (binding [lucene.tu/*embed-fn* embed-by-topic]
+         (search.tu/with-temp-index-table
+           ~@body)))))
+
+(defn- search-names
+  "Names of the results `user` gets for `q`, via the search API."
+  [user q]
+  (->> (mt/user-http-request user :get 200 "search" :q q)
+       :data
+       (map :name)))
+
+(deftest semantic-engine-serves-search-test
+  (with-semantic-search
+    (is (= :search.engine/semantic (search.engine/default-engine))
+        "a licence plus a ready embedder is all the Lucene backend needs to take over search")))
+
+(deftest semantic-beats-a-keyword-miss-test
+  (with-semantic-search
+    (mt/with-temp [:model/Card _ {:name dog-card}
+                   :model/Card _ {:name revenue-card}]
+      (is (pos? (lucene.index/live-count)) "ingestion wrote both cards to this node's Lucene index")
+      (testing "a concept query with no keyword overlap still finds the card it is about"
+        (let [names (search-names :crowberto "puppy")]
+          (is (contains? (set names) dog-card))
+          (testing "and the card it is not about stays below the cosine floor"
+            (is (not (contains? (set names) revenue-card)))))))))
+
+(deftest semantic-results-carry-the-fusion-scores-test
+  (with-semantic-search
+    (mt/with-temp [:model/Card _ {:name dog-card}]
+      (let [result (->> (search.tu/search-results "puppy" {:current-user-id (mt/user->id :crowberto)})
+                        (filter (comp #{dog-card} :name))
+                        first)]
+        (is (some? result))
+        (is (= #{:rrf :semantic-distance :bookmarked :user-recency}
+               (set (map :name (:all-scores result))))
+            "the two fusion scorers rank the result, and the appdb scorers are layered on afterwards")))))
+
+(defn- search-context
+  "A full search context for `search-string`, as the API builds one for `:crowberto`."
+  [search-string]
+  (search/search-context {:current-user-id    (mt/user->id :crowberto)
+                          :current-user-perms #{"/"}
+                          :is-superuser?      true
+                          :archived           false
+                          :context            :default
+                          :search-string      search-string
+                          :models             search.config/all-models
+                          :model-ancestors?   false}))
+
+(deftest query-passes-a-missing-keyword-arm-through-test
+  (with-semantic-search
+    (mt/with-temp [:model/Card _ {:name dog-card}]
+      (mt/with-test-user :crowberto
+        (testing "an app DB without an appdb index reports no keyword results at all, not an empty set"
+          ;; Built outside the redef: search-context resolves engines, which reads the multimethod's methods.
+          (let [ctx (search-context "puppy")]
+            (with-redefs [search.engine/supported-engine? (constantly false)]
+              (is (nil? (:keyword-results (lucene.query/query ctx)))))))
+        (testing "otherwise the rows it already fetched come back for the caller to reuse"
+          (is (some? (:keyword-results (lucene.query/query (search-context "Dog"))))))))))
+
+(defn- result-named
+  "The result called `nm` that `:crowberto` gets for `q`, straight from the engine."
+  [q nm]
+  (->> (search.tu/search-results q {:current-user-id (mt/user->id :crowberto)})
+       (filter (comp #{nm} :name))
+       first))
+
+(deftest both-arms-rebuild-timestamps-the-same-way-test
+  (with-semantic-search
+    (mt/with-temp [:model/Card _ {:name dog-card}]
+      ;; "puppy" has no keyword overlap, so that row is rebuilt by the vector arm alone; "Dog" matches the
+      ;; keyword arm, whose body the fusion prefers. Both come out of the same JSON `legacy_input`, where every
+      ;; timestamp is a string -- and `POST /api/agent/v1/search` rejects the strings.
+      (let [vector-only (result-named "puppy" dog-card)
+            keyword-hit (result-named "Dog" dog-card)]
+        (is (some? vector-only))
+        (is (some? keyword-hit))
+        (mt/with-test-user :crowberto
+          (is (empty? (:keyword-results (lucene.query/query (search-context "puppy"))))
+              "the keyword arm matches nothing here, so that row really is the vector arm's own"))
+        (testing "a hit only the vector arm found still carries real java.time timestamps"
+          (is (instance? OffsetDateTime (:updated_at vector-only)))
+          (is (not (string? (:created_at vector-only)))))
+        (testing "and both arms agree on the type of every timestamp they rebuild"
+          (are [k] (= (class (k keyword-hit)) (class (k vector-only)))
+            :created_at
+            :updated_at
+            :last_edited_at))))))
+
+(deftest blank-search-string-falls-back-to-the-keyword-arm-test
+  (with-semantic-search
+    (mt/with-temp [:model/Card _ {:name dog-card}]
+      (testing "the native-editor table picker sends no search string; semantic yields to the appdb engine"
+        (is (= {:results [] :raw-count 0} (lucene.query/query {:search-string ""})))
+        (is (contains? (set (search-names :crowberto "Dog")) dog-card))))))
+
+(deftest permissions-are-applied-to-fused-results-test
+  (with-semantic-search
+    (mt/with-non-admin-groups-no-root-collection-perms
+      (mt/with-temp [:model/Collection collection {:name "Private Kennel"}
+                     :model/Card _ {:name dog-card :collection_id (:id collection)}]
+        (testing "an admin sees the card"
+          (is (contains? (set (search-names :crowberto "puppy")) dog-card)))
+        (testing "a user without collection permissions does not"
+          (is (not (contains? (set (search-names :rasta "puppy")) dog-card))))))))
+
+(deftest an-unavailable-embedder-falls-back-to-the-keyword-arm-test
+  (with-semantic-search
+    (mt/with-temp [:model/Card _ {:name dog-card}]
+      (binding [lucene.tu/*embedder-fails?* true]
+        (testing "the query cannot be embedded, so search answers from the keyword arm instead of failing"
+          (is (contains? (set (search-names :crowberto "Dog")) dog-card)))))))
+
+(deftest diagnose-reports-the-stage-that-dropped-a-document-test
+  (with-semantic-search
+    (mt/with-temp [:model/Card card {:name dog-card}]
+      (mt/with-test-user :crowberto
+        (testing "a document that is not indexed at all"
+          (is (= :missing-from-index
+                 (:type (lucene.query/diagnose {:search-string "puppy"} "card" 999999)))))
+        (testing "a document a filter excludes"
+          (is (= {:type :filtered :details {:excluded-by :models}}
+                 (lucene.query/diagnose {:search-string "puppy" :models #{"dashboard"}} "card" (:id card)))))
+        (testing "a document that survives every stage"
+          (is (= :candidate
+                 (:type (lucene.query/diagnose {:search-string "puppy"} "card" (:id card))))))))))
+
+(deftest diagnose-reports-permissions-before-filters-test
+  (with-semantic-search
+    (mt/with-non-admin-groups-no-root-collection-perms
+      (mt/with-temp [:model/Collection collection {:name "Private Kennel"}
+                     :model/Card card {:name dog-card :collection_id (:id collection)}]
+        (testing "a document the current user may not read reads as permission-filtered, not merely filtered"
+          (mt/with-test-user :rasta
+            (is (= {:type :filtered :details {:excluded-by :permissions}}
+                   (lucene.query/diagnose {:search-string "puppy" :models #{"dashboard"}} "card" (:id card))))))))))

enterprise/backend/test/metabase_enterprise/semantic_search/lucene/store_test.clj

@@ -0,0 +1,186 @@
+(ns metabase-enterprise.semantic-search.lucene.store-test
+  (:require
+   [clojure.test :refer :all]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.lucene.store :as lucene.store]
+   [metabase-enterprise.semantic-search.lucene.test-util :as lucene.tu]
+   [metabase-enterprise.semantic-search.models.embedding :as semantic.models.embedding]
+   [metabase.test :as mt]
+   [toucan2.core :as t2])
+  (:import
+   (org.apache.lucene.document LongPoint)
+   (org.apache.lucene.index Term)
+   (org.apache.lucene.search IndexSearcher TermQuery)))
+
+(set! *warn-on-reflection* true)
+
+(defn- term-count
+  "How many indexed documents carry `value` in `field`."
+  [^IndexSearcher searcher ^String field ^String value]
+  (.count searcher (TermQuery. (Term. field value))))
+
+(defn- stored-rows []
+  (t2/select :model/SemanticSearchEmbedding {:order-by [[:model :asc] [:model_id :asc]]}))
+
+(deftest upsert-documents-writes-one-row-per-document-test
+  (lucene.tu/with-lucene-store [4]
+    (let [{:keys [rows report]} (lucene.store/upsert-documents!
+                                 [(lucene.tu/document "card" 1)
+                                  (lucene.tu/document "card" 2)
+                                  (lucene.tu/document "dashboard" 3)])]
+      (is (= {"card" 2 "dashboard" 1} report))
+      (is (= 3 (count rows)))
+      (let [[card-1 :as stored] (stored-rows)]
+        (is (= 3 (count stored)))
+        (is (= "card" (:model card-1)))
+        (is (= "1" (:model_id card-1)))
+        (is (= (lucene.store/space-id) (:embedding_space_id card-1)))
+        (is (= 4 (:dims card-1)))
+        (testing "the stored document round-trips as JSON, with the resolved personal owner added"
+          (is (= "card" (get-in card-1 [:document :model])))
+          (is (contains? (:document card-1) :legacy_input))
+          (is (contains? (:document card-1) :personal_owner_id)))
+        (testing "the vector round-trips through the blob column"
+          (is (= [1 2 3 4]
+                 (mapv #(Math/round (* 100.0 (double %)))
+                       (vec (semantic.models.embedding/bytes->floats (:embedding card-1)))))))))))
+
+(deftest upsert-documents-is-idempotent-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)])
+    (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)])
+    (is (= 1 (count (stored-rows))))))
+
+(deftest embeddings-are-reused-across-writes-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)
+                                     (lucene.tu/document "card" 2)])
+    (is (= 2 (count (lucene.tu/embedded-texts))))
+    (testing "re-writing the same documents does not call the embedder again"
+      (lucene.tu/reset-embedded-texts!)
+      (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)
+                                       (lucene.tu/document "card" 2)])
+      (is (= [] (lucene.tu/embedded-texts))))
+    (testing "a metadata-only change rewrites the row without calling the embedder"
+      (lucene.tu/reset-embedded-texts!)
+      (lucene.store/upsert-documents! [(lucene.tu/document "card" 1 :archived true)])
+      (is (= [] (lucene.tu/embedded-texts)))
+      (is (true? (:archived (t2/select-one :model/SemanticSearchEmbedding :model_id "1")))))
+    (testing "changing the embedded text embeds exactly that one text"
+      (lucene.tu/reset-embedded-texts!)
+      (lucene.store/upsert-documents! [(lucene.tu/document "card" 1 :name "renamed")])
+      (is (= ["[card]\nname: renamed"] (lucene.tu/embedded-texts)))
+      (is (= "renamed" (:name (t2/select-one :model/SemanticSearchEmbedding :model_id "1")))))
+    (testing "two documents sharing the same text are embedded once"
+      (lucene.tu/reset-embedded-texts!)
+      (lucene.store/upsert-documents! [(lucene.tu/document "card" 10 :name "twin")
+                                       (lucene.tu/document "dashboard" 11 :name "twin")])
+      (is (= ["[card]\nname: twin" "[dashboard]\nname: twin"]
+             (sort (lucene.tu/embedded-texts)))))))
+
+(deftest embedder-failure-skips-the-batch-test
+  (lucene.tu/with-lucene-store [4]
+    (testing "a failing embedder writes nothing and does not throw"
+      (binding [lucene.tu/*embedder-fails?* true]
+        (is (= {:rows [] :report {}}
+               (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)]))))
+      (is (= [] (stored-rows))))
+    (testing "the next write once the embedder recovers backfills it"
+      (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)])
+      (is (= 1 (count (stored-rows)))))))
+
+(deftest delete-documents-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)
+                                     (lucene.tu/document "card" 2)
+                                     (lucene.tu/document "dashboard" 3)])
+    (is (= 1 (lucene.store/delete-documents! "card" [1])))
+    (is (= [["card" "2"] ["dashboard" "3"]]
+           (map (juxt :model :model_id) (stored-rows))))
+    (testing "deleting nothing is a no-op"
+      (is (= 0 (lucene.store/delete-documents! "card" []))))))
+
+(deftest space-isolation-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)])
+    (let [four-dim-space (lucene.store/space-id)]
+      (mt/with-temporary-setting-values [ee-embedding-model-dimensions 8]
+        (testing "a different embedding model is a different space, with its own rows"
+          (is (not= four-dim-space (lucene.store/space-id)))
+          (is (= {:n 0 :mx nil} (lucene.store/space-stats (lucene.store/space-id))))
+          (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)])
+          (is (= 1 (:n (lucene.store/space-stats (lucene.store/space-id)))))))
+      (testing "the original space is untouched"
+        (is (= 1 (:n (lucene.store/space-stats four-dim-space))))
+        (is (= 2 (count (stored-rows))))))))
+
+(deftest read-helpers-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)
+                                     (lucene.tu/document "card" 2)
+                                     (lucene.tu/document "dashboard" 3)])
+    (let [space (lucene.store/space-id)]
+      (testing "space-stats counts the space and reports its watermark"
+        (let [{:keys [n mx]} (lucene.store/space-stats space)]
+          (is (= 3 n))
+          (is (some? mx))))
+      (testing "model-ids lists every row's identity"
+        (is (= #{["card" "1"] ["card" "2"] ["dashboard" "3"]}
+               (into #{} (map (juxt :model :model_id)) (lucene.store/model-ids space)))))
+      (testing "rows-after pages by id"
+        (let [[first-row :as page] (lucene.store/rows-after space 0 nil 2)]
+          (is (= 2 (count page)))
+          (is (= 1 (count (lucene.store/rows-after space (:id (last page)) nil 2))))
+          (testing "rows come back with their document decoded and their vector as bytes"
+            (is (map? (:document first-row)))
+            (is (= 16 (alength ^bytes (:embedding first-row)))))))
+      (testing "rows-after can be bounded by a watermark"
+        (is (= [] (lucene.store/rows-after space 0 (java.time.OffsetDateTime/now java.time.ZoneOffset/UTC) 10))))
+      (testing "rows-for-model-ids fetches exactly the requested rows"
+        (is (= #{["card" "2"] ["dashboard" "3"]}
+               (into #{}
+                     (map (juxt :model :model_id))
+                     (lucene.store/rows-for-model-ids space [["card" "2"] ["dashboard" "3"]]))))))))
+
+(deftest delete-space-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.store/upsert-documents! [(lucene.tu/document "card" 1)])
+    (is (= 1 (lucene.store/delete-space! (lucene.store/space-id))))
+    (is (= [] (stored-rows)))))
+
+(deftest stored-document-rebuilds-every-indexed-field-test
+  (lucene.tu/with-lucene-store [4]
+    (lucene.index/ensure-open!)
+    (let [{:keys [rows]} (lucene.store/upsert-documents!
+                          [(lucene.tu/document "card" 1
+                                               :display_type   "table"
+                                               :verified       true
+                                               :curated        true
+                                               :collection_id  7
+                                               :creator_id     11
+                                               :last_editor_id 12
+                                               :database_id    13
+                                               :created_at     "2026-01-02T03:04:05Z"
+                                               :updated_at     "2026-02-03T04:05:06Z")])]
+      (testing "the stored document keeps only what the Lucene document is built from"
+        (is (= #{:model :id :legacy_input :display_type :archived :verified :curated :collection_id
+                 :creator_id :last_editor_id :database_id :created_at :updated_at :personal_owner_id}
+               (set (keys (:document (first rows)))))))
+      (testing "and that is still enough to rebuild every field of the Lucene document schema"
+        (lucene.index/upsert-rows! rows)
+        (lucene.index/with-searcher [^IndexSearcher searcher]
+          (are [field value] (= 1 (term-count searcher field value))
+            "id"                "card_1"
+            "model"             "card"
+            "model_id"          "1"
+            "display_type"      "table"
+            "archived"          "false"
+            "verified"          "true"
+            "curated"           "true"
+            "collection_id"     "7"
+            "creator_id"        "11"
+            "last_editor_id"    "12"
+            "database_id"       "13"
+            "personal_owner_id" lucene.index/null-owner)
+          (is (= 1 (.count searcher (LongPoint/newRangeQuery "model_created_at" Long/MIN_VALUE Long/MAX_VALUE))))
+          (is (= 1 (.count searcher (LongPoint/newRangeQuery "model_updated_at" Long/MIN_VALUE Long/MAX_VALUE)))))))))

enterprise/backend/test/metabase_enterprise/semantic_search/lucene/sync_test.clj

@@ -0,0 +1,133 @@
+(ns metabase-enterprise.semantic-search.lucene.sync-test
+  (:require
+   [clojure.test :refer :all]
+   [java-time.api :as t]
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.lucene.store :as lucene.store]
+   [metabase-enterprise.semantic-search.lucene.sync :as lucene.sync]
+   [metabase-enterprise.semantic-search.lucene.test-util :as lucene.tu]
+   [metabase-enterprise.semantic-search.util :as semantic.util]
+   [metabase.test :as mt]
+   [metabase.util.json :as json]
+   [toucan2.core :as t2])
+  (:import
+   (org.apache.lucene.index Term)
+   (org.apache.lucene.search IndexSearcher TermQuery TopDocs)))
+
+(set! *warn-on-reflection* true)
+
+(defmacro ^:private with-active-engine
+  "Run `body` as if the semantic engine were one of the active search engines, which the sync gates on."
+  [& body]
+  `(mt/with-dynamic-fn-redefs [semantic.util/semantic-search-active? (constantly true)]
+     ~@body))
+
+(defn- write-elsewhere!
+  "Write documents to the embedding table only — what another node's write looks like from here."
+  [documents]
+  (lucene.store/upsert-documents! documents))
+
+(defn- indexed-legacy-input
+  "The `legacy_input` this node has indexed for `document-id`, or nil when it has none."
+  [document-id]
+  (lucene.index/with-searcher [^IndexSearcher searcher]
+    (let [^TopDocs hits (.search searcher (TermQuery. (Term. "id" ^String document-id)) (int 1))]
+      (when-let [hit (first (.scoreDocs hits))]
+        (.get (.document (.storedFields searcher) (.-doc hit)) "legacy_input")))))
+
+(deftest sync-imports-rows-written-elsewhere-test
+  (lucene.tu/with-lucene-store [4]
+    (with-active-engine
+      (write-elsewhere! [(lucene.tu/document "card" 1)
+                         (lucene.tu/document "dashboard" 2)])
+      (testing "a tick imports everything the table has"
+        (is (= {:space (lucene.store/space-id) :indexed 2 :stale 0 :missing 0}
+               (lucene.sync/sync-tick!)))
+        (is (= #{"card_1" "dashboard_2"} (lucene.index/live-ids)))))))
+
+(deftest sync-does-nothing-when-nothing-changed-test
+  (lucene.tu/with-lucene-store [4]
+    (with-active-engine
+      (write-elsewhere! [(lucene.tu/document "card" 1)])
+      (lucene.sync/sync-tick!)
+      (testing "a tick with the same row count and watermark does no work"
+        (is (= {:space (lucene.store/space-id) :skipped true}
+               (lucene.sync/sync-tick!)))))))
+
+(deftest sync-removes-rows-deleted-elsewhere-test
+  (lucene.tu/with-lucene-store [4]
+    (with-active-engine
+      (write-elsewhere! [(lucene.tu/document "card" 1)
+                         (lucene.tu/document "card" 2)])
+      (lucene.sync/sync-tick!)
+      (t2/delete! :model/SemanticSearchEmbedding :model_id "2")
+      (testing "the next tick notices the row count changed and drops the local document"
+        (is (= 1 (:stale (lucene.sync/sync-tick!))))
+        (is (= #{"card_1"} (lucene.index/live-ids)))))))
+
+(deftest sync-picks-up-documents-updated-elsewhere-test
+  (lucene.tu/with-lucene-store [4]
+    (with-active-engine
+      (write-elsewhere! [(lucene.tu/document "card" 1)])
+      (lucene.sync/sync-tick!)
+      (is (= "doc 1" (:name (json/decode+kw (indexed-legacy-input "card_1")))))
+      (write-elsewhere! [(lucene.tu/document "card" 1 :name "renamed elsewhere")])
+      (testing "the watermark moved, so the next tick re-indexes the row"
+        (is (pos? (:indexed (lucene.sync/sync-tick!))))
+        (is (= "renamed elsewhere" (:name (json/decode+kw (indexed-legacy-input "card_1")))))))))
+
+(deftest sync-repairs-a-local-index-that-fell-behind-test
+  (lucene.tu/with-lucene-store [4]
+    (with-active-engine
+      (write-elsewhere! [(lucene.tu/document "card" 1)
+                         (lucene.tu/document "card" 2)])
+      (lucene.sync/sync-tick!)
+      (lucene.index/delete-ids! ["card_2"])
+      (is (= 1 (lucene.index/live-count)))
+      (testing "a document lost locally comes back, even when it is too old for the watermark window"
+        ;; Age every row past the overlap window, so only the id comparison can find the gap.
+        (t2/update! :model/SemanticSearchEmbedding
+                    :embedding_space_id (lucene.store/space-id)
+                    {:updated_at (t/minus (t/offset-date-time) (t/hours 2))})
+        (let [{:keys [indexed missing]} (lucene.sync/sync-tick!)]
+          (is (= 0 indexed))
+          (is (= 1 missing)))
+        (is (= #{"card_1" "card_2"} (lucene.index/live-ids)))))))
+
+(deftest sync-rebuilds-from-scratch-after-reset-test
+  (lucene.tu/with-lucene-store [4]
+    (with-active-engine
+      (write-elsewhere! [(lucene.tu/document "card" 1)])
+      (lucene.sync/sync-tick!)
+      (lucene.index/delete-all!)
+      (lucene.sync/reset-state!)
+      (testing "forgetting the watermark makes the next tick a full import"
+        (is (= 1 (:indexed (lucene.sync/sync-tick!))))
+        (is (= #{"card_1"} (lucene.index/live-ids)))))))
+
+(deftest sync-is-inert-while-the-engine-is-inactive-test
+  (lucene.tu/with-lucene-store [4]
+    (write-elsewhere! [(lucene.tu/document "card" 1)])
+    (testing "nothing is indexed while the semantic engine is not active"
+      (is (nil? (lucene.sync/sync-tick!))))))
+
+(deftest start-is-idempotent-test
+  (lucene.tu/with-lucene-store [4]
+    (testing "only the first start! creates the timer"
+      (is (true? (lucene.sync/start!)))
+      (is (false? (lucene.sync/start!))))
+    (testing "stopping twice is harmless"
+      (is (nil? (lucene.sync/stop!)))
+      (is (nil? (lucene.sync/stop!))))))
+
+(deftest sync-watermark-survives-a-restart-test
+  (lucene.tu/with-lucene-store [4]
+    (with-active-engine
+      (write-elsewhere! [(lucene.tu/document "card" 1)])
+      (lucene.sync/sync-tick!)
+      (testing "after a restart the on-disk index is still warm, so the tick does not re-walk the space"
+        ;; Forgetting the in-memory state is what a restart looks like from here; the index and the watermark
+        ;; file both survive it.
+        (lucene.sync/reset-state!)
+        (is (= {:space (lucene.store/space-id) :skipped true} (lucene.sync/sync-tick!)))
+        (is (= #{"card_1"} (lucene.index/live-ids)))))))

enterprise/backend/test/metabase_enterprise/semantic_search/lucene/test_util.clj

@@ -0,0 +1,101 @@
+(ns metabase-enterprise.semantic-search.lucene.test-util
+  "Fixtures for the Lucene semantic search backend: a capturing embedding provider, a scratch index directory and a
+  clean `semantic_search_embedding` table."
+  (:require
+   [metabase-enterprise.semantic-search.lucene.index :as lucene.index]
+   [metabase-enterprise.semantic-search.lucene.sync :as lucene.sync]
+   [metabase-enterprise.semantic-search.settings :as semantic.settings]
+   [metabase.embeddings.provider :as embeddings.provider]
+   [metabase.test :as mt]
+   [metabase.util.json :as json]
+   [toucan2.core :as t2]))
+
+(comment semantic.settings/keep-me)
+
+(set! *warn-on-reflection* true)
+
+(def ^:dynamic *embeddings*
+  "Text → vector lookup the test provider answers from. Texts outside it get a constant vector."
+  {})
+
+(def ^:dynamic *embedded-texts*
+  "Atom collecting every text the test provider was asked to embed, so tests can assert on cache hits."
+  nil)
+
+(def ^:dynamic *embedder-fails?*
+  "When true the test provider throws, standing in for an embedding service that is down."
+  false)
+
+(def ^:dynamic *embed-fn*
+  "Function from text to a vector, overriding the [[*embeddings*]] lookup. Use it when the exact text a document
+  embeds is awkward to predict, e.g. the `embeddable_text` search ingestion builds for a card."
+  nil)
+
+(def ^:private default-vector [0.01 0.02 0.03 0.04])
+
+(embeddings.provider/register-provider!
+ "lucene-test"
+ {:embedding-spi-version embeddings.provider/embedding-spi-version
+  :readiness             (constantly {:ready? true})
+  :resolve-model         embeddings.provider/legacy-resolved-model
+  :embed-texts           (fn [model texts _opts]
+                           (when *embedder-fails?*
+                             (throw (ex-info "Embedding service is down" {})))
+                           (when *embedded-texts*
+                             (swap! *embedded-texts* into texts))
+                           (let [dims   (:vector-dimensions model)
+                                 lookup (or *embed-fn* #(get *embeddings* % default-vector))]
+                             (mapv (fn [text]
+                                     (vec (take dims (concat (lookup text) (repeat 0.0)))))
+                                   texts)))})
+
+(defn embedded-texts
+  "Every text the test provider has been asked to embed since the fixture started."
+  []
+  (some-> *embedded-texts* deref vec))
+
+(defn reset-embedded-texts!
+  "Forget what the test provider has embedded so far."
+  []
+  (some-> *embedded-texts* (reset! [])))
+
+(defn document
+  "A minimal ingestion document for `model`/`id`, shaped like [[metabase.search.ingestion]] produces."
+  [model id & {:as overrides}]
+  (let [document-name (or (:name overrides) (str "doc " id))]
+    (merge {:model            model
+            :id               id
+            :name             document-name
+            :archived         false
+            :collection_id    nil
+            :searchable_text  document-name
+            :embeddable_text  (str "[" model "]\nname: " document-name)
+            :display_data     {:name document-name}
+            :legacy_input     (json/encode {:id id :model model :name document-name})}
+           overrides)))
+
+(defn do-with-lucene-store!
+  "Impl for [[with-lucene-store]]."
+  [dims f]
+  (mt/with-temp-dir [dir nil]
+    (binding [lucene.index/*index-root* (str dir)
+              *embedded-texts*          (atom [])]
+      (mt/with-temporary-setting-values [ee-embedding-provider         "lucene-test"
+                                         ee-embedding-model            "test-model"
+                                         ee-embedding-model-dimensions dims]
+        (lucene.sync/stop!)
+        (t2/delete! :model/SemanticSearchEmbedding)
+        (try
+          (f)
+          (finally
+            (lucene.sync/stop!)
+            (lucene.index/close!)
+            (t2/delete! :model/SemanticSearchEmbedding)))))))
+
+(defmacro with-lucene-store
+  "Run `body` with the capturing test embedding provider configured, an empty `semantic_search_embedding` table, a
+  stopped sync timer and [[lucene.index/*index-root*]] pointed at a scratch directory. `dims` sets the provider's
+  vector width."
+  {:style/indent 1}
+  [[dims] & body]
+  `(do-with-lucene-store! ~dims (fn [] ~@body)))

enterprise/backend/test/metabase_enterprise/semantic_search/task/index_repair_test.clj

@@ -5,10 +5,17 @@
    [metabase-enterprise.semantic-search.health :as semantic.health]
    [metabase-enterprise.semantic-search.repair :as semantic.repair]
    [metabase-enterprise.semantic-search.task.index-repair :as index-repair]
+   [metabase-enterprise.semantic-search.util :as semantic.util]
    [metabase.search.ingestion :as search.ingestion]
    [metabase.test :as mt]
    [next.jdbc :as jdbc]))
 
+;; The repair metrics these tests assert on are pgvector's; the instance default backend is :lucene, which
+;; skips reporting them.
+(use-fixtures :each (fn [thunk]
+                      (mt/with-dynamic-fn-redefs [semantic.util/lucene-backend? (constantly false)]
+                        (thunk))))
+
 (deftest successful-repair-reports-current-metric-snapshot-test
   (let [reported    (atom [])
         snapshot-at ::snapshot-at]

enterprise/backend/test/metabase_enterprise/semantic_search/test_util.clj

@@ -43,7 +43,9 @@
   [f]
   (when semantic.db.datasource/db-url
     (initialize/initialize-if-needed! :db)
-    (f)))
+    ;; These namespaces exercise the pgvector backend specifically; the instance default is :lucene.
+    (mt/with-dynamic-fn-redefs [semantic.util/lucene-backend? (constantly false)]
+      (f))))
 
 (def default-test-db "my_test_db")
 

enterprise/backend/test/metabase_enterprise/serialization/models/entity_id_test.clj

@@ -121,6 +121,7 @@
     :model/RemoteSyncObject
     :model/RemoteSyncTask
     :model/Revision
+    :model/SemanticSearchEmbedding
     :model/SemanticSearchTokenTracking
     :model/SearchIndexMetadata
     :model/Secret

resources/migrations/065/20260923_semantic_search_embedding.yaml

@@ -0,0 +1,134 @@
+## Create semantic_search_embedding: the app-DB home of the vectors behind semantic search.
+##
+## The Lucene semantic-search backend keeps its vector index on each node's local disk, rebuilt from this
+## table. Embeddings are the expensive part (network calls and tokens), so they live in the app DB: that
+## survives restarts, rides along in backups, and lets every node in a cluster sync a local index from
+## shared state. One row per (embedding space, search document); `content_hash` keys the embedding cache,
+## so a metadata-only change (archiving, a view-count bump) never calls the embedding provider again.
+
+databaseChangeLog:
+  - changeSet:
+      id: v65.2026-09-23T00:00:00
+      author: paolo
+      comment: Add semantic_search_embedding
+      changes:
+        - createTable:
+            tableName: semantic_search_embedding
+            remarks: Embedding vectors for search documents, one row per search document per embedding space
+            columns:
+              - column:
+                  name: id
+                  type: int
+                  remarks: Unique ID
+                  autoIncrement: true
+                  constraints:
+                    primaryKey: true
+                    nullable: false
+              - column:
+                  name: embedding_space_id
+                  type: varchar(128)
+                  remarks: Identifies the provider/model/dimensions the vector was produced in; vectors from different spaces are not comparable
+                  constraints:
+                    nullable: false
+              - column:
+                  name: model
+                  type: varchar(64)
+                  remarks: The search model of the indexed entity, e.g. card or dashboard
+                  constraints:
+                    nullable: false
+              - column:
+                  name: model_id
+                  type: varchar(64)
+                  remarks: ID of the indexed entity within its search model
+                  constraints:
+                    nullable: false
+              - column:
+                  name: name
+                  type: ${text.type}
+                  remarks: Name of the indexed entity, kept for diagnostics
+              - column:
+                  name: archived
+                  type: ${boolean.type}
+                  remarks: Whether the indexed entity is archived
+                  defaultValueBoolean: false
+                  constraints:
+                    nullable: false
+              - column:
+                  name: content_hash
+                  type: char(64)
+                  remarks: SHA-256 hex of the embedded text; rows sharing a hash reuse one embedding
+                  constraints:
+                    nullable: false
+              - column:
+                  name: dims
+                  type: int
+                  remarks: Number of components in the embedding
+                  constraints:
+                    nullable: false
+              - column:
+                  name: embedding
+                  type: ${blob.type}
+                  remarks: The vector as little-endian float32, 4 bytes per component
+                  constraints:
+                    nullable: false
+              - column:
+                  name: document
+                  type: ${text.type}
+                  remarks: JSON of the whole search document, so the local vector index can be rebuilt from this table alone
+                  constraints:
+                    nullable: false
+              - column:
+                  name: created_at
+                  type: ${timestamp_type}
+                  remarks: The timestamp this row was created
+                  defaultValueComputed: current_timestamp
+                  constraints:
+                    nullable: false
+              - column:
+                  name: updated_at
+                  type: ${timestamp_type}
+                  remarks: The timestamp this row was last written, used as the watermark when nodes sync their local index
+                  defaultValueComputed: current_timestamp
+                  constraints:
+                    nullable: false
+
+  - changeSet:
+      id: v65.2026-09-23T00:00:01
+      author: paolo
+      comment: Add unique constraint on semantic_search_embedding space/model/model_id
+      changes:
+        - addUniqueConstraint:
+            tableName: semantic_search_embedding
+            columnNames: embedding_space_id, model, model_id
+            constraintName: uq_semantic_search_embedding_space_model_model_id
+      rollback: # nothing to do: rolling back the createTable changeset drops the whole table
+
+  - changeSet:
+      id: v65.2026-09-23T00:00:02
+      author: paolo
+      comment: Index semantic_search_embedding by embedding space and update time
+      changes:
+        - createIndex:
+            tableName: semantic_search_embedding
+            indexName: idx_semantic_search_embedding_space_updated_at
+            columns:
+              - column:
+                  name: embedding_space_id
+              - column:
+                  name: updated_at
+      rollback: # nothing to do: rolling back the createTable changeset drops the whole table
+
+  - changeSet:
+      id: v65.2026-09-23T00:00:03
+      author: paolo
+      comment: Index semantic_search_embedding by embedding space and content hash
+      changes:
+        - createIndex:
+            tableName: semantic_search_embedding
+            indexName: idx_semantic_search_embedding_space_content_hash
+            columns:
+              - column:
+                  name: embedding_space_id
+              - column:
+                  name: content_hash
+      rollback: # nothing to do: rolling back the createTable changeset drops the whole table

src/metabase/models/resolution.clj

@@ -127,6 +127,7 @@
     :model/Secret                            metabase.secrets.models.secret
     :model/SecurityAdvisory                  metabase-enterprise.security-center.models.security-advisory
     :model/Segment                           metabase.segments.models.segment
+    :model/SemanticSearchEmbedding           metabase-enterprise.semantic-search.models.embedding
     :model/SemanticSearchTokenTracking       metabase-enterprise.semantic-search.models.token-tracking
     :model/Session                           metabase.session.models.session
     :model/Setting                           metabase.settings.models.setting

src/metabase/search/appdb/core.clj

@@ -28,7 +28,6 @@
    [metabase.util.log :as log]
    [methodical.core :as methodical])
   (:import
-   (java.time OffsetDateTime)
    (java.util Queue)))
 
 ;; Register the multimethods for each specialization
@@ -54,9 +53,6 @@
       terms
       [(str/join " OR " (map #(str "(" % ")") terms))])))
 
-(defn- parse-datetime [s]
-  (when s (OffsetDateTime/parse s)))
-
 (defn- rehydrate [weights active-scorers index-row]
   (-> (json/decode+kw (:legacy_input index-row))
       search.util/collapse-id
@@ -68,9 +64,7 @@
        :all-scores (search.scoring/all-scores weights active-scorers index-row))
       ;; internal permission signal (published tables) — never surfaced in API responses
       (dissoc :is_published)
-      (update :created_at parse-datetime)
-      (update :updated_at parse-datetime)
-      (update :last_edited_at parse-datetime)))
+      search.util/rehydrate-timestamps))
 
 (defn- view-count-percentiles*
   [p-value]

src/metabase/search/core.clj

@@ -53,6 +53,7 @@
   collapse-id
   indexed-entity-id->model-index-id
   indexed-entity-id->model-pk
+  rehydrate-timestamps
   tsv-language
   to-tsquery-expr
   weighted-tsvector])

src/metabase/search/util.clj

@@ -7,7 +7,11 @@
    [metabase.search.settings :as search.settings]
    [metabase.util :as u]
    [metabase.util.i18n :as i18n]
-   [metabase.util.string :as u.str]))
+   [metabase.util.string :as u.str])
+  (:import
+   (java.time OffsetDateTime)))
+
+(set! *warn-on-reflection* true)
 
 (defn impossible-condition?
   "An (incomplete) check where queries will definitely return nothing, to help avoid spurious index update queries."
@@ -46,6 +50,21 @@
   [{:keys [id] :as row}]
   (assoc row :id (if (number? id) id (indexed-entity-id->model-pk id))))
 
+(defn- parse-datetime [s]
+  (when s (OffsetDateTime/parse s)))
+
+(defn rehydrate-timestamps
+  "Restore the `java.time` values of a search result rebuilt from its JSON `legacy_input`.
+
+  Every engine reconstructs result rows out of `legacy_input`, where JSON has flattened each timestamp to a string.
+  Consumers downstream -- `POST /api/agent/v1/search`'s response schema among them -- require real temporal values,
+  and rows must look the same whichever engine produced them, so every engine runs this over the same keys."
+  [row]
+  (-> row
+      (update :created_at parse-datetime)
+      (update :updated_at parse-datetime)
+      (update :last_edited_at parse-datetime)))
+
 ;;; ============================================================================
 ;;; Postgres-specific utilities
 ;;; ============================================================================

test/metabase/cmd/copy_test.clj

@@ -113,6 +113,7 @@
     :model/ReplacementRun
     :model/SearchIndexMetadata
     :model/SecurityAdvisory
+    :model/SemanticSearchEmbedding
     :model/SemanticSearchTokenTracking
     :model/SourceDimensionDaily
     :model/SourceDimensionProfileDaily