scanner.go

package main

import (
    "bytes"
    "encoding/binary"
    "fmt"
)

// Scanner wraps the BIter to provide high-level Range Queries over Relational Tables.
type Scanner struct {
    Cmp1 int
    Cmp2 int
    Key1 Record
    Key2 Record

    tx     *DBTX   
    tdef   *TableDef
    index  int 
    iter   *BIter
    keyEnd []byte
}

// Scan initializes a scanner to iterate over a specified range within a transaction.
func (tx *DBTX) Scan(table string, req *Scanner) error {
    tdef := getTableDef(tx, table)
    if tdef == nil {
        return fmt.Errorf("table not found: %s", table)
    }
    
    req.tdef = tdef
    req.tx = tx

    //Index Selection: Find an index that covers the requested columns
    isCovered := func(index []string) bool {
        keyCols := req.Key1.Cols
        if len(index) < len(keyCols) {
            return false
        }
        for i := range keyCols {
            if index[i] != keyCols[i] {
                return false
            }
        }
        return true
    }

    req.index = -1
    for i, idxCols := range tdef.Indexes {
        if isCovered(idxCols) {
            req.index = i
            break
        }
    }
    if req.index == -1 {
        return fmt.Errorf("no suitable index found for query columns")
    }

    // Extract values based on the matched index
    val1, err := checkRecord(tdef, req.Key1, len(req.Key1.Cols))
    if err != nil { return err }
    val2, err := checkRecord(tdef, req.Key2, len(req.Key2.Cols))
    if err != nil { return err }

    prefix := tdef.Prefixes[req.index]
    keyStart := encodeKeyPartial(nil, prefix, val1, req.Cmp1)
    req.keyEnd = encodeKeyPartial(nil, prefix, val2, req.Cmp2)
    req.iter = tx.kv.Seek(keyStart, req.Cmp1)
    return nil
}

// Valid verifies if the current scanner position is still within the user's requested range constraints.
func (sc *Scanner) Valid() bool {
    if !sc.iter.Valid() {
        return false
    }
    
    key, _ := sc.iter.Deref()
    
    var buf [4]byte
    binary.BigEndian.PutUint32(buf[:], sc.tdef.Prefixes[sc.index])
    if !bytes.HasPrefix(key, buf[:]) {
        return false 
    }
    
    return cmpOK(key, sc.Cmp2, sc.keyEnd)
}

// Next moves to the next row in the table.
func (sc *Scanner) Next() {
    sc.iter.Next()
}

// Deref extracts the current row into a human-readable Record.
func (sc *Scanner) Deref(rec *Record) {
    key, val := sc.iter.Deref()

    if sc.index == 0 {
        // Primary Index Scan
        values := make([]Value, len(sc.tdef.Cols))
        
        pkLen := len(sc.tdef.Indexes[0])
        for i := 0; i < pkLen; i++ { values[i].Type = sc.tdef.Types[i] }
        decodeValues(key[4:], values[:pkLen])

        for i := pkLen; i < len(sc.tdef.Cols); i++ { values[i].Type = sc.tdef.Types[i] }
        decodeValues(val, values[pkLen:])

        rec.Cols = sc.tdef.Cols
        rec.Vals = values
    } else {
        // Secondary Index Scan: Extract Primary Key from the back of the key, then do a Point Query
        numEncodedVals := len(sc.tdef.Indexes[sc.index]) + len(sc.tdef.Indexes[0])
        encodedVals := make([]Value, numEncodedVals)
        
        decodeValues(key[4:], encodedVals)
        
        pkRec := Record{}
        pkCols := sc.tdef.Indexes[0]
        pkVals := encodedVals[len(sc.tdef.Indexes[sc.index]):] 
        for i, col := range pkCols {
            pkRec.Cols = append(pkRec.Cols, col)
            pkRec.Vals = append(pkRec.Vals, pkVals[i])
        }
        
        sc.tx.Get(sc.tdef.Name, &pkRec)
        *rec = pkRec
    }
}