nsmcp::AccessRecordStoreSQLite method ingest (public)

 <instance of nsmcp::AccessRecordStoreSQLite> ingest fid \
    checkpoint tuples [ batched ]

Defined in /usr/local/ns/tcl/nsmcp/log-traffic-record-store.tcl

Persist the scanner's supplied requests and checkpoint atomically. Insert every supplied request; duplicate source positions are errors. Commit rollups, pending peer addresses, parsing-loss counters and byte coverage in the same transaction. This method does not read the source file.

Parameters:

fid (required)

Stored source-file identifier.

checkpoint (required)

Scanner-prepared coverage, framing, timestamp and parsing-loss metadata.

tuples (required)

Prepared request rows, or serialized SQL tuples for the reference path.

batched (optional, defaults to "false")

True for the batch API; false for the SQL reference implementation.

Returns:

Committed request and diagnostic counts, together with the insertion-backend label.

Testcases:

No testcase defined.

Source code:

foreach key {begin end lo hi anchor stamp now dropping} {
    set $key [dict get $checkpoint $key]
}
set malformed [:dict getdef $checkpoint malformed 0]
set oversized [:dict getdef $checkpoint oversized 0]
set skipped [:dict getdef $checkpoint skipped 0]
foreach count [list $malformed $oversized $skipped] {
    if {![string is entier -strict $count] || $count<0} {
        error "invalid checkpoint counter"
    }
}
set committed [dict create  records 0  requestRecords 0  malformedRecords $malformed  oversizedRecords $oversized]
if {$batched && [llength $tuples] > 512} {
    error "record-store-batch-limit"
}
:accessIngestionObserve begin-transaction
${:db} transaction {
    if {![${:db} scalar {SELECT count(*) FROM nsmcp_records.files WHERE fid=:fid}]} {
        error "record-store-generation-changed"
    }
    if {[llength $tuples]} {
        set previousRowid  [${:db} scalar {SELECT coalesce(max(rowid),0) FROM nsmcp_records.rows}]
        if {$previousRowid == 9223372036854775807} {
            error "record-store-rowid-limit"
        }
        :accessIngestionObserve insert-records
        if {$batched} {
            ${:db} batchInsert -schema ${:batchSchema} {nsmcp_records rows} $tuples
        } else {
            ${:db} executeDml "INSERT INTO nsmcp_records.rows VALUES [join $tuples ,]"
        }
        # The writer lock makes this watermark identify precisely the
        # inserted batch, without staging or filtering its records.
        set batchSource "(SELECT * FROM nsmcp_records.rows WHERE rowid>$previousRowid)"
        :accessIngestionObserve count-records
        set committed [dict merge $committed [lindex [${:db} execute "
            SELECT count(*) AS records,
            coalesce(sum(kind=0),0) AS requestRecords FROM $batchSource
        "] 0]]
        :accessIngestionObserve rollup
        :rollBatch $batchSource
        :accessIngestionObserve observe-records
        :observeBatch $batchSource
    }
    :accessIngestionObserve store-peers
    if {[dict size ${:peers}]} {
        set values {}
        dict for {key data} ${:peers} {
            lassign $data address seen
            lappend values [list $key $address $seen]
        }
        ${:db} executeDml  -bindtypes {values {rows {binary scalar scalar}}} {
                INSERT INTO nsmcp_records.peer_addresses VALUES :values
                ON CONFLICT(k) DO UPDATE SET address=excluded.address,seen=excluded.seen
                WHERE excluded.seen>=seen AND (excluded.seen<>seen OR excluded.address<>address)
            }
    }
    :accessIngestionObserve checkpoint-counters
    # Parsing losses are committed with the same source interval. A
    # bad-only batch advances coverage without inserting a request row.
    if {$malformed+$oversized+$skipped>0} {
        set observed  [:stateGet observations {first 0 latest 0 completed 0 malformed 0 oversized 0}]
        dict incr observed malformed $malformed
        dict incr observed oversized $oversized
        :statePut observations $observed
        ${:db} execute {
            INSERT INTO nsmcp_records.scan_counts VALUES(:fid,:begin,:end,:malformed,:oversized,:skipped,:now)
        }
    }
    dict incr committed records [expr {$malformed + $oversized + $skipped}]
    :accessIngestionObserve save-ranges
    ${:db} execute {
        DELETE FROM nsmcp_records.ranges WHERE fid=:fid AND lo<=:end AND hi>=:begin;
        INSERT INTO nsmcp_records.ranges VALUES(:fid,:lo,:hi,:anchor);
        UPDATE nsmcp_records.files
        SET dropping=CASE WHEN :end>=high THEN :dropping ELSE dropping END,
        stamp=CASE WHEN :end>=high THEN :stamp ELSE stamp END,
        high=max(high,:end),touched=:now
        WHERE fid=:fid
    }
    :accessIngestionObserve check-storage
    if {[:diskBytes] > [dict get ${:settings} maxBytes]} {
        error "record-store-disk-limit"
    }
    :accessIngestionObserve commit
}
set :peers {}
dict set committed insertionBackend [expr {$batched ? "batch-api" : "sql-values"}]
return $committed

XQL Not present:

Generic, PostgreSQL, Oracle