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
LOCK_WAIT =

Processes share a corpus's write lock by one rule: a writer waits up to LOCK_WAIT seconds for its turn, and none keeps the lock longer than a turn of work.

10
LOCK_TURN = 
1.0
TURN_PAUSE =

How long a writer that used a whole turn leaves the lock free: many of a waiter's 1ms busy retries, and a fair chance at an older iriq's 100ms ones.

0.02
VIEW_TABLES =

Every table derived from the observation log.

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

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.



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/iriq/storage/sqlite.rb', line 166

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)
  # Before any PRAGMA: journal_mode itself can wait on a lock. SQLite's
  # own busy_timeout backs off to 100ms between retries, which would
  # mostly miss a TURN_PAUSE; this one retries every 1ms.
  @db.busy_handler_timeout = LOCK_WAIT * 1000
  enable_wal!
  @db.execute("PRAGMA synchronous = NORMAL")
  # Up to 64MB of pages (a ceiling, not an allocation): an ingest that
  # commits a turn at a time re-reads the pages it wrote last turn, and
  # SQLite's default 2MB cache turns that into disk reads.
  @db.execute("PRAGMA cache_size = -64000")
  @db.execute("PRAGMA foreign_keys = ON")
  @in_batch       = false
  @in_transaction = false
  # Values tracked per position, remembered so each new value needn't
  # re-count them. Exact only inside a transaction (the write lock is
  # held) and while @counts_version still matches.
  @value_counts   = {}
  # PRAGMA data_version when @value_counts was last known exact; it
  # changes only when another connection commits.
  @counts_version = nil
  # When the transaction in progress took the write lock, and until
  # when a connection that used a whole turn leaves the lock free.
  @locked_at      = nil
  @next_turn      = nil
  # A rebuild is in progress: TEMP tables named like the views shadow
  # them for this connection alone, so every view statement writes the
  # rebuild. Its own transaction touches only those tables; it ends
  # before a log read, so the read sees the latest log, and before the
  # write lock is taken.
  @rebuilding     = false
  @rebuild_txn    = false
end

Instance Attribute Details

#max_values_per_position ⇒ Object (readonly)

Returns the value of attribute max_values_per_position.



144
145
146
 
# File 'lib/iriq/storage/sqlite.rb', line 144

def max_values_per_position
  @max_values_per_position
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



144
145
146
 
# File 'lib/iriq/storage/sqlite.rb', line 144

def path
  @path
end

Class Method Details

.corpus_error(path, e) ⇒ Object

A failure SQLite reports (not a database, can't open, read-only, full disk) reads like every other corpus failure: corpus PATH: cause. The gem appends the failing SQL to its message; the cause is the part before.



156
157
158
159
160
161
162
163
164
 
# File 'lib/iriq/storage/sqlite.rb', line 156

def self.corpus_error(path, e)
  # Busy only ever means the wait for another process's lock ran out.
  cause = if e.is_a?(SQLite3::BusyException)
    "another process held the corpus lock for over #{LOCK_WAIT}s"
  else
    e.message.split(":\n", 2).first
  end
  CorpusError.new("corpus #{path}: #{cause}")
end

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



146
147
148
149
150
151
 
# File 'lib/iriq/storage/sqlite.rb', line 146

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!)
rescue SQLite3::Exception => e
  raise corpus_error(path, e)
end

Instance Method Details

#activated_recognizer_count ⇒ Object 



548
549
550
 
# File 'lib/iriq/storage/sqlite.rb', line 548

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

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



435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
 
# File 'lib/iriq/storage/sqlite.rb', line 435

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("    INSERT INTO clusters (key, host, scheme, shape, count, ord)\n    VALUES (?, ?, ?, ?, 1, (SELECT COALESCE(MAX(ord), 0) + 1 FROM clusters))\n    ON CONFLICT(key) DO UPDATE SET count = count + 1\n  SQL\n\n  # Examples \u2014 capped at Cluster::MAX_EXAMPLES, deduped by canonical\n  # (mirrors Cluster#add so the SQLite view matches the in-memory one).\n  canonical = identifier.canonical\n  examples_count = @db.get_first_value(\n    \"SELECT COUNT(*) FROM cluster_examples WHERE cluster_key = ?\", [key],\n  )\n  already_present = @db.get_first_value(\n    \"SELECT 1 FROM cluster_examples WHERE cluster_key = ? AND canonical = ?\",\n    [key, canonical],\n  )\n  if examples_count < Cluster::MAX_EXAMPLES && already_present.nil?\n    @db.execute(<<~SQL, [key, examples_count, canonical])\n      INSERT INTO cluster_examples (cluster_key, position, canonical)\n      VALUES (?, ?, ?)\n    SQL\n  end\n\n  # Per-position segment counts \u2014 uncapped.\n  identifier.path_segments.each_with_index do |seg, i|\n    @db.execute(<<~SQL, [key, i, seg])\n      INSERT INTO cluster_segments (cluster_key, position, value, count) VALUES (?, ?, ?, 1)\n      ON CONFLICT(cluster_key, position, value) DO UPDATE SET count = count + 1\n    SQL\n  end\n\n  # Per-param stats (presence + value cardinality + type) \u2014 mirrors the\n  # in-memory Cluster#add path. Value table respects the same per-key\n  # cap as position_values.\n  (identifier.query_params || {}).each do |name, value|\n    v = value.to_s\n    type = @classifier.classify(v).to_s\n\n    @db.execute(<<~SQL, [key, name])\n      INSERT INTO cluster_params (cluster_key, name, total) VALUES (?, ?, 1)\n      ON CONFLICT(cluster_key, name) DO UPDATE SET total = total + 1\n    SQL\n    @db.execute(<<~SQL, [key, name, type])\n      INSERT INTO cluster_param_types (cluster_key, name, type, count) VALUES (?, ?, ?, 1)\n      ON CONFLICT(cluster_key, name, type) DO UPDATE SET count = count + 1\n    SQL\n\n    @db.execute(<<~SQL, [key, name, v])\n      UPDATE cluster_param_values SET count = count + 1\n      WHERE cluster_key = ? AND name = ? AND value = ?\n    SQL\n    if @db.changes.zero?\n      card = @db.get_first_value(\n        \"SELECT COUNT(*) FROM cluster_param_values WHERE cluster_key = ? AND name = ?\",\n        [key, name],\n      )\n      if card < @max_values_per_position\n        @db.execute(\n          \"INSERT INTO cluster_param_values (cluster_key, name, value, count) VALUES (?, ?, ?, 1)\",\n          [key, name, v],\n        )\n      end\n    end\n  end\nend\n", [key, host, scheme, shape])

#batch ⇒ Object

Wrap many observations in a single transaction. Cuts SQLite write overhead from O(observations) fsyncs to O(1). Yields whether another connection may have committed since this one's last transaction.



248
249
250
251
252
253
254
255
256
257
 
# File 'lib/iriq/storage/sqlite.rb', line 248

def batch
  return yield(false) if @in_batch

  @in_batch = true
  begin
    write_transaction { |changed| yield changed }
  ensure
    @in_batch = false
  end
end

#begin_rebuild ⇒ Object

--- Rebuilding the views ---------------------------------------------

Rebuild out of sight: view writes after begin_rebuild go to fresh, empty views only this connection sees, until install_rebuild (inside a transaction) makes them the corpus's views. discard_rebuild drops a rebuild not installed, and also runs after an install whose transaction rolled back, which brings the TEMP tables back. Other connections keep reading and writing the live views meanwhile.



311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
 
# File 'lib/iriq/storage/sqlite.rb', line 311

def begin_rebuild
  discard_rebuild
  VIEW_TABLES.each do |table|
    # The live table's own definition, so the copy lines up column for
    # column whichever iriq created the corpus.
    sql = @db.get_first_value("SELECT sql FROM main.sqlite_master WHERE type = 'table' AND name = ?", [table])
    @db.execute(sql.sub("CREATE TABLE", "CREATE TEMP TABLE"))
  end
  unless @in_transaction
    @db.transaction
    @rebuild_txn = true
  end
  @value_counts.clear
  @rebuilding = true
rescue SQLite3::Exception => e
  raise corpus_error(e)
end

#clear_materialized_views ⇒ Object

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



554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
 
# File 'lib/iriq/storage/sqlite.rb', line 554

def clear_materialized_views
  @value_counts.clear
  @db.execute_batch("    DELETE FROM host_counts;\n    DELETE FROM path_length_counts;\n    DELETE FROM raw_shape_counts;\n    DELETE FROM fingerprint_counts;\n    DELETE FROM position_stats;\n    DELETE FROM position_values;\n    DELETE FROM position_types;\n    DELETE FROM clusters;\n    DELETE FROM cluster_examples;\n    DELETE FROM cluster_segments;\n    DELETE FROM cluster_params;\n    DELETE FROM cluster_param_values;\n    DELETE FROM cluster_param_types;\n  SQL\nend\n")

#close ⇒ Object 



363
364
365
366
367
368
 
# File 'lib/iriq/storage/sqlite.rb', line 363

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 



647
648
649
 
# File 'lib/iriq/storage/sqlite.rb', line 647

def cluster_for(key)
  load_cluster(key)
end

#cluster_size ⇒ Object 



643
644
645
 
# File 'lib/iriq/storage/sqlite.rb', line 643

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

#clusters ⇒ Object 



635
636
637
638
639
640
641
 
# File 'lib/iriq/storage/sqlite.rb', line 635

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

#corpus_error(e) ⇒ Object

SQLite failures inside a transaction become CorpusErrors; anything else passes through.



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

def corpus_error(e)
  e.is_a?(SQLite3::Exception) ? Sqlite.corpus_error(@path, e) : e
end

#discard_rebuild ⇒ Object 



339
340
341
342
343
344
345
346
347
 
# File 'lib/iriq/storage/sqlite.rb', line 339

def discard_rebuild
  if @rebuild_txn
    @db.rollback rescue nil
    @rebuild_txn = false
  end
  VIEW_TABLES.each { |table| @db.execute("DROP TABLE IF EXISTS temp.#{table}") rescue nil }
  @rebuilding = false
  @value_counts.clear
end

#each_activated_recognizer ⇒ Object 



542
543
544
545
546
 
# File 'lib/iriq/storage/sqlite.rb', line 542

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 



511
512
513
514
515
 
# File 'lib/iriq/storage/sqlite.rb', line 511

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

#each_observed_iri_since(mark) ⇒ Object

The observations logged after mark (0 for all), in id order; returns the id of the last one (mark when there are none).



519
520
521
522
523
524
525
526
527
 
# File 'lib/iriq/storage/sqlite.rb', line 519

def each_observed_iri_since(mark)
  if @rebuild_txn
    @db.commit
    @db.transaction
  end
  rows = @db.execute("SELECT id, canonical FROM observed_iris WHERE id > ? ORDER BY id", [mark])
  rows.each { |_id, canonical| yield canonical }
  rows.empty? ? mark : rows.last[0]
end

#each_position_stats ⇒ Object 



624
625
626
627
628
629
630
631
632
633
 
# File 'lib/iriq/storage/sqlite.rb', line 624

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 



589
590
591
 
# File 'lib/iriq/storage/sqlite.rb', line 589

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.



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

def flush; end

#host_counts ⇒ Object

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



575
576
577
 
# File 'lib/iriq/storage/sqlite.rb', line 575

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

#increment_fingerprint(shape) ⇒ Object 



392
393
394
 
# File 'lib/iriq/storage/sqlite.rb', line 392

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

#increment_host(host) ⇒ Object

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



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

def increment_host(host)
  return unless host

  @db.execute("    INSERT INTO host_counts (host, count) VALUES (?, 1)\n    ON CONFLICT(host) DO UPDATE SET count = count + 1\n  SQL\nend\n", host)

#increment_path_length(length) ⇒ Object 



381
382
383
384
385
386
 
# File 'lib/iriq/storage/sqlite.rb', line 381

def increment_path_length(length)
  @db.execute("    INSERT INTO path_length_counts (length, count) VALUES (?, 1)\n    ON CONFLICT(length) DO UPDATE SET count = count + 1\n  SQL\nend\n", length)

#increment_raw_shape(shape) ⇒ Object 



388
389
390
 
# File 'lib/iriq/storage/sqlite.rb', line 388

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

#install_rebuild ⇒ Object 



329
330
331
332
333
334
335
336
337
 
# File 'lib/iriq/storage/sqlite.rb', line 329

def install_rebuild
  VIEW_TABLES.each do |table|
    @db.execute("DELETE FROM main.#{table}")
    @db.execute("INSERT INTO main.#{table} SELECT * FROM temp.#{table}")
    @db.execute("DROP TABLE temp.#{table}")
  end
  @rebuilding = false
  @value_counts.clear
end

#observe_position(position, value, type) ⇒ Object 



396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
 
# File 'lib/iriq/storage/sqlite.rb', line 396

def observe_position(position, value, type)
  host    = position.host || ""
  scope   = position.scope.to_s
  locator = position.locator
  @db.execute("    INSERT INTO position_stats (host, scope, locator, total) VALUES (?, ?, ?, 1)\n    ON CONFLICT(host, scope, locator) DO UPDATE SET total = total + 1\n  SQL\n\n  # Type counts are unbounded \u2014 always upsert.\n  @db.execute(<<~SQL, [host, scope, locator, type.to_s])\n    INSERT INTO position_types (host, scope, locator, type, count) VALUES (?, ?, ?, ?, 1)\n    ON CONFLICT(host, scope, locator, type) DO UPDATE SET count = count + 1\n  SQL\n\n  # Value counts are capped at max_values_per_position. If the value\n  # already exists, increment it; otherwise insert only when\n  # cardinality is below the cap. Two-step rather than ON CONFLICT\n  # because we need to enforce the cap on insert.\n  @db.execute(<<~SQL, [host, scope, locator, value])\n    UPDATE position_values SET count = count + 1\n    WHERE host = ? AND scope = ? AND locator = ? AND value = ?\n  SQL\n  if @db.changes.zero?\n    where = [host, scope, locator]\n    card = (counts_exact? && @value_counts[where]) || @db.get_first_value(\n      \"SELECT COUNT(*) FROM position_values WHERE host = ? AND scope = ? AND locator = ?\", where,\n    )\n    if card < @max_values_per_position\n      @db.execute(\n        \"INSERT INTO position_values (host, scope, locator, value, count) VALUES (?, ?, ?, ?, 1)\",\n        [host, scope, locator, value],\n      )\n      card += 1\n    end\n    @value_counts[where] = card if counts_exact?\n  end\nend\n", [host, scope, locator])

#observed_iri_count ⇒ Object 



529
530
531
 
# File 'lib/iriq/storage/sqlite.rb', line 529

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

#param_stats(key, name) ⇒ Object

One query param's stats — narrower than cluster_for, which also loads the cluster's examples and segment counts.



680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
 
# File 'lib/iriq/storage/sqlite.rb', line 680

def param_stats(key, name)
  total = @db.get_first_value(
    "SELECT total FROM cluster_params WHERE cluster_key = ? AND name = ?", [key, name],
  )
  return nil if total.nil?

  stats = PositionStats.new(max_values: @max_values_per_position)
  stats.instance_variable_set(:@total, coerce_int!(total, 0, "total"))
  @db.execute(
    "SELECT value, count FROM cluster_param_values WHERE cluster_key = ? AND name = ?", [key, name],
  ) { |r| stats.value_counts[r[0]] = coerce_int!(r[1], 1, "count") }
  @db.execute(
    "SELECT type, count FROM cluster_param_types WHERE cluster_key = ? AND name = ?", [key, name],
  ) { |r| stats.type_counts[r[0].to_sym] = coerce_int!(r[1], 1, "count") }
  recompute_numeric!(stats)
  stats
end

#path_length_counts ⇒ Object 



579
580
581
582
583
 
# File 'lib/iriq/storage/sqlite.rb', line 579

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

#position_evidence(position, value) ⇒ Object

What Corpus#classify reads: counts, not every value tracked at the position.



653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
 
# File 'lib/iriq/storage/sqlite.rb', line 653

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

  type_counts = Hash.new(0)
  @db.execute(
    "SELECT type, count FROM position_types WHERE host = ? AND scope = ? AND locator = ?", where,
  ) { |r| type_counts[r[0].to_sym] = coerce_int!(r[1], 1, "count") }
  value_count = @db.get_first_value(
    "SELECT count FROM position_values WHERE host = ? AND scope = ? AND locator = ? AND value = ?",
    [*where, value],
  )
  PositionEvidence.new(
    total:       coerce_int!(total, 0, "total"),
    type_counts: type_counts,
    cardinality: @db.get_first_value(
      "SELECT COUNT(*) FROM position_values WHERE host = ? AND scope = ? AND locator = ?", where,
    ),
    value_count: value_count.nil? ? nil : coerce_int!(value_count, 0, "count"),
  )
end

#position_stats(position) ⇒ Object 



593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
 
# File 'lib/iriq/storage/sqlite.rb', line 593

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, coerce_int!(total, 0, "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]] = coerce_int!(r[1], 1, "count") }
  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] = coerce_int!(r[1], 1, "count") }
  stats.instance_variable_set(:@type_counts, tc)

  recompute_numeric!(stats)
  stats
end

#raw_shape_counts ⇒ Object 



585
586
587
 
# File 'lib/iriq/storage/sqlite.rb', line 585

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

#record_activated_recognizer(dump) ⇒ Object

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



535
536
537
538
539
540
 
# File 'lib/iriq/storage/sqlite.rb', line 535

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

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



507
508
509
 
# File 'lib/iriq/storage/sqlite.rb', line 507

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

#refuse_newer_schema! ⇒ Object

Checked before SCHEMA runs, so iriq never adds tables to a corpus written by a newer iriq.

Raises:



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

def refuse_newer_schema!
  has_meta = @db.get_first_value("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'meta'")
  return unless has_meta

  version = @db.get_first_value("SELECT value FROM meta WHERE key = 'schema_version'").to_i
  return if version <= SCHEMA_VERSION

  @db.close
  raise CorpusError, "corpus #{@path}: schema version #{version} is newer than this iriq supports (#{SCHEMA_VERSION}); upgrade iriq"
end

#save(_path = nil) ⇒ Object 



359
360
361
 
# File 'lib/iriq/storage/sqlite.rb', line 359

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

#setup! ⇒ Object 



218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
 
# File 'lib/iriq/storage/sqlite.rb', line 218

def setup!
  refuse_newer_schema!
  @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 



237
238
239
240
241
242
243
 
# File 'lib/iriq/storage/sqlite.rb', line 237

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

  write_transaction { yield self }
end

#turn_over? ⇒ Boolean

Whether the transaction in progress has held the write lock for a whole turn, so a long-running writer should commit and let others in.

Returns:



298
299
300
 
# File 'lib/iriq/storage/sqlite.rb', line 298

def turn_over?
  !@locked_at.nil? && monotonic_now - @locked_at >= LOCK_TURN
end