21187239412

Committed 20 Jan 2026 09:05PM UTC coverage: 71.335% (+0.2%) from 71.178%

Build # 21187239412

Build Type

push

github

Committed by

freeeve

Commit Message

refactor(reader): extract helpers to reduce cognitive complexity

Extract helper functions from findTableForPrefix, searchNextBlocks,
and scanPrefixLoop to reduce cognitive complexity below the threshold of 15.

Run Details

41 of 50 new or added lines in 1 file covered. (82.0%)

242 existing lines in 4 files now uncovered.

5761 of 8076 relevant lines covered (71.33%)

404816.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.79

/cmd/tinykvs/shell_select.go

package main

import (
        "encoding/json"
        "fmt"
        "math"
        "os"
        "os/signal"
        "strconv"
        "strings"
        "sync/atomic"
        "syscall"
        "time"

        "github.com/blastrain/vitess-sqlparser/sqlparser"
        "github.com/freeeve/msgpck"
        "github.com/freeeve/tinykvs"
)

// aggType represents an aggregation function type.
type aggType int

const (
        aggCount aggType = iota
        aggSum
        aggAvg
        aggMin
        aggMax
)

// aggregator holds state for a streaming aggregation.
type aggregator struct {
        typ   aggType
        field string // field path for sum/avg/min/max (empty for count)
        alias string // display name

        // streaming state
        count    int64
        sum      float64
        min      float64
        max      float64
        hasValue bool
}

func (a *aggregator) update(val tinykvs.Value) {
        a.count++

        if a.field == "" {
                return // count() doesn't need field extraction
        }

        // Extract numeric value from field
        num, ok := a.extractNumeric(val)
        if !ok {
                return
        }

        a.sum += num
        if !a.hasValue {
                a.min = num
                a.max = num
                a.hasValue = true
        } else {
                if num < a.min {
                        a.min = num
                }
                if num > a.max {
                        a.max = num
                }
        }
}

func (a *aggregator) extractNumeric(val tinykvs.Value) (float64, bool) {
        // For non-record types, use the value directly
        if a.field == "" {
                switch val.Type {
                case tinykvs.ValueTypeInt64:
                        return float64(val.Int64), true
                case tinykvs.ValueTypeFloat64:
                        return val.Float64, true
                default:
                        return 0, false
                }
        }

        // For record/msgpack types, extract the field
        var record map[string]any
        var err error
        switch val.Type {
        case tinykvs.ValueTypeRecord:
                record = val.Record
        case tinykvs.ValueTypeMsgpack:
                record, err = msgpck.UnmarshalMapStringAny(val.Bytes, false)
                if err != nil {
                        return 0, false
                }
        default:
                return 0, false
        }
        if record == nil {
                return 0, false
        }

        fieldVal, ok := extractNestedField(record, a.field)
        if !ok {
                return 0, false
        }

        switch v := fieldVal.(type) {
        case int:
                return float64(v), true
        case int8:
                return float64(v), true
        case int16:
                return float64(v), true
        case int32:
                return float64(v), true
        case int64:
                return float64(v), true
        case uint:
                return float64(v), true
        case uint8:
                return float64(v), true
        case uint16:
                return float64(v), true
        case uint32:
                return float64(v), true
        case uint64:
                return float64(v), true
        case float32:
                return float64(v), true
        case float64:
                return v, true
        default:
                return 0, false
        }
}

func (a *aggregator) result() string {
        switch a.typ {
        case aggCount:
                return fmt.Sprintf("%d", a.count)
        case aggSum:
                if !a.hasValue {
                        return "NULL"
                }
                return formatNumber(a.sum)
        case aggAvg:
                if a.count == 0 || !a.hasValue {
                        return "NULL"
                }
                return formatNumber(a.sum / float64(a.count))
        case aggMin:
                if !a.hasValue {
                        return "NULL"
                }
                return formatNumber(a.min)
        case aggMax:
                if !a.hasValue {
                        return "NULL"
                }
                return formatNumber(a.max)
        default:
                return "NULL"
        }
}

func formatNumber(f float64) string {
        if f == math.Trunc(f) {
                return fmt.Sprintf("%.0f", f)
        }
        return fmt.Sprintf("%g", f)
}

// valueFilter represents a filter condition on a value field
type valueFilter struct {
        field    string // e.g., "ttl", "name.first"
        operator string // "=", "like", ">", "<", ">=", "<="
        value    string // the comparison value
}

func (vf *valueFilter) matches(val tinykvs.Value) bool {
        // Extract the record
        var record map[string]any
        var err error
        switch val.Type {
        case tinykvs.ValueTypeRecord:
                record = val.Record
        case tinykvs.ValueTypeMsgpack:
                record, err = msgpck.UnmarshalMapStringAny(val.Bytes, false)
                if err != nil {
                        return false
                }
        default:
                return false
        }

        // Get field value
        fieldVal, ok := extractNestedField(record, vf.field)
        if !ok {
                return false
        }

        fieldStr := fmt.Sprintf("%v", fieldVal)

        switch vf.operator {
        case "=":
                return fieldStr == vf.value
        case "!=", "<>":
                return fieldStr != vf.value
        case "like":
                // Only prefix matching supported
                if strings.HasSuffix(vf.value, "%") {
                        prefix := vf.value[:len(vf.value)-1]
                        return strings.HasPrefix(fieldStr, prefix)
                }
                return fieldStr == vf.value
        case ">":
                return fieldStr > vf.value
        case "<":
                return fieldStr < vf.value
        case ">=":
                return fieldStr >= vf.value
        case "<=":
                return fieldStr <= vf.value
        default:
                return true
        }
}

// selectContext holds state for a SELECT query execution.
type selectContext struct {
        keyEquals    string
        keyPrefix    string
        keyStart     string
        keyEnd       string
        valueFilters []*valueFilter
        limit        int
        fields       []string
        aggs         []*aggregator
        headers      []string
        orderBy      []SortOrder
        scanned      int64
        scanStats    tinykvs.ScanStats
        matchCount   int
        bufferedRows [][]string
        lastProgress time.Time
        startTime    time.Time
        interrupted  int32
        scanErr      error
}

func (s *Shell) handleSelect(stmt *sqlparser.Select, orderBy []SortOrder) {
        ctx := s.parseSelectStatement(stmt, orderBy)

        sigChan := setupInterruptHandler(&ctx.interrupted)
        defer signal.Stop(sigChan)

        progressCallback := ctx.createProgressCallback()
        processRow := ctx.createRowProcessor()
        safeProcessRow := ctx.wrapRowProcessor(processRow)

        err := s.executeScan(ctx, safeProcessRow, progressCallback)

        ctx.clearProgressLine()

        wasInterrupted := atomic.LoadInt32(&ctx.interrupted) != 0
        if wasInterrupted {
                fmt.Fprintf(os.Stderr, "\r%s\r", strings.Repeat(" ", 80))
                fmt.Println("^C")
        }

        if ctx.scanErr != nil {
                fmt.Printf("Scan error: %v\n", ctx.scanErr)
                return
        }
        if err != nil {
                fmt.Printf("Error: %v\n", err)
                return
        }

        ctx.renderResults()
        ctx.reportStatistics(wasInterrupted)
}

func (s *Shell) parseSelectStatement(stmt *sqlparser.Select, orderBy []SortOrder) *selectContext {
        ctx := &selectContext{
                limit:     100,
                orderBy:   orderBy,
                startTime: time.Now(),
        }

        ctx.fields, ctx.aggs = parseSelectExpressions(stmt.SelectExprs)

        if stmt.Limit != nil && stmt.Limit.Rowcount != nil {
                if val, ok := stmt.Limit.Rowcount.(*sqlparser.SQLVal); ok {
                        if n, err := strconv.Atoi(string(val.Val)); err == nil {
                                ctx.limit = n
                        }
                }
        }

        if stmt.Where != nil {
                s.parseWhere(stmt.Where.Expr, &ctx.keyEquals, &ctx.keyPrefix, &ctx.keyStart, &ctx.keyEnd, &ctx.valueFilters)
        }

        if len(ctx.fields) > 0 {
                ctx.headers = append([]string{"k"}, ctx.fields...)
        } else {
                ctx.headers = []string{"k", "v"}
        }

        return ctx
}

func parseSelectExpressions(exprs sqlparser.SelectExprs) ([]string, []*aggregator) {
        var fields []string
        var aggs []*aggregator

        for _, expr := range exprs {
                switch e := expr.(type) {
                case *sqlparser.AliasedExpr:
                        if agg := tryParseAggregate(e); agg != nil {
                                aggs = append(aggs, agg)
                        } else if field := tryParseColName(e); field != "" {
                                fields = append(fields, field)
                        }
                case *sqlparser.StarExpr:
                        fields = nil
                }
        }
        return fields, aggs
}

// tryParseAggregate attempts to parse an aliased expression as an aggregate function.
func tryParseAggregate(e *sqlparser.AliasedExpr) *aggregator {
        funcExpr, ok := e.Expr.(*sqlparser.FuncExpr)
        if !ok {
                return nil
        }
        return parseAggregateFunc(funcExpr)
}

// tryParseColName attempts to parse an aliased expression as a column name.
// Returns the field name with proper qualification, or empty string if not a column.
func tryParseColName(e *sqlparser.AliasedExpr) string {
        col, ok := e.Expr.(*sqlparser.ColName)
        if !ok {
                return ""
        }
        qualifier := strings.ToLower(col.Qualifier.Name.String())
        fieldName := col.Name.String()
        if qualifier == "v" {
                return fieldName
        }
        if qualifier != "" && qualifier != "kv" {
                return qualifier + "." + fieldName
        }
        return ""
}

func setupInterruptHandler(interrupted *int32) chan os.Signal {
        sigChan := make(chan os.Signal, 1)
        signal.Notify(sigChan, syscall.SIGINT)
        go func() {
                <-sigChan
                atomic.StoreInt32(interrupted, 1)
        }()
        return sigChan
}

func (ctx *selectContext) createProgressCallback() tinykvs.ScanProgress {
        return func(stats tinykvs.ScanStats) bool {
                if atomic.LoadInt32(&ctx.interrupted) != 0 {
                        return false
                }
                ctx.scanStats = stats
                if time.Since(ctx.lastProgress) > time.Second {
                        elapsed := time.Since(ctx.startTime)
                        rate := int64(float64(stats.KeysExamined) / elapsed.Seconds())
                        fmt.Fprintf(os.Stderr, "\rScanned %s keys (%s blocks) in %s (%s keys/sec)...    ",
                                formatIntCommas(stats.KeysExamined), formatIntCommas(stats.BlocksLoaded),
                                formatDuration(elapsed), formatIntCommas(rate))
                        ctx.lastProgress = time.Now()
                }
                return true
        }
}

func (ctx *selectContext) createRowProcessor() func([]byte, tinykvs.Value) bool {
        isAggregate := len(ctx.aggs) > 0
        hasOrderBy := len(ctx.orderBy) > 0

        return func(key []byte, val tinykvs.Value) bool {
                if ctx.isInterrupted() {
                        return false
                }
                ctx.scanned++
                ctx.maybeShowFilterProgress()

                if !isAggregate && !hasOrderBy && ctx.matchCount >= ctx.limit {
                        return false
                }
                if !ctx.matchesFilters(val) {
                        return true
                }
                ctx.processMatch(key, val, isAggregate)
                return true
        }
}

// isInterrupted checks if query execution was interrupted.
func (ctx *selectContext) isInterrupted() bool {
        return atomic.LoadInt32(&ctx.interrupted) != 0
}

// maybeShowFilterProgress prints progress if filters are active and enough time passed.
func (ctx *selectContext) maybeShowFilterProgress() {
        if len(ctx.valueFilters) > 0 && time.Since(ctx.lastProgress) > time.Second {
                ctx.printFilterProgress()
        }
}

// matchesFilters checks if value passes all value filters.
func (ctx *selectContext) matchesFilters(val tinykvs.Value) bool {
        for _, vf := range ctx.valueFilters {
                if !vf.matches(val) {
                        return false
                }
        }
        return true
}

// processMatch handles a matching row by updating aggregates or buffering the row.
func (ctx *selectContext) processMatch(key []byte, val tinykvs.Value, isAggregate bool) {
        if isAggregate {
                for _, agg := range ctx.aggs {
                        agg.update(val)
                }
        } else {
                row := extractRowFields(key, val, ctx.fields)
                ctx.bufferedRows = append(ctx.bufferedRows, row)
                ctx.matchCount++
        }
}

func (ctx *selectContext) printFilterProgress() {
        elapsed := time.Since(ctx.startTime)
        rate := int64(float64(ctx.scanned) / elapsed.Seconds())
        fmt.Fprintf(os.Stderr, "\rScanned %s keys (%s blocks) in %s (%s keys/sec), found %d matches...    ",
                formatIntCommas(ctx.scanned), formatIntCommas(ctx.scanStats.BlocksLoaded),
                formatDuration(elapsed), formatIntCommas(rate), ctx.matchCount)
        ctx.lastProgress = time.Now()
}

func (ctx *selectContext) wrapRowProcessor(processRow func([]byte, tinykvs.Value) bool) func([]byte, tinykvs.Value) bool {
        return func(key []byte, val tinykvs.Value) bool {
                defer func() {
                        if r := recover(); r != nil {
                                ctx.scanErr = fmt.Errorf("panic processing key %x: %v", key[:min(8, len(key))], r)
                        }
                }()
                return processRow(key, val)
        }
}

func (s *Shell) executeScan(ctx *selectContext, processRow func([]byte, tinykvs.Value) bool, progress tinykvs.ScanProgress) error {
        if ctx.keyEquals != "" {
                return s.executePointLookup(ctx, processRow)
        }
        if ctx.keyPrefix != "" {
                var err error
                ctx.scanStats, err = s.store.ScanPrefixWithStats([]byte(ctx.keyPrefix), processRow, progress)
                return err
        }
        if ctx.keyStart != "" && ctx.keyEnd != "" {
                return s.store.ScanRange([]byte(ctx.keyStart), []byte(ctx.keyEnd), processRow)
        }
        var err error
        ctx.scanStats, err = s.store.ScanPrefixWithStats(nil, processRow, progress)
        return err
}

func (s *Shell) executePointLookup(ctx *selectContext, processRow func([]byte, tinykvs.Value) bool) error {
        val, err := s.store.Get([]byte(ctx.keyEquals))
        if err == tinykvs.ErrKeyNotFound {
                if len(ctx.aggs) > 0 {
                        printAggregateResults(ctx.aggs)
                } else {
                        printTable(ctx.headers, nil)
                        fmt.Printf("(0 rows)\n")
                }
                return nil
        }
        if err != nil {
                return err
        }
        processRow([]byte(ctx.keyEquals), val)
        return nil
}

func (ctx *selectContext) clearProgressLine() {
        hasValueFilters := len(ctx.valueFilters) > 0
        if (hasValueFilters || ctx.scanned > 10000) && ctx.scanned > 0 {
                fmt.Fprintf(os.Stderr, "\r%s\r", strings.Repeat(" ", 80))
        }
}

func (ctx *selectContext) renderResults() {
        if len(ctx.aggs) > 0 {
                printAggregateResults(ctx.aggs)
                return
        }

        if len(ctx.orderBy) > 0 {
                SortRows(ctx.headers, ctx.bufferedRows, ctx.orderBy)
        }

        if len(ctx.bufferedRows) > ctx.limit {
                ctx.bufferedRows = ctx.bufferedRows[:ctx.limit]
        }
        ctx.matchCount = len(ctx.bufferedRows)

        printTable(ctx.headers, ctx.bufferedRows)
}

func (ctx *selectContext) reportStatistics(wasInterrupted bool) {
        if ctx.scanned == 0 && ctx.scanStats.BlocksLoaded == 0 {
                fmt.Printf("(%d rows)\n", ctx.matchCount)
                return
        }

        elapsed := time.Since(ctx.startTime)
        rate := float64(ctx.scanned) / elapsed.Seconds()

        if wasInterrupted {
                fmt.Printf("(%d rows) - interrupted, scanned %s keys (%s blocks) in %s (%s keys/sec)\n",
                        ctx.matchCount, formatIntCommas(ctx.scanned),
                        formatIntCommas(ctx.scanStats.BlocksLoaded), formatDuration(elapsed), formatIntCommas(int64(rate)))
                return
        }

        blockDetails := fmt.Sprintf("%d blocks", ctx.scanStats.BlocksLoaded)
        if ctx.scanStats.BlocksCacheHit > 0 || ctx.scanStats.BlocksDiskRead > 0 {
                blockDetails = fmt.Sprintf("%d blocks (%d cache, %d disk)",
                        ctx.scanStats.BlocksLoaded, ctx.scanStats.BlocksCacheHit, ctx.scanStats.BlocksDiskRead)
        }
        tableDetails := ""
        if ctx.scanStats.TablesChecked > 0 {
                tableDetails = fmt.Sprintf(", %d/%d tables", ctx.scanStats.TablesAdded, ctx.scanStats.TablesChecked)
        }
        fmt.Printf("(%d rows) scanned %s keys, %s%s, %s\n",
                ctx.matchCount, formatIntCommas(ctx.scanned), blockDetails, tableDetails, formatDuration(elapsed))
}

// printStreamingHeader prints column headers for streaming output
func printStreamingHeader(headers []string) {
        for i, h := range headers {
                if i > 0 {
                        fmt.Print("\t")
                }
                fmt.Print(h)
        }
        fmt.Println()
        // Print separator
        for i, h := range headers {
                if i > 0 {
                        fmt.Print("\t")
                }
                fmt.Print(strings.Repeat("-", len(h)))
        }
        fmt.Println()
}

// printStreamingRow prints a single row for streaming output
func printStreamingRow(row []string) {
        for i, cell := range row {
                if i > 0 {
                        fmt.Print("\t")
                }
                // Truncate long values
                if len(cell) > 60 {
                        fmt.Print(cell[:57] + "...")
                } else {
                        fmt.Print(cell)
                }
        }
        fmt.Println()
}

// printTable prints rows in DuckDB-style box format
func printTable(headers []string, rows [][]string) {
        if len(headers) == 0 {
                return
        }

        widths := calculateColumnWidths(headers, rows)

        printBoxLine(widths, "┌", "┬", "┐")
        printTableRow(headers, widths)
        printBoxLine(widths, "├", "┼", "┤")
        for _, row := range rows {
                printTableRow(row, widths)
        }
        printBoxLine(widths, "└", "┴", "┘")
}

// calculateColumnWidths computes column widths based on headers and data rows.
func calculateColumnWidths(headers []string, rows [][]string) []int {
        widths := make([]int, len(headers))
        for i, h := range headers {
                widths[i] = len(h)
        }
        for _, row := range rows {
                updateRowWidths(widths, row)
        }
        capWidths(widths, 50)
        return widths
}

// updateRowWidths updates widths based on a single row's cell lengths.
func updateRowWidths(widths []int, row []string) {
        for i, cell := range row {
                if i < len(widths) && len(cell) > widths[i] {
                        widths[i] = len(cell)
                }
        }
}

// capWidths limits all widths to the specified maximum.
func capWidths(widths []int, maxWidth int) {
        for i := range widths {
                if widths[i] > maxWidth {
                        widths[i] = maxWidth
                }
        }
}

// printTableRow prints a single table row with proper formatting.
func printTableRow(cells []string, widths []int) {
        fmt.Print("│")
        for i, w := range widths {
                cell := ""
                if i < len(cells) {
                        cell = cells[i]
                }
                fmt.Printf(" %-*s │", w, truncate(cell, w))
        }
        fmt.Println()
}

func printBoxLine(widths []int, left, mid, right string) {
        fmt.Print(left)
        for i, w := range widths {
                fmt.Print(strings.Repeat("─", w+2))
                if i < len(widths)-1 {
                        fmt.Print(mid)
                }
        }
        fmt.Println(right)
}

func truncate(s string, maxLen int) string {
        if len(s) <= maxLen {
                return s
        }
        if maxLen <= 3 {
                return s[:maxLen]
        }
        return s[:maxLen-3] + "..."
}

func parseAggregateFunc(funcExpr *sqlparser.FuncExpr) *aggregator {
        funcName := strings.ToLower(funcExpr.Name.String())

        var typ aggType
        switch funcName {
        case "count":
                typ = aggCount
        case "sum":
                typ = aggSum
        case "avg":
                typ = aggAvg
        case "min":
                typ = aggMin
        case "max":
                typ = aggMax
        default:
                return nil
        }

        agg := &aggregator{typ: typ}

        // Extract field argument for sum/avg/min/max
        if len(funcExpr.Exprs) > 0 {
                if aliased, ok := funcExpr.Exprs[0].(*sqlparser.AliasedExpr); ok {
                        if col, ok := aliased.Expr.(*sqlparser.ColName); ok {
                                qualifier := strings.ToLower(col.Qualifier.Name.String())
                                fieldName := col.Name.String()

                                if qualifier == "v" {
                                        agg.field = fieldName
                                } else if qualifier != "" && qualifier != "kv" {
                                        agg.field = qualifier + "." + fieldName
                                }
                        }
                }
        }

        // Build alias for display
        if agg.field != "" {
                agg.alias = fmt.Sprintf("%s(v.%s)", funcName, agg.field)
        } else {
                agg.alias = fmt.Sprintf("%s()", funcName)
        }

        return agg
}

func printAggregateResults(aggs []*aggregator) {
        // Print header
        headers := make([]string, len(aggs))
        for i, agg := range aggs {
                headers[i] = agg.alias
        }
        fmt.Println(strings.Join(headers, " | "))

        // Print separator
        seps := make([]string, len(aggs))
        for i, agg := range aggs {
                seps[i] = strings.Repeat("-", len(agg.alias))
        }
        fmt.Println(strings.Join(seps, "-+-"))

        // Print values
        values := make([]string, len(aggs))
        for i, agg := range aggs {
                result := agg.result()
                // Pad to match header width
                if len(result) < len(agg.alias) {
                        result = strings.Repeat(" ", len(agg.alias)-len(result)) + result
                }
                values[i] = result
        }
        fmt.Println(strings.Join(values, " | "))
}

// extractRowFields extracts field values as strings for tabular display
func extractRowFields(key []byte, val tinykvs.Value, fields []string) []string {
        keyStr := formatKey(key)

        if len(fields) == 0 {
                // SELECT * - return key and full value
                return []string{keyStr, formatValue(val)}
        }

        // Extract specific fields
        row := []string{keyStr}

        var record map[string]any
        var err error
        switch val.Type {
        case tinykvs.ValueTypeRecord:
                record = val.Record
        case tinykvs.ValueTypeMsgpack:
                record, err = msgpck.UnmarshalMapStringAny(val.Bytes, false)
                if err != nil {
                        // Can't decode, return NULLs for all fields
                        for range fields {
                                row = append(row, "NULL")
                        }
                        return row
                }
        default:
                // Non-record type, return NULLs for all fields
                for range fields {
                        row = append(row, "NULL")
                }
                return row
        }

        for _, field := range fields {
                if v, ok := extractNestedField(record, field); ok {
                        row = append(row, fmt.Sprintf("%v", v))
                } else {
                        row = append(row, "NULL")
                }
        }
        return row
}

func formatValue(val tinykvs.Value) string {
        switch val.Type {
        case tinykvs.ValueTypeInt64:
                return fmt.Sprintf("%d", val.Int64)
        case tinykvs.ValueTypeFloat64:
                return fmt.Sprintf("%f", val.Float64)
        case tinykvs.ValueTypeBool:
                return fmt.Sprintf("%t", val.Bool)
        case tinykvs.ValueTypeString, tinykvs.ValueTypeBytes:
                if len(val.Bytes) > 100 {
                        return string(val.Bytes[:100]) + "..."
                }
                return string(val.Bytes)
        case tinykvs.ValueTypeRecord:
                if val.Record == nil {
                        return "{}"
                }
                jsonBytes, _ := json.Marshal(val.Record)
                return string(jsonBytes)
        case tinykvs.ValueTypeMsgpack:
                record, err := msgpck.UnmarshalMapStringAny(val.Bytes, false)
                if err != nil {
                        return "(msgpack)"
                }
                jsonBytes, _ := json.Marshal(record)
                return string(jsonBytes)
        default:
                return "(unknown)"
        }
}

func (s *Shell) parseWhere(expr sqlparser.Expr, keyEquals, keyPrefix, keyStart, keyEnd *string, valueFilters *[]*valueFilter) {
        switch e := expr.(type) {
        case *sqlparser.ComparisonExpr:
                s.parseComparisonExpr(e, keyEquals, keyPrefix, keyStart, keyEnd, valueFilters)
        case *sqlparser.RangeCond:
                parseRangeCond(e, keyStart, keyEnd)
        case *sqlparser.AndExpr:
                s.parseWhere(e.Left, keyEquals, keyPrefix, keyStart, keyEnd, valueFilters)
                s.parseWhere(e.Right, keyEquals, keyPrefix, keyStart, keyEnd, valueFilters)
        }
}

// parseComparisonExpr handles a WHERE comparison expression.
func (s *Shell) parseComparisonExpr(e *sqlparser.ComparisonExpr, keyEquals, keyPrefix, keyStart, keyEnd *string, valueFilters *[]*valueFilter) {
        col, ok := e.Left.(*sqlparser.ColName)
        if !ok {
                return
        }

        qualifier := strings.ToLower(col.Qualifier.Name.String())
        colName := strings.ToLower(col.Name.String())

        if isKeyColumn(colName, qualifier) {
                parseKeyFilter(e, keyEquals, keyPrefix, keyStart, keyEnd)
        } else {
                parseValueFieldFilter(e, qualifier, colName, valueFilters)
        }
}

// isKeyColumn returns true if the column represents the key (k).
func isKeyColumn(colName, qualifier string) bool {
        return (colName == "k" && qualifier == "") || (colName == "k" && qualifier == "kv")
}

// parseKeyFilter handles key-based WHERE conditions.
func parseKeyFilter(e *sqlparser.ComparisonExpr, keyEquals, keyPrefix, keyStart, keyEnd *string) {
        val, isHexPrefix := extractValueForLike(e.Right, e.Operator)
        switch e.Operator {
        case "=":
                *keyEquals = val
        case "like":
                parseKeyLikeFilter(val, isHexPrefix, keyPrefix)
        case ">=":
                *keyStart = val
        case "<=":
                *keyEnd = val
        }
}

// parseKeyLikeFilter handles LIKE conditions on keys.
func parseKeyLikeFilter(val string, isHexPrefix bool, keyPrefix *string) {
        if isHexPrefix {
                *keyPrefix = val
                return
        }
        if strings.HasSuffix(val, "%") && !strings.Contains(val[:len(val)-1], "%") {
                *keyPrefix = val[:len(val)-1]
                return
        }
        fmt.Println("Warning: LIKE only supports prefix matching (e.g., 'prefix%' or x'14%')")
}

// parseValueFieldFilter handles value field WHERE conditions.
func parseValueFieldFilter(e *sqlparser.ComparisonExpr, qualifier, colName string, valueFilters *[]*valueFilter) {
        fieldName := resolveFieldName(qualifier, colName)
        val := extractValue(e.Right)
        *valueFilters = append(*valueFilters, &valueFilter{
                field:    fieldName,
                operator: e.Operator,
                value:    val,
        })
}

// resolveFieldName determines the field name from qualifier and column name.
func resolveFieldName(qualifier, colName string) string {
        if qualifier == "v" {
                return colName
        }
        if qualifier != "" && qualifier != "kv" {
                return qualifier + "." + colName
        }
        return colName
}

// parseRangeCond handles BETWEEN conditions.
func parseRangeCond(e *sqlparser.RangeCond, keyStart, keyEnd *string) {
        col, ok := e.Left.(*sqlparser.ColName)
        if !ok {
                return
        }
        if strings.ToLower(col.Name.String()) == "k" {
                *keyStart = extractValue(e.From)
                *keyEnd = extractValue(e.To)
        }
}

freeeve / tinykvs / 21187239412

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous