abstract class Memo::Queries

• Memo::Queries
• Reference
• Object

Overview

Abstract query interface for database-backend-specific SQL.

Each backend (SQLite, PostgreSQL) implements all query methods with SQL native to that backend. This avoids placeholder translation and keeps SQL explicit per backend.

Direct Known Subclasses

• Memo::Queries::Postgres
• Memo::Queries::SQLite

Defined in:

memo/queries/base.cr

Constructors

• .for(db : DB::Database, connection_string : String) : Queries
• .new(db : DBHandle)

Class Method Summary

• .register_pg(&factory : DB::Database -> Queries)

Instance Method Summary

• #add_meta(key : String, value : String) : Nil

  Set a meta value unless the key already has one

• #chunk_content_ids(source_id : Int64) : Array(Int64)

  The texts a source's chunks were cut from

• #chunk_states(source_id : Int64, service_id : Int64) : Array(Tuple(Int64 | Nil, Int64 | Nil, Int64 | Nil, String, Bool))

  Per chunk of a source: {content_id, pair_id, parent_id, source_type, whether its hash has an embedding, with a stored vector, for the service}

• #clear_completed_queue : Int32
• #clear_completed_queue_texts : Nil

  Drop the text of finished items (from before memo did so itself)

• #clear_default_service : Nil
• #clear_query_cache(service_id : Int64) : Nil
• #clear_queue : Int32
• #complete_queue_item(id : Int64, text : String, processed_at : Int64) : Bool

  The mark_queue_* methods take the item's text as it was read, and do nothing if it has changed since: the source was queued again, and the newer text still needs embedding.

• #count_chunks_by_hash(hash : Bytes) : Int64
• #count_chunks_for_service(service_id : Int64) : Int64
• #count_embeddings_for_service(service_id : Int64) : Int64
• #count_files : Int64
• #count_query_cache(service_id : Int64) : Int64
• #count_service_chunks(service_id : Int64) : Int64
• #count_service_embeddings(service_id : Int64) : Int64
• #count_service_sources(service_id : Int64) : Int64
• #count_services : Int64
• #count_vectors(service_id : Int64) : Int64
• #count_vocab(service_id : Int64) : Int64
• #db : DBHandle
• #delete_chunks(hash : Bytes, source_id : Int64, source_type : String) : Int32
• #delete_chunks(hash : Bytes, source_id : Int64) : Int32
• #delete_embeddings_by_hash(hash : Bytes) : Nil
• #delete_embeddings_by_service(service_id : Int64) : Nil
• #delete_file(source_id : Int64) : Int64
• #delete_index_data(service_id : Int64) : Nil

  Remove a service's vectors, journal entries and journal state

• #delete_queue_item(source_id : Int64) : Nil
• #delete_service(id : Int64) : Nil
• #delete_service_chunks(service_id : Int64) : Nil

  Delete the chunks whose embeddings only this service has

• #delete_source_blob(source_type : String, external_id : Bytes) : Int64
• #delete_source_by_id(internal_id : Int64) : Int64
• #delete_source_int(source_type : String, external_id : Int64) : Int64
• #delete_source_text(source_type : String, external_id : String) : Int64
• #delete_vector(embedding_id : Int64) : Nil
• #delete_vocab(service_id : Int64) : Nil
• #embedded_hashes(hashes : Array(Bytes), service_id : Int64) : Set(Bytes)

  Which of hashes have an embedding for the service with a stored vector (one without has been lost: embed it again)

• #embedding_ids_for_hash(hash : Bytes) : Array(Tuple(Int64, Int64))

  {embedding id, service id} for every service's embedding of a hash

• #embedding_ids_without_vectors(service_id : Int64) : Array(Int64)
• #enqueue(source_id : Int64, text : String, created_at : Int64) : Nil
• #fetch_search_results(service_id : Int64, params : Array(DB::Any), where_clauses : Array(String), text_join : String, fts_join : String, rowids : Array(Int64), include_text : Bool, per_embedding : Int32) : Array(Tuple(UInt64, Int64, Bytes, String, Int64, Int64 | Nil, String | Nil, Bytes | Nil, Int64 | Nil, Int64 | Nil, String | Nil, Bytes | Nil, Int64 | Nil, Int64 | Nil, String | Nil, Bytes | Nil, Int32 | Nil, Int32, Int32, Int32, String | Nil))

  The chunks of rowids' embeddings that pass the filter clauses, at most per_embedding of each.

• #files_under(prefix : String) : Array(Tuple(Int64, String))

  {source_id, path} of the files whose path starts with prefix

• #filter_candidates(service_id : Int64, params : Array(DB::Any), where_clauses : Array(String), text_join : String, fts_join : String, candidate_ids : Array(Int64)) : Set(UInt64)

  Which of candidate_ids pass the filter clauses (looked up by id, so the cost depends on the number of candidates, not the collection)

• #find_content_id(hash : Bytes) : Int64 | Nil
• #find_service_id(name : String) : Int64 | Nil
• #find_source_blob(source_type : String, external_id : Bytes) : Int64 | Nil
• #find_source_blob_any_type(external_id : Bytes) : Int64 | Nil
• #find_source_int(source_type : String, external_id : Int64) : Int64 | Nil
• #find_source_int_any_type(external_id : Int64) : Int64 | Nil
• #find_source_text(source_type : String, external_id : String) : Int64 | Nil
• #find_source_text_any_type(external_id : String) : Int64 | Nil
• #for_connection(cnn : DB::Connection) : Queries

  Same queries, run on one connection (for use inside a transaction).

• #get_all_texts : Array(String)

  Every source's stored text (a text shared by several, once for each)

• #get_chunk_hashes(source_id : Int64, source_type : String | Nil) : Array(Bytes)
• #get_chunk_id(source_id : Int64, offset : Int32 | Nil) : Int64
• #get_chunks_for_reindex(source_type : String, service_id : Int64) : Array(Tuple(Int64, Int64 | Nil, String | Nil, Int64 | Nil, Int64 | Nil))
• #get_default_service : ServiceProvider::Info | Nil
• #get_embedding_rowid(hash : Bytes, service_id : Int64) : Int64
• #get_embedding_rowid?(hash : Bytes, service_id : Int64) : Int64 | Nil
• #get_existing_words(service_id : Int64, words : Array(String)) : Set(String)
• #get_file_by_hash(hash : Bytes) : Files::FileRecord | Nil
• #get_file_by_path(path : String) : Files::FileRecord | Nil
• #get_file_by_source(source_id : Int64) : Files::FileRecord | Nil
• #get_index_state(service_id : Int64) : Tuple(Int64, Bool) | Nil

  {pruned_through, vectors_backfilled}

• #get_meta(key : String) : String | Nil
• #get_pending_queue(limit : Int32) : Array(Tuple(Int64, Int64, String))
• #get_query_cache(query : String, service_id : Int64) : Tuple(Bytes, Int32) | Nil
• #get_queue_attempts(id : Int64) : Int32 | Nil

  nil once the item is gone (its source deleted)

• #get_queue_item(source_id : Int64) : Tuple(Int64, String) | Nil
• #get_recent_query_cache(service_id : Int64, limit : Int32) : Array(Tuple(String, Bytes, Int32))
• #get_service_by_format_model(format : String, model : String) : Tuple(Int64, String, String | Nil, String, Int32, Int32, Float64) | Nil
• #get_service_by_name(name : String) : Tuple(Int64, String, String | Nil, String, Int32, Int32, Float64) | Nil
• #get_service_info(id : Int64) : ServiceProvider::Info | Nil
• #get_service_info_by_name(name : String) : ServiceProvider::Info | Nil
• #get_service_tokens_per_byte(service_id : Int64) : Float64 | Nil
• #get_source_external(internal_id : Int64) : Tuple(String, Int64 | Nil, String | Nil, Bytes | Nil) | Nil
• #get_source_text(source_id : Int64) : String | Nil

  A source's latest text (nil if not stored)

• #get_texts_for_reindex(source_type : String) : Array(Tuple(Int64, Int64 | Nil, Int64 | Nil, String))
• #get_tokens_per_byte(service_id : Int64) : Float64 | Nil
• #get_vector(embedding_id : Int64, service_id : Int64) : Bytes | Nil
• #get_vocab(service_id : Int64, &block : DB::ResultSet -> ) : Nil
• #increment_match_count(chunk_ids : Array(Int64)) : Nil
• #increment_read_count(chunk_ids : Array(Int64)) : Nil
• #index_changes_since(service_id : Int64, seq : Int64) : Array(Int64)

  Distinct embedding ids logged after seq

• #index_removals_between(service_id : Int64, after : Int64, through : Int64) : Array(Int64)

  Embeddings logged for the service after after and through through that it no longer has (no row, no vector): removals

• #insert_chunk_ignore(hash : Bytes, source_id : Int64, source_type : String, content_id : Int64 | Nil, pair_id : Int64 | Nil, parent_id : Int64 | Nil, offset : Int32 | Nil, size : Int32, created_at : Int64) : Int64
• #insert_embedding_ignore(hash : Bytes, service_id : Int64, token_count : Int32, created_at : Int64) : Int64
• #insert_service(name : String, format : String, base_url : String | Nil, model : String, dimensions : Int32, max_tokens : Int32, created_at : Int64) : Int64
• #insert_service_full(name : String, format : String, base_url : String | Nil, model : String, dimensions : Int32, max_tokens : Int32, is_default : Int32, created_at : Int64) : Int64

  Insert service with is_default and return ID (for ServiceProvider.create)

• #insert_source(source_type : String, created_at : Int64) : Int64
• #insert_source_blob(source_type : String, external_id : Bytes, created_at : Int64) : Int64 | Nil
• #insert_source_int(source_type : String, external_id : Int64, created_at : Int64) : Int64 | Nil

  Insert a source unless one with this id exists (another writer may have added it since the caller looked).

• #insert_source_text(source_type : String, external_id : String, created_at : Int64) : Int64 | Nil
• #legacy_texts_after(source_id : Int64, limit : Int32) : Array(Tuple(Int64, String, Bytes | Nil))

  memo_texts rows (databases from before 0.13): {source_id, content, content_hash}

• #list_files(limit : Int32, offset : Int32) : Array(Files::FileRecord)
• #list_services : Array(ServiceProvider::Info)
• #list_services_by_format(format : String) : Array(ServiceProvider::Info)
• #list_sources(source_type : String, limit : Int32, offset : Int32) : Array(Tuple(Int64, Int64 | Nil, String | Nil, Bytes | Nil))
• #load_embedding_rowids(service_id : Int64, source_type : String, external_ids : Array(Int64)) : Array(Tuple(Int64, Int64))
• #log_index_change(service_id : Int64, embedding_id : Int64) : Nil
• #mark_queue_failed(id : Int64, text : String, error_message : String | Nil, attempts : Int32, processed_at : Int64) : Nil
• #mark_queue_retry(id : Int64, text : String, attempts : Int32, error_message : String | Nil) : Nil
• #max_index_log_seq(service_id : Int64) : Int64
• #prune_index_log(service_id : Int64, through : Int64) : Nil
• #prune_query_cache(service_id : Int64, count : Int64) : Nil
• #queue_stats : Tuple(Int64, Int64)
• #release_content(content_id : Int64) : Nil

  Delete a text unless a source or chunk still points at it

• #search_filtered_rowids(service_id : Int64, params : Array(DB::Any), where_clauses : Array(String), text_join : String, fts_join : String) : Set(UInt64)
• #service_exists?(id : Int64) : Bool
• #set_chunks_content(source_id : Int64, content_id : Int64) : Nil
• #set_default_service(id : Int64) : Nil
• #set_index_pruned_through(service_id : Int64, seq : Int64) : Nil
• #set_meta(key : String, value : String) : Nil
• #set_source_content(source_id : Int64, content_id : Int64 | Nil) : Nil
• #set_vectors_backfilled(service_id : Int64) : Nil
• #source_content_id(source_id : Int64) : Int64 | Nil
• #update_file_path(source_id : Int64, path : String) : Nil
• #update_service(id : Int64, updates : Array(String), params : Array(DB::Any)) : Int64
• #update_tokens_per_byte(tokens_per_byte : Float64, service_id : Int64) : Nil
• #update_word_frequency(count : Int32, word : String, service_id : Int64) : Nil
• #upsert_content(hash : Bytes, content : String | Nil, created_at : Int64) : Int64

  The id of the text with this hash, storing it if new.

• #upsert_file(source_id : Int64, path : String, content_hash : Bytes, mtime : Int64, size : Int64, created_at : Int64) : Nil
• #upsert_query_cache(query : String, service_id : Int64, embedding : Bytes, token_count : Int32, created_at : Int64) : Nil
• #upsert_vector(embedding_id : Int64, service_id : Int64, vector : Bytes) : Nil
• #upsert_vocab(word : String, service_id : Int64, embedding : Bytes, frequency : Int32, created_at : Int64) : Nil
• #vectors_after(service_id : Int64, after_id : Int64, limit : Int32) : Array(Tuple(Int64, Bytes))

  Up to limit {embedding id, vector} pairs with ids above after_id, in id order

Constructor Detail

Class Method Detail

Instance Method Detail