Add EvalMap for pre-decoded JSON input (RecoLabs#2)

* Add EvalMap for pre-decoded JSON input

Add StreamEvaluator.EvalMap that accepts map[string]json.RawMessage
directly, avoiding the need to serialize data to []byte before
evaluation. Internally, DecodeRawMap converts each value individually
via DecodeJSON into an *OrderedMap, then delegates to the shared
evalInternal method (refactored out of EvalMany).

- Add DecodeRawMap to internal/evaluator/ordered_map.go
- Refactor EvalMany into evalInternal + thin EvalMany wrapper
- Add EvalMap with schemaKey support for GroupPlan caching
- Add resolveGjsonPath for O(1) map key lookup in fast paths
- Add tests for EvalMap (parity, multi-expr, nil/empty edge cases)

Signed-off-by: NirBarak-RecoLabs <nirb@recolabs.ai>

* Add EvalMap to README documentation

- Add EvalMap code example to Tier 3 StreamEvaluator section
- Add pre-decoded map input bullet to Streaming API key properties
- Update project structure to mention EvalMap in stream.go

Signed-off-by: NirBarak-RecoLabs <nirb@recolabs.ai>

---------

Signed-off-by: NirBarak-RecoLabs <nirb@recolabs.ai>

Author: NirBarak-RecoLabs

README.md

@@ -122,6 +122,9 @@ for i, result := range results {
         handleMatch(indices[i], result)
     }
 }
+
+// Alternative: pre-decoded map input (avoids re-serialization)
+results, _ = se.EvalMap(ctx, fieldMap, schemaKey, indices)
 ```
 
 ### Custom Function Registration
@@ -156,6 +159,7 @@ Hot Path (millions/day, lock-free)
 - **Schema-keyed caching** — the `GroupPlan` (merged paths, expression groupings, selective unmarshal targets) is computed once per schema key and reused immutably.
 - **Lock-free reads** — `BoundedCache` publishes an `atomic.Pointer` snapshot on every write; reads scan the snapshot without acquiring a lock. Writes are serialised by a mutex.
 - **Selective unmarshal** — full-path expressions unmarshal only the subtrees they need (e.g., just the `items` array from a 10KB event), not the entire document.
+- **Pre-decoded map input** — `EvalMap` accepts `map[string]json.RawMessage` directly, skipping full-document serialization when the caller already has individually-encoded fields. Fast paths resolve top-level keys via O(1) map lookup.
 - **Dynamic mutation** — `Replace`, `Remove`, and `Reset` methods allow modifying registered expressions at runtime with automatic cache invalidation.
 - **Observability** — implement `MetricsHook` to receive per-evaluation callbacks for cache hits/misses, eval latency, fast-path usage, and errors.
 
@@ -417,7 +421,7 @@ All standard regex features (character classes, quantifiers, alternation, groupi
 ```
 gnata/
 ├── gnata.go                     # Public API: Compile, Eval, EvalBytes, EvalWithVars, CustomFunc
-├── stream.go                    # StreamEvaluator, GroupPlan, EvalMany, MetricsHook
+├── stream.go                    # StreamEvaluator, GroupPlan, EvalMany, EvalMap, MetricsHook
 ├── bounded_cache.go             # Lock-free FIFO ring-buffer plan cache
 ├── deep_equal.go                # JSONata-compatible deep equality
 ├── internal/

func_fast.go

@@ -72,8 +72,8 @@ var funcFastHandlers = map[parser.FuncFastKind]funcFastHandler{
 	parser.FuncFastAverage:   evalFuncAverage,
 }
 
-func evalFuncBytes(f *parser.FuncFastPath, data json.RawMessage) (result any, handled bool, err error) {
-	r := gjson.GetBytes(data, f.Path)
+func evalFunc(f *parser.FuncFastPath, data json.RawMessage, mapData map[string]json.RawMessage) (result any, handled bool, err error) {
+	r := resolveGjsonPath(data, mapData, f.Path)
 	if !r.Exists() {
 		// Fall through to full evaluator — gjson doesn't auto-map through
 		// arrays, so the path might still resolve via the AST walker.

gnata.go

@@ -14,6 +14,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"slices"
+	"strings"
 
 	"github.com/recolabs/gnata/functions"
 	"github.com/recolabs/gnata/internal/evaluator"
@@ -206,12 +207,12 @@ func (e *Expression) EvalBytes(ctx context.Context, data json.RawMessage) (resul
 		// which handles auto-mapping through arrays correctly.
 	}
 	if e.cmpFast != nil {
-		if res, handled, evalErr := evalComparisonBytes(e.cmpFast, data); handled || evalErr != nil {
+		if res, handled, evalErr := evalComparison(e.cmpFast, data, nil); handled || evalErr != nil {
 			return res, evalErr
 		}
 	}
 	if e.funcFast != nil {
-		if res, handled, evalErr := evalFuncBytes(e.funcFast, data); handled || evalErr != nil {
+		if res, handled, evalErr := evalFunc(e.funcFast, data, nil); handled || evalErr != nil {
 			return res, evalErr
 		}
 	}
@@ -222,14 +223,44 @@ func (e *Expression) EvalBytes(ctx context.Context, data json.RawMessage) (resul
 	return e.Eval(ctx, v)
 }
 
-// evalComparisonBytes evaluates a pre-compiled comparison fast path against raw JSON bytes.
-// Returns (result, true, nil) on success. Returns (nil, false, nil) when the expression
-// cannot safely short-circuit (e.g. the LHS is a JSON array that requires auto-mapping),
-// signalling the caller to fall back to full evaluation.
+// resolveGjsonPath resolves a gjson path from either raw bytes or a pre-decoded map.
+// When data is available (EvalMany), it delegates to gjson.GetBytes on the full blob.
+// When mapData is available (EvalMap), it does an O(1) map lookup for the top-level
+// key, then uses gjson on the nested value bytes — skipping sibling keys entirely.
+// Paths containing gjson special characters fall through to return an empty result,
+// letting the caller fall back to full AST evaluation.
+func resolveGjsonPath(data json.RawMessage, mapData map[string]json.RawMessage, path string) gjson.Result {
+	if data != nil {
+		return gjson.GetBytes(data, path)
+	}
+	if mapData == nil {
+		return gjson.Result{}
+	}
+	if strings.ContainsAny(path, `\#*?@`) {
+		return gjson.Result{}
+	}
+	key, rest, hasDot := strings.Cut(path, ".")
+	raw, ok := mapData[key]
+	if !ok {
+		return gjson.Result{}
+	}
+	if !hasDot {
+		return gjson.ParseBytes(raw)
+	}
+	return gjson.GetBytes(raw, rest)
+}
+
+// evalComparison evaluates a pre-compiled comparison fast path against raw JSON
+// bytes or a pre-decoded map. Returns (result, true, nil) on success. Returns
+// (nil, false, nil) when the expression cannot safely short-circuit (e.g. the
+// LHS is a JSON array that requires auto-mapping), signalling the caller to
+// fall back to full evaluation.
 //
 //nolint:unparam // err is part of the funcFastHandler contract; always nil for now
-func evalComparisonBytes(c *parser.ComparisonFastPath, data json.RawMessage) (result any, handled bool, err error) {
-	lhs := gjson.GetBytes(data, c.LHSPath)
+func evalComparison(
+	c *parser.ComparisonFastPath, data json.RawMessage, mapData map[string]json.RawMessage,
+) (result any, handled bool, err error) {
+	lhs := resolveGjsonPath(data, mapData, c.LHSPath)
 	if !lhs.Exists() {
 		// gjson couldn't resolve the path. This could be because the path is
 		// truly undefined OR because an intermediate element is a JSON array

internal/evaluator/ordered_map.go

@@ -3,6 +3,7 @@ package evaluator
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"maps"
 	"slices"
 )
@@ -147,6 +148,21 @@ func DecodeJSON(b json.RawMessage) (any, error) {
 	return decodeValue(dec)
 }
 
+// DecodeRawMap converts a map of field names to raw JSON values into an
+// *OrderedMap by decoding each value individually via DecodeJSON. Objects
+// in the values preserve key insertion order, consistent with DecodeJSON.
+func DecodeRawMap(m map[string]json.RawMessage) (*OrderedMap, error) {
+	om := NewOrderedMapWithCapacity(len(m))
+	for key, raw := range m {
+		val, err := DecodeJSON(raw)
+		if err != nil {
+			return nil, fmt.Errorf("decode key %q: %w", key, err)
+		}
+		om.Set(key, val)
+	}
+	return om, nil
+}
+
 func decodeValue(dec *json.Decoder) (any, error) {
 	tok, err := dec.Token()
 	if err != nil {

stream.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/recolabs/gnata/internal/evaluator"
 	"github.com/recolabs/gnata/internal/parser"
-	"github.com/tidwall/gjson"
 )
 
 // StreamEvaluator manages compiled expressions for high-throughput streaming evaluation.
@@ -216,6 +215,25 @@ func (se *StreamEvaluator) Reset() {
 // Returns results[i] = evaluation of expressions[exprIndices[i]], or nil for undefined.
 func (se *StreamEvaluator) EvalMany(
 	ctx context.Context, data json.RawMessage, schemaKey string, exprIndices []int,
+) ([]any, error) {
+	return se.evalInternal(ctx, data, nil, nil, schemaKey, exprIndices)
+}
+
+// EvalMap evaluates the specified expressions against a map of raw JSON values.
+//   - data: map of field names to raw JSON-encoded values (decoded individually).
+//   - schemaKey: external key identifying the event schema. On first encounter, builds
+//     and caches a GroupPlan. Subsequent calls are lock-free. Pass "" to disable caching.
+//   - exprIndices: which compiled expressions to evaluate.
+//
+// Returns results[i] = evaluation of expressions[exprIndices[i]], or nil for undefined.
+func (se *StreamEvaluator) EvalMap(
+	ctx context.Context, data map[string]json.RawMessage, schemaKey string, exprIndices []int,
+) ([]any, error) {
+	return se.evalInternal(ctx, nil, nil, data, schemaKey, exprIndices)
+}
+
+func (se *StreamEvaluator) evalInternal(
+	ctx context.Context, data json.RawMessage, preparsed any, mapData map[string]json.RawMessage, schemaKey string, exprIndices []int,
 ) (results []any, err error) {
 	defer recoverEvalPanic(&err)
 	if len(exprIndices) == 0 {
@@ -247,7 +265,7 @@ func (se *StreamEvaluator) EvalMany(
 	}
 
 	results = make([]any, len(exprIndices))
-	batch := evalBatch{se: se, plan: plan, data: data}
+	batch := evalBatch{se: se, plan: plan, data: data, mapData: mapData, parsed: preparsed, parseAttempted: preparsed != nil}
 	for i, idx := range exprIndices {
 		if err := ctx.Err(); err != nil {
 			return nil, err
@@ -275,6 +293,7 @@ type evalBatch struct {
 	se             *StreamEvaluator
 	plan           *GroupPlan
 	data           json.RawMessage
+	mapData        map[string]json.RawMessage
 	parsed         any
 	parsedErr      error
 	parseAttempted bool
@@ -300,18 +319,17 @@ func (b *evalBatch) evalSingleExpr(ctx context.Context, i, idx int, expr *Expres
 // or (nil, true, err) on error.
 func (b *evalBatch) tryFastPaths(i, idx int, start time.Time) (result any, done bool, err error) {
 	if b.plan != nil && i < len(b.plan.ExprFastPath) && b.plan.ExprFastPath[i] && b.plan.FastPaths[i] != "" {
-		r := gjson.GetBytes(b.data, b.plan.FastPaths[i])
+		r := resolveGjsonPath(b.data, b.mapData, b.plan.FastPaths[i])
 		if r.Exists() {
 			if b.se.metrics != nil {
 				b.se.metrics.OnEval(idx, true, time.Since(start), nil)
 			}
 			return gjsonValueToAny(&r), true, nil
 		}
-		// gjson miss — fall through to full evaluator (handles array auto-mapping)
 	}
 
 	if b.plan != nil && i < len(b.plan.CmpFast) && b.plan.CmpFast[i] != nil {
-		if result, handled, err := evalComparisonBytes(b.plan.CmpFast[i], b.data); err != nil {
+		if result, handled, err := evalComparison(b.plan.CmpFast[i], b.data, b.mapData); err != nil {
 			if b.se.metrics != nil {
 				b.se.metrics.OnEval(idx, true, time.Since(start), err)
 			}
@@ -325,7 +343,7 @@ func (b *evalBatch) tryFastPaths(i, idx int, start time.Time) (result any, done
 	}
 
 	if b.plan != nil && i < len(b.plan.FuncFast) && b.plan.FuncFast[i] != nil {
-		if result, handled, err := evalFuncBytes(b.plan.FuncFast[i], b.data); err != nil {
+		if result, handled, err := evalFunc(b.plan.FuncFast[i], b.data, b.mapData); err != nil {
 			if b.se.metrics != nil {
 				b.se.metrics.OnEval(idx, true, time.Since(start), err)
 			}
@@ -345,7 +363,11 @@ func (b *evalBatch) tryFastPaths(i, idx int, start time.Time) (result any, done
 func (b *evalBatch) fullEval(ctx context.Context, idx int, expr *Expression, start time.Time) (any, error) {
 	if !b.parseAttempted {
 		b.parseAttempted = true
-		b.parsed, b.parsedErr = evaluator.DecodeJSON(b.data)
+		if len(b.mapData) > 0 {
+			b.parsed, b.parsedErr = evaluator.DecodeRawMap(b.mapData)
+		} else if len(b.data) > 0 {
+			b.parsed, b.parsedErr = evaluator.DecodeJSON(b.data)
+		}
 	}
 	if b.parsedErr != nil {
 		return nil, b.parsedErr