Class: Iriq::Storage::Sqlite

Inherits:
Object
  • Object
show all
Defined in:
lib/iriq/storage/sqlite.rb

Overview

Sqlite is the incremental-write backend. Each observation translates to a handful of UPSERTs against a long-lived connection; nothing is materialized in memory beyond what reads explicitly ask for.

WAL journaling lets multiple processes observe against the same file concurrently — the writer is serialized, readers are not blocked, and the existing iriq --corpus c.db <url> pattern works without a flock at the application layer.

Constant Summary collapse

SCHEMA_VERSION = 
4
SCHEMA = 
<<~SQL.freeze
  CREATE TABLE IF NOT EXISTS meta (
    key   TEXT PRIMARY KEY,
    value TEXT
  );
  CREATE TABLE IF NOT EXISTS host_counts (
    host  TEXT PRIMARY KEY,
    count INTEGER NOT NULL
  );
  CREATE TABLE IF NOT EXISTS path_length_counts (
    length INTEGER PRIMARY KEY,
    count  INTEGER NOT NULL
  );
  CREATE TABLE IF NOT EXISTS raw_shape_counts (
    shape TEXT PRIMARY KEY,
    count INTEGER NOT NULL
  );
  CREATE TABLE IF NOT EXISTS fingerprint_counts (
    shape TEXT PRIMARY KEY,
    count INTEGER NOT NULL
  );
  -- Position is (host, scope, locator). For scope='path' the locator
  -- is the typed prefix; for scope='query' it's the param name.
  -- Today only 'path' is observed here (query params live on the
  -- cluster_* tables) — scope is in the schema so future commits
  -- can fold query positions in without another migration.
  CREATE TABLE IF NOT EXISTS position_stats (
    host    TEXT NOT NULL,
    scope   TEXT NOT NULL,
    locator TEXT NOT NULL,
    total   INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (host, scope, locator)
  );
  CREATE TABLE IF NOT EXISTS position_values (
    host    TEXT NOT NULL,
    scope   TEXT NOT NULL,
    locator TEXT NOT NULL,
    value   TEXT NOT NULL,
    count   INTEGER NOT NULL,
    PRIMARY KEY (host, scope, locator, value)
  );
  CREATE TABLE IF NOT EXISTS position_types (
    host    TEXT NOT NULL,
    scope   TEXT NOT NULL,
    locator TEXT NOT NULL,
    type    TEXT NOT NULL,
    count   INTEGER NOT NULL,
    PRIMARY KEY (host, scope, locator, type)
  );
  CREATE TABLE IF NOT EXISTS clusters (
    key    TEXT PRIMARY KEY,
    host   TEXT,
    scheme TEXT,
    shape  TEXT,
    count  INTEGER NOT NULL DEFAULT 0,
    ord    INTEGER NOT NULL
  );
  CREATE TABLE IF NOT EXISTS cluster_examples (
    cluster_key TEXT NOT NULL,
    position    INTEGER NOT NULL,
    canonical   TEXT NOT NULL,
    PRIMARY KEY (cluster_key, position)
  );
  CREATE TABLE IF NOT EXISTS cluster_segments (
    cluster_key TEXT NOT NULL,
    position    INTEGER NOT NULL,
    value       TEXT NOT NULL,
    count       INTEGER NOT NULL,
    PRIMARY KEY (cluster_key, position, value)
  );
  CREATE TABLE IF NOT EXISTS cluster_params (
    cluster_key TEXT NOT NULL,
    name        TEXT NOT NULL,
    total       INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (cluster_key, name)
  );
  CREATE TABLE IF NOT EXISTS cluster_param_values (
    cluster_key TEXT NOT NULL,
    name        TEXT NOT NULL,
    value       TEXT NOT NULL,
    count       INTEGER NOT NULL,
    PRIMARY KEY (cluster_key, name, value)
  );
  CREATE TABLE IF NOT EXISTS cluster_param_types (
    cluster_key TEXT NOT NULL,
    name        TEXT NOT NULL,
    type        TEXT NOT NULL,
    count       INTEGER NOT NULL,
    PRIMARY KEY (cluster_key, name, type)
  );
  -- Source-IRI log. The materialized views above are derived from
  -- this log via events + reducers. Corpus#reinfer drops the views
  -- and replays the log to rebuild them. id is monotonic so
  -- iteration order is observation order.
  CREATE TABLE IF NOT EXISTS observed_iris (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    canonical TEXT NOT NULL
  );
  -- Recognizers promoted from RecognizerProposal via
  -- Corpus#activate_proposal. Re-applied to the corpus's
  -- classifier on Corpus.open so a reopen picks up its learned
  -- patterns. Keyed by prefix; activating the same prefix twice
  -- is a no-op.
  CREATE TABLE IF NOT EXISTS activated_recognizers (
    prefix      TEXT PRIMARY KEY,
    type        TEXT NOT NULL,
    specificity REAL NOT NULL DEFAULT 1.0
  );
SQL

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path:, classifier: SegmentClassifier::DEFAULT, max_values_per_position: PositionStats::DEFAULT_MAX_VALUES) ⇒ Sqlite

Returns a new instance of Sqlite.



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/iriq/storage/sqlite.rb', line 133

def initialize(path:, classifier: SegmentClassifier::DEFAULT,
               max_values_per_position: PositionStats::DEFAULT_MAX_VALUES)
  @path                    = path
  @classifier              = classifier
  @max_values_per_position = max_values_per_position
  @db                      = SQLite3::Database.new(path)
  # busy_timeout MUST come first: other PRAGMAs (journal_mode in
  # particular) can themselves block on the write lock under
  # concurrent open, and without busy_timeout set they fail
  # immediately with SQLITE_BUSY.
  @db.execute("PRAGMA busy_timeout = 30000")
  enable_wal!
  @db.execute("PRAGMA synchronous = NORMAL")
  @db.execute("PRAGMA foreign_keys = ON")
  @in_batch = false
end

Instance Attribute Details

#max_values_per_position ⇒ Object (readonly)

Returns the value of attribute max_values_per_position.



126
127
128
 
# File 'lib/iriq/storage/sqlite.rb', line 126

def max_values_per_position
  @max_values_per_position
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



126
127
128
 
# File 'lib/iriq/storage/sqlite.rb', line 126

def path
  @path
end

Class Method Details

.open(path, classifier: SegmentClassifier::DEFAULT, max_values_per_position: PositionStats::DEFAULT_MAX_VALUES) ⇒ Object 



128
129
130
131
 
# File 'lib/iriq/storage/sqlite.rb', line 128

def self.open(path, classifier: SegmentClassifier::DEFAULT,
                    max_values_per_position: PositionStats::DEFAULT_MAX_VALUES)
  new(path: path, classifier: classifier, max_values_per_position: max_values_per_position).tap(&:setup!)
end

Instance Method Details

#activated_recognizer_count ⇒ Object 



380
381
382
 
# File 'lib/iriq/storage/sqlite.rb', line 380

def activated_recognizer_count
  @db.get_first_value("SELECT COUNT(*) FROM activated_recognizers") || 0
end

#add_to_cluster(key, host, scheme, shape, identifier) ⇒ Object 



277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
 
# File 'lib/iriq/storage/sqlite.rb', line 277

def add_to_cluster(key, host, scheme, shape, identifier)
  # Insert the cluster row if new (with a monotonic ord for stable
  # iteration), then bump its count.
  @db.execute(<<~SQL, [key, host, scheme, shape])
    INSERT INTO clusters (key, host, scheme, shape, count, ord)
    VALUES (?, ?, ?, ?, 1, (SELECT COALESCE(MAX(ord), 0) + 1 FROM clusters))
    ON CONFLICT(key) DO UPDATE SET count = count + 1
  SQL

  # Examples — capped at Cluster::MAX_EXAMPLES, deduped by canonical
  # (mirrors Cluster#add so the SQLite view matches the in-memory one).
  canonical = identifier.canonical
  examples_count = @db.get_first_value(
    "SELECT COUNT(*) FROM cluster_examples WHERE cluster_key = ?", [key],
  )
  already_present = @db.get_first_value(
    "SELECT 1 FROM cluster_examples WHERE cluster_key = ? AND canonical = ?",
    [key, canonical],
  )
  if examples_count < Cluster::MAX_EXAMPLES && already_present.nil?
    @db.execute(<<~SQL, [key, examples_count, canonical])
      INSERT INTO cluster_examples (cluster_key, position, canonical)
      VALUES (?, ?, ?)
    SQL
  end

  # Per-position segment counts — uncapped.
  identifier.path_segments.each_with_index do |seg, i|
    @db.execute(<<~SQL, [key, i, seg])
      INSERT INTO cluster_segments (cluster_key, position, value, count) VALUES (?, ?, ?, 1)
      ON CONFLICT(cluster_key, position, value) DO UPDATE SET count = count + 1
    SQL
  end

  # Per-param stats (presence + value cardinality + type) — mirrors the
  # in-memory Cluster#add path. Value table respects the same per-key
  # cap as position_values.
  (identifier.query_params || {}).each do |name, value|
    v = value.to_s
    type = @classifier.classify(v).to_s

    @db.execute(<<~SQL, [key, name])
      INSERT INTO cluster_params (cluster_key, name, total) VALUES (?, ?, 1)
      ON CONFLICT(cluster_key, name) DO UPDATE SET total = total + 1
    SQL
    @db.execute(<<~SQL, [key, name, type])
      INSERT INTO cluster_param_types (cluster_key, name, type, count) VALUES (?, ?, ?, 1)
      ON CONFLICT(cluster_key, name, type) DO UPDATE SET count = count + 1
    SQL

    @db.execute(<<~SQL, [key, name, v])
      UPDATE cluster_param_values SET count = count + 1
      WHERE cluster_key = ? AND name = ? AND value = ?
    SQL
    if @db.changes.zero?
      card = @db.get_first_value(
        "SELECT COUNT(*) FROM cluster_param_values WHERE cluster_key = ? AND name = ?",
        [key, name],
      )
      if card < @max_values_per_position
        @db.execute(
          "INSERT INTO cluster_param_values (cluster_key, name, value, count) VALUES (?, ?, ?, 1)",
          [key, name, v],
        )
      end
    end
  end

  load_cluster(key)
end

#batch ⇒ Object

Wrap many observations in a single transaction. Cuts SQLite write overhead from O(observations) fsyncs to O(1).



183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
# File 'lib/iriq/storage/sqlite.rb', line 183

def batch
  return yield if @in_batch

  @in_batch = true
  @db.transaction
  begin
    yield
    @db.commit
  rescue
    @db.rollback rescue nil
    raise
  ensure
    @in_batch = false
  end
end

#clear_materialized_views ⇒ Object

Drop every materialized view without touching the source-IRI log. Corpus#reinfer calls this before replaying the log.



386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
 
# File 'lib/iriq/storage/sqlite.rb', line 386

def clear_materialized_views
  @db.execute_batch(<<~SQL)
    DELETE FROM host_counts;
    DELETE FROM path_length_counts;
    DELETE FROM raw_shape_counts;
    DELETE FROM fingerprint_counts;
    DELETE FROM position_stats;
    DELETE FROM position_values;
    DELETE FROM position_types;
    DELETE FROM clusters;
    DELETE FROM cluster_examples;
    DELETE FROM cluster_segments;
    DELETE FROM cluster_params;
    DELETE FROM cluster_param_values;
    DELETE FROM cluster_param_types;
  SQL
end

#close ⇒ Object 



207
208
209
210
211
212
 
# File 'lib/iriq/storage/sqlite.rb', line 207

def close
  # Checkpoint + truncate the WAL so the .db-wal sidecar doesn't grow
  # unbounded across long-lived `iriq --corpus c.db` sessions.
  @db.execute("PRAGMA wal_checkpoint(TRUNCATE)") rescue nil
  @db.close
end

#cluster_for(key) ⇒ Object 



478
479
480
 
# File 'lib/iriq/storage/sqlite.rb', line 478

def cluster_for(key)
  load_cluster(key)
end

#cluster_size ⇒ Object 



474
475
476
 
# File 'lib/iriq/storage/sqlite.rb', line 474

def cluster_size
  @db.get_first_value("SELECT COUNT(*) FROM clusters")
end

#clusters ⇒ Object 



466
467
468
469
470
471
472
 
# File 'lib/iriq/storage/sqlite.rb', line 466

def clusters
  out = []
  @db.execute("SELECT key FROM clusters ORDER BY ord") do |row|
    out << load_cluster(row[0])
  end
  out
end

#each_activated_recognizer ⇒ Object 



374
375
376
377
378
 
# File 'lib/iriq/storage/sqlite.rb', line 374

def each_activated_recognizer
  @db.execute("SELECT prefix, type, specificity FROM activated_recognizers ORDER BY prefix") do |row|
    yield({ "prefix" => row[0], "type" => row[1], "specificity" => row[2] })
  end
end

#each_observed_iri ⇒ Object 



355
356
357
358
359
 
# File 'lib/iriq/storage/sqlite.rb', line 355

def each_observed_iri
  @db.execute("SELECT canonical FROM observed_iris ORDER BY id") do |row|
    yield row[0]
  end
end

#each_position_stats ⇒ Object 



455
456
457
458
459
460
461
462
463
464
 
# File 'lib/iriq/storage/sqlite.rb', line 455

def each_position_stats
  seen = []
  @db.execute("SELECT DISTINCT host, scope, locator FROM position_stats ORDER BY ROWID") do |row|
    seen << row
  end
  seen.each do |host, scope, locator|
    pos = Position.new(host: host, scope: scope.to_sym, locator: locator)
    yield pos, position_stats(pos)
  end
end

#fingerprint_counts ⇒ Object 



420
421
422
 
# File 'lib/iriq/storage/sqlite.rb', line 420

def fingerprint_counts
  rows_to_count_hash("fingerprint_counts", "shape")
end

#flush ⇒ Object

Saving is automatic — incremental UPSERTs hit disk on commit. flush makes that explicit; close releases the connection.



201
 
# File 'lib/iriq/storage/sqlite.rb', line 201

def flush; end

#host_counts ⇒ Object

--- Reads ------------------------------------------------------------



406
407
408
 
# File 'lib/iriq/storage/sqlite.rb', line 406

def host_counts
  rows_to_count_hash("host_counts", "host")
end

#increment_fingerprint(shape) ⇒ Object 



236
237
238
 
# File 'lib/iriq/storage/sqlite.rb', line 236

def increment_fingerprint(shape)
  upsert_shape("fingerprint_counts", shape)
end

#increment_host(host) ⇒ Object

--- Increments -------------------------------------------------------



216
217
218
219
220
221
222
223
 
# File 'lib/iriq/storage/sqlite.rb', line 216

def increment_host(host)
  return unless host

  @db.execute(<<~SQL, host)
    INSERT INTO host_counts (host, count) VALUES (?, 1)
    ON CONFLICT(host) DO UPDATE SET count = count + 1
  SQL
end

#increment_path_length(length) ⇒ Object 



225
226
227
228
229
230
 
# File 'lib/iriq/storage/sqlite.rb', line 225

def increment_path_length(length)
  @db.execute(<<~SQL, length)
    INSERT INTO path_length_counts (length, count) VALUES (?, 1)
    ON CONFLICT(length) DO UPDATE SET count = count + 1
  SQL
end

#increment_raw_shape(shape) ⇒ Object 



232
233
234
 
# File 'lib/iriq/storage/sqlite.rb', line 232

def increment_raw_shape(shape)
  upsert_shape("raw_shape_counts", shape)
end

#observe_position(position, value, type) ⇒ Object 



240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/iriq/storage/sqlite.rb', line 240

def observe_position(position, value, type)
  host    = position.host || ""
  scope   = position.scope.to_s
  locator = position.locator
  @db.execute(<<~SQL, [host, scope, locator])
    INSERT INTO position_stats (host, scope, locator, total) VALUES (?, ?, ?, 1)
    ON CONFLICT(host, scope, locator) DO UPDATE SET total = total + 1
  SQL

  # Type counts are unbounded — always upsert.
  @db.execute(<<~SQL, [host, scope, locator, type.to_s])
    INSERT INTO position_types (host, scope, locator, type, count) VALUES (?, ?, ?, ?, 1)
    ON CONFLICT(host, scope, locator, type) DO UPDATE SET count = count + 1
  SQL

  # Value counts are capped at max_values_per_position. If the value
  # already exists, increment it; otherwise insert only when
  # cardinality is below the cap. Two-step rather than ON CONFLICT
  # because we need to enforce the cap on insert.
  @db.execute(<<~SQL, [host, scope, locator, value])
    UPDATE position_values SET count = count + 1
    WHERE host = ? AND scope = ? AND locator = ? AND value = ?
  SQL
  if @db.changes.zero?
    card = @db.get_first_value(
      "SELECT COUNT(*) FROM position_values WHERE host = ? AND scope = ? AND locator = ?",
      [host, scope, locator],
    )
    if card < @max_values_per_position
      @db.execute(
        "INSERT INTO position_values (host, scope, locator, value, count) VALUES (?, ?, ?, ?, 1)",
        [host, scope, locator, value],
      )
    end
  end
end

#observed_iri_count ⇒ Object 



361
362
363
 
# File 'lib/iriq/storage/sqlite.rb', line 361

def observed_iri_count
  @db.get_first_value("SELECT COUNT(*) FROM observed_iris") || 0
end

#path_length_counts ⇒ Object 



410
411
412
413
414
 
# File 'lib/iriq/storage/sqlite.rb', line 410

def path_length_counts
  h = Hash.new(0)
  @db.execute("SELECT length, count FROM path_length_counts") { |r| h[r[0]] = r[1] }
  h
end

#position_stats(position) ⇒ Object 



424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
 
# File 'lib/iriq/storage/sqlite.rb', line 424

def position_stats(position)
  host    = position.host || ""
  scope   = position.scope.to_s
  locator = position.locator
  total = @db.get_first_value(
    "SELECT total FROM position_stats WHERE host = ? AND scope = ? AND locator = ?",
    [host, scope, locator],
  )
  return nil if total.nil?

  stats = PositionStats.new(max_values: @max_values_per_position)
  stats.instance_variable_set(:@total, total)

  vc = Hash.new(0)
  @db.execute(
    "SELECT value, count FROM position_values WHERE host = ? AND scope = ? AND locator = ?",
    [host, scope, locator],
  ) { |r| vc[r[0]] = r[1] }
  stats.instance_variable_set(:@value_counts, vc)

  tc = Hash.new(0)
  @db.execute(
    "SELECT type, count FROM position_types WHERE host = ? AND scope = ? AND locator = ?",
    [host, scope, locator],
  ) { |r| tc[r[0].to_sym] = r[1] }
  stats.instance_variable_set(:@type_counts, tc)

  recompute_numeric!(stats)
  stats
end

#raw_shape_counts ⇒ Object 



416
417
418
 
# File 'lib/iriq/storage/sqlite.rb', line 416

def raw_shape_counts
  rows_to_count_hash("raw_shape_counts", "shape")
end

#record_activated_recognizer(dump) ⇒ Object

--- Activated recognizers --------------------------------------------



367
368
369
370
371
372
 
# File 'lib/iriq/storage/sqlite.rb', line 367

def record_activated_recognizer(dump)
  @db.execute(<<~SQL, [dump["prefix"], dump["type"], dump.fetch("specificity", 1.0)])
    INSERT INTO activated_recognizers (prefix, type, specificity) VALUES (?, ?, ?)
    ON CONFLICT(prefix) DO UPDATE SET type = excluded.type, specificity = excluded.specificity
  SQL
end

#record_observation(canonical) ⇒ Object

Append a canonical IRI to the source-IRI log. Inside the same transaction as the event reducers, so the log and views stay consistent.



351
352
353
 
# File 'lib/iriq/storage/sqlite.rb', line 351

def record_observation(canonical)
  @db.execute("INSERT INTO observed_iris (canonical) VALUES (?)", [canonical])
end

#save(_path = nil) ⇒ Object 



203
204
205
 
# File 'lib/iriq/storage/sqlite.rb', line 203

def save(_path = nil)
  # Already persisted. Provided for parity with the JSON backend.
end

#setup! ⇒ Object 



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/iriq/storage/sqlite.rb', line 150

def setup!
  @db.execute_batch(SCHEMA)
  existing = @db.get_first_value("SELECT value FROM meta WHERE key = 'schema_version'")
  if existing.nil?
    # OR IGNORE: two processes can race to initialize a fresh corpus
    # concurrently — both read schema_version as nil, and the loser's
    # INSERT must not blow up on the PRIMARY KEY.
    @db.execute("INSERT OR IGNORE INTO meta (key, value) VALUES ('schema_version', ?)", SCHEMA_VERSION.to_s)
    @db.execute("INSERT OR IGNORE INTO meta (key, value) VALUES ('max_values_per_position', ?)",
                @max_values_per_position.to_s)
  else
    @max_values_per_position = (@db.get_first_value(
      "SELECT value FROM meta WHERE key = 'max_values_per_position'"
    ) || @max_values_per_position).to_i
  end
  self
end

#transaction ⇒ Object 



168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/iriq/storage/sqlite.rb', line 168

def transaction
  # While inside an outer batch, observe()-time transactions become
  # no-ops — the outer batch wraps everything in one txn for speed.
  return yield(self) if @in_batch

  @db.transaction
  yield self
  @db.commit
rescue
  @db.rollback rescue nil
  raise
end