nsmcp::AccessRecordPartitions method partitionIngest (protected)

 <instance of nsmcp::AccessRecordPartitions> partitionIngest fid \
    checkpoint tuples batched

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

Parameters:

fid (required)

checkpoint (required)

tuples (required)

batched (required)

Testcases:

No testcase defined.

Source code:

set groups {}
set firsts {}
set lasts {}
set floor [:retentionFloor]
if {![:dict getdef ${:settings} historicalBackfill false]} {
    set floor [expr {max($floor,[:stateGet bootstrapFrom $floor])}]
}
set skipped 0
foreach tuple $tuples {
    if {$batched} {
        set time [lindex $tuple 4]
        if {[lindex $tuple 3]!=0} {
            error "partitioned ingestion requires valid requests"
        }
    } else {
        if {![dict exists ${:tupleTimes} $tuple]} {
            error "partitioned SQL ingestion requires tuple routing metadata"
        }
        set time [dict get ${:tupleTimes} $tuple]
    }
    if {![string is double -strict $time] || $time<0 || $time>[clock seconds]+3600} {
        error "record-store-invalid-partition-time"
    }
    if {$time<$floor} {
        incr skipped
        continue
    }
    set bucket [expr {wide(floor($time/${:partitionInterval}))*${:partitionInterval}}]
    dict lappend groups $bucket $tuple
    if {![dict exists $firsts $bucket]} {
        dict set firsts $bucket $time
        dict set lasts $bucket $time
    } else {
        dict set firsts $bucket [expr {min([dict get $firsts $bucket],$time)}]
        dict set lasts $bucket [expr {max([dict get $lasts $bucket],$time)}]
    }
}
dict set checkpoint skipped [expr {[:dict getdef $checkpoint skipped 0]+$skipped}]
if {$groups eq {}} {
    dict set groups empty {}
}
set committed {records 0 requestRecords 0 malformedRecords 0 oversizedRecords 0}
set remaining [dict size $groups]
try {
    :accessIngestionObserve begin-transaction
    ${:db} transaction {
        dict for {bucket rows} $groups {
            set target main.row_template
            if {$bucket ne "empty"} {
                set target [:partitionEnsure $bucket]
            }
            set :recordTable $target
            set portion $checkpoint
            incr remaining -1
            if {$remaining>0} {
                dict set portion malformed 0
                dict set portion oversized 0
                dict set portion skipped 0
            }
            set added [:ingestCanonical $fid $portion $rows $batched true]
            foreach key {records requestRecords malformedRecords oversizedRecords} {
                dict incr committed $key [dict get $added $key]
            }
            if {$bucket ne "empty"} {
                set n [dict get $added requestRecords]
                set first [dict get $firsts $bucket]
                set last [dict get $lasts $bucket]
                ${:db} execute {
                    UPDATE main.partitions
                       SET first=CASE WHEN n=0 THEN :first ELSE min(first,:first) END,
                           last=CASE WHEN n=0 THEN :last ELSE max(last,:last) END,n=n+:n
                     WHERE start=:bucket
                }
            }
        }
        :sqliteBeforeCommit
    }
} finally {
    set :recordTable main.rows
    set :tupleTimes {}
}
dict set committed insertionBackend [expr {$batched ? "batch-api" : "sql-values"}]
return $committed

XQL Not present:

Generic, PostgreSQL, Oracle