Documentation
¶
Overview ¶
Package datasetindex is the catalog-dataset consumer of the shared indexjobs framework (#1131). It gives a fact written into the DataHub catalog a search-time route that does not depend on a tool result already naming the entity the fact hangs off.
Before it, the catalog's only text route through `search` was DataHub's own keyword search: a description an agent applied to a dataset was found when a query happened to share its words, and only once DataHub's index caught up. Knowledge pages had no such gap — they ride the platform's own embedding search — so whether applied knowledge was discoverable depended on which sink an operator picked. This consumer removes that asymmetry by mirroring the catalog's dataset text into the platform's own index and ranking it there.
The unit is the whole corpus (one Key: SourceKind/SourceID), not one dataset per unit, because the corpus does not live in a table the reconciler can diff: it lives in DataHub. So Source.LoadItems enumerates the catalog and materializes catalog_datasets from the result, the framework embeds the rows, and Sink.Upsert's atomic replace drops the rows for datasets the catalog no longer returns. Two consequences follow from that shape:
- The periodic re-enumeration needs no goroutine of its own. Sink.FindGaps reports the corpus stale once SyncInterval has passed since the last enumeration, the reconciler enqueues the unit, and the queue's own claim/lease machinery makes the sweep single-flight across replicas.
- A sweep is cheap when nothing changed: the framework's text-hash dedup re-embeds only the datasets whose text actually moved.
The catalog stays the system of record. Nothing here writes back to DataHub, every hit is dereferenced against DataHub by URN, and the tables are safe to drop — the cost is only that `search` falls back to DataHub's own keyword search, which is what it used before this package existed.
Index ¶
- Constants
- func IndexText(e Entry) string
- func RegisterConsumer(reg interface{ ... }, db *sql.DB, lister Lister, currentModel string, ...) error
- func Searcher(db *sql.DB, cfg Config) knowledge.CatalogIndexSearcher
- type Config
- type Entry
- type Lister
- type Sink
- func (s *Sink) Coverage(ctx context.Context) (indexjobs.Coverage, error)
- func (s *Sink) FindGaps(ctx context.Context) ([]string, error)
- func (*Sink) Kind() string
- func (s *Sink) ListExisting(ctx context.Context, _ indexjobs.Key) (map[string]indexjobs.Vector, error)
- func (*Sink) StampExpected(context.Context, indexjobs.Key, int) error
- func (s *Sink) Upsert(ctx context.Context, _ indexjobs.Key, rows []indexjobs.Vector) error
- func (s *Sink) UpsertBatch(ctx context.Context, _ indexjobs.Key, rows []indexjobs.Vector) error
- type Source
- type Store
- func (s *Store) Coverage(ctx context.Context) (indexed, expected int, err error)
- func (s *Store) ListVectors(ctx context.Context) (map[string]indexjobs.Vector, error)
- func (s *Store) NeedsSweep(ctx context.Context, currentModel string, interval time.Duration) (bool, error)
- func (s *Store) ReplaceVectors(ctx context.Context, rows []indexjobs.Vector) error
- func (s *Store) SearchCatalogIndex(ctx context.Context, q knowledge.CatalogIndexQuery) ([]knowledge.CatalogIndexHit, error)
- func (s *Store) StampSync(ctx context.Context) error
- func (s *Store) Sync(ctx context.Context, entries []Entry) error
- func (s *Store) UpsertVectors(ctx context.Context, rows []indexjobs.Vector) error
Constants ¶
const ( // DefaultSyncInterval is how long a synced corpus is considered fresh. // Thirty minutes keeps a newly applied description discoverable within // one working pause while enumerating a large catalog only twice an hour. DefaultSyncInterval = 30 * time.Minute // DefaultMaxEntries caps how many datasets are mirrored. The cap bounds // both the table and the embed pass's working set (one vector per entry is // held in memory during a sweep). A deployment whose catalog exceeds it // indexes the first page-order slice and logs the truncation; raise // knowledge.catalog_index.max_entries to cover the rest. DefaultMaxEntries = 5000 )
Default tuning for the catalog sweep. Both are deliberately conservative: the sweep is a paged read against an external catalog, so it trades freshness for load.
const SourceID = "catalog"
SourceID is the single logical catalog-corpus identifier. A deployment has one configured semantic catalog, so a constant source id is sufficient; the rows keyed under it are shared by every replica.
const SourceKind = "catalog-datasets"
SourceKind is the indexjobs source_kind this package serves.
Variables ¶
This section is empty.
Functions ¶
func IndexText ¶
IndexText composes the text a catalog dataset is embedded and lexically indexed on: its name, description, tags, and domain. The Source and the request-path ranking MUST agree on this composition so a stored embedding lives in the same space as the query; it is defined once here for both, and catalog_dataset_fts (migration 000096) composes the same corpus from the same columns for the lexical arm. Empty fields are skipped so a bare dataset does not pad the text with blank lines.
func RegisterConsumer ¶
func RegisterConsumer(reg interface {
Register(indexjobs.Source, indexjobs.Sink) error
}, db *sql.DB, lister Lister, currentModel string, cfg Config,
) error
RegisterConsumer registers the catalog-dataset Source/Sink pair on the shared indexjobs registry. lister enumerates the catalog; currentModel is the embedding provider's model identifier, which the gap query diffs stored rows against so a model swap re-embeds them.
func Searcher ¶
func Searcher(db *sql.DB, cfg Config) knowledge.CatalogIndexSearcher
Searcher returns the request-path ranking capability over the mirrored catalog, or nil when the deployment has no database or the operator disabled the index. It returns the interface (never a typed nil) so a caller can pass the result straight into the search federation and have a nil check mean "no index".
Types ¶
type Config ¶
type Config struct {
// Enabled turns the consumer on. Nil (unset) means enabled: a deployment
// with a DataHub catalog, a database, and an embedding provider gets
// topic-level discoverability of catalog descriptions out of the box. Set
// false to opt out, which leaves `search` ranking catalog datasets through
// DataHub's own keyword search alone.
Enabled *bool `yaml:"enabled"`
// SyncInterval is how often the catalog is re-enumerated. Zero uses
// DefaultSyncInterval.
SyncInterval time.Duration `yaml:"sync_interval"`
// MaxEntries caps how many datasets are mirrored. Zero uses
// DefaultMaxEntries.
MaxEntries int `yaml:"max_entries"`
}
Config is the operator's control over the catalog index. It is embedded in the platform's knowledge config as `knowledge.catalog_index`.
func (Config) IsEnabled ¶
IsEnabled reports whether the catalog index is enabled, defaulting to true when not explicitly set.
func (Config) ResolvedMaxEntries ¶
ResolvedMaxEntries returns the configured entry cap, substituting the default for an unset (non-positive) value.
func (Config) ResolvedSyncInterval ¶
ResolvedSyncInterval returns the configured sweep interval, substituting the default for an unset (non-positive) value.
type Entry ¶
Entry is one catalog dataset as the platform mirrors it: the identity the catalog and `fetch` share (URN) plus the text an agent searches by. It carries no schema, lineage, or ownership — those stay in DataHub and are read through the entity path when a caller drills in.
type Lister ¶
type Lister interface {
SearchTables(ctx context.Context, filter semantic.SearchFilter) ([]semantic.TableSearchResult, error)
}
Lister is the catalog capability the Source needs: a paged relevance search it can drive as an enumeration. It matches semantic.Provider; declared locally so the Source depends on the capability and tests can supply a fake.
type Sink ¶
type Sink struct {
// contains filtered or unexported fields
}
Sink implements indexjobs.Sink for the catalog-datasets kind over catalog_datasets. currentModel is the provider model the gap query diffs stored rows against, so a model swap re-embeds rows stamped with the previous model; syncInterval is how long an enumeration stays fresh before the gap sweep asks for another one.
func NewSink ¶
NewSink returns a Sink backed by the given store. currentModel is the embedding provider's model identifier (embedding.ModelName); pass "" on a deployment whose provider does not name its model, in which case only missing-embedding rows count as gaps.
func (*Sink) Coverage ¶
Coverage reports the catalog-datasets kind's indexed-vs-expected totals (mirrored datasets carrying a vector vs all mirrored datasets). ExpectedKnown is true: every mirrored dataset is expected to converge to one vector. It describes the mirror, not the catalog — a corpus truncated by the entry cap reports full coverage of what it mirrors, which is why the Source logs the truncation.
func (*Sink) FindGaps ¶
FindGaps returns the corpus unit when it owes work: the enumeration has aged out (which is how the periodic sweep is scheduled without a goroutine of its own) or some mirrored dataset has no vector, or one from a superseded model.
func (*Sink) ListExisting ¶
func (s *Sink) ListExisting(ctx context.Context, _ indexjobs.Key) (map[string]indexjobs.Vector, error)
ListExisting returns every mirrored dataset's vector for the worker's dedup pass. The unit is the whole corpus, so the key carries no narrowing.
func (*Sink) StampExpected ¶
StampExpected is a no-op. Gap detection here is condition-based (sweep age, missing or stale-model vectors), not count-based: the expected count lives in DataHub, so a stamped number could only ever restate the last sweep's own result. The sweep timestamp — the part that does drive scheduling — is stamped by the Source when the enumeration it describes actually happened.
type Source ¶
type Source struct {
// contains filtered or unexported fields
}
Source implements indexjobs.Source for the catalog-datasets kind. The unit is the whole corpus, so LoadItems both enumerates the catalog and materializes the mirror the ranking reads; see the package doc for why the corpus is one unit rather than one unit per dataset.
func NewSource ¶
NewSource returns a Source over the mirror store and the catalog lister. maxEntries caps how many datasets are mirrored.
func (*Source) LoadItems ¶
LoadItems enumerates the catalog, writes the result into the mirror, stamps the sweep, and returns one embeddable item per dataset.
It never reports a partial corpus: any enumeration error fails the job, so the atomic replace that follows a successful pass cannot prune datasets that merely went unread. A catalog that legitimately holds nothing yields an empty item set, which the framework treats as a clean completion — and the Sink's replace then clears a mirror left over from a catalog that has been emptied.
func (*Source) OnSucceeded ¶
OnSucceeded is a no-op: the ranked search reads catalog_datasets directly on every query, so there is no in-memory cache to refresh after a sweep.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store owns the platform's copy of the catalog's dataset text (catalog_datasets) and the sweep marker (catalog_dataset_sync). It serves both sides of the consumer: the write path the indexjobs worker drives (Sync/StampSync, the vector reads and writes, gap detection) and the request-path ranking the search federation reads.
func (*Store) Coverage ¶
Coverage returns the mirrored datasets carrying a vector and the total mirrored, the two halves of the admin Indexing dashboard's ratio.
func (*Store) ListVectors ¶
ListVectors returns every mirrored dataset's persisted vector keyed by URN (the item id), for the worker's text-hash + model dedup pass. Rows with no embedding yet are omitted so the worker embeds them.
func (*Store) NeedsSweep ¶
func (s *Store) NeedsSweep(ctx context.Context, currentModel string, interval time.Duration) (bool, error)
NeedsSweep reports whether the corpus owes work: either the enumeration is older than interval (or has never run), or some mirrored dataset carries no vector or one produced by a model other than currentModel. It is the whole of gap detection for this kind — the corpus lives in DataHub, so "what is missing" cannot be answered by diffing two local tables, and the freshness half is what turns the reconciler's tick into the sweep schedule.
func (*Store) ReplaceVectors ¶
ReplaceVectors is the Sink's atomic replace: it deletes every mirrored dataset outside the supplied set and writes the set's vectors, in one transaction. Deleting the ROW (not just its vector) is what prunes a dataset the catalog no longer returns — the row is the index entry, so a dropped dataset must not linger as a lexically-matchable hit whose fetch 404s.
An empty set clears the mirror. That is the correct reading of "the catalog returned nothing": the Source fails the job on any enumeration error rather than reporting a partial corpus, so an empty set means an empty catalog.
func (*Store) SearchCatalogIndex ¶
func (s *Store) SearchCatalogIndex(ctx context.Context, q knowledge.CatalogIndexQuery) ([]knowledge.CatalogIndexHit, error)
SearchCatalogIndex ranks the mirrored catalog datasets by relevance to the query. A non-nil q.Embedding selects hybrid (semantic + lexical) ranking; a nil embedding selects lexical-only ranking, which is what `search` falls back to when the intent could not be embedded (a slow or unreachable embedder). The mirror itself is only ever populated by the index queue, which requires a configured embedding provider, so a deployment without one has an empty mirror and this returns nothing rather than a degraded ranking.
func (*Store) StampSync ¶
StampSync records that the catalog was enumerated now. The single row is the schedule the gap sweep reads: without it an empty catalog (a legitimate state) would look permanently un-synced and re-enumerate on every reconciler tick.
func (*Store) Sync ¶
Sync materializes the enumerated catalog into catalog_datasets: every entry is inserted or updated in one transaction. Rows for datasets the enumeration did NOT return are left alone here and dropped by ReplaceVectors, which is the Sink's atomic-replace contract — keeping one delete path means a failed embed pass cannot leave the mirror pruned against a corpus that was never re-indexed.
The embedding columns are deliberately untouched: the framework's text-hash dedup compares the stored hash against the freshly composed text in the same job, so a changed description is re-embedded without this write having to invalidate anything, and an unchanged one keeps a usable vector throughout.
func (*Store) UpsertVectors ¶
UpsertVectors writes the embedding columns for the supplied rows onto their mirrored dataset. A row whose dataset vanished between the sync and the write updates nothing rather than resurrecting it (the UPDATE matches no row), so a concurrent prune is not undone.
Source Files