assay

README

Assay: High-Performance Data Sampling & Schema Inference Engine

assay is a lightweight, ultra-fast Go library designed to profile incoming data streams in real-time, infer their schema, and track properties like type probabilities and field requirement (optionality) on-the-fly.

Engineered for hot-path execution in high-throughput data planes, assay operates with sub-millisecond overhead and is capable of processing high-volume data streams with sub-microsecond ingestion latency.

---

✨ Key Features

• Dynamic Schema Inference: Infers root-level and deeply nested object/array schemas on raw JSON bytes or native Go values (structs, maps).
• Type Probability Tracking: Computes type distribution percentage per key path (e.g., how often user.age is a string vs. a number).
• Optionality Detection: Evaluates field requirement recursively relative to parent objects based on observed frequencies.
• Storage Decoupling: Fully storage-agnostic. Define a simple interface to persist statistics to a local sharded in-memory cache, Redis, or Capacitor.
• High Performance: Implements zero-allocation path traversal, local write-back aggregation buffers, and streaming JSON token parsing to minimize GC pressure.
• Safety First: Prevents stack overflows and Out-Of-Memory (OOM) situations via maximum depth limits and path cardinality caps.

---

📦 Installation

go get github.com/cuprite-io/assay

---

🚀 Public API Design

Assay exposes a minimal public surface, keeping internal data structures and parsing state private.

Sampler & Config

To run Assay, construct a Sampler passing a StatsBackend and configuration:

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/cuprite-io/assay"
)

func main() {
	// 1. Initialize your storage backend (e.g., Capacitor, Redis, In-Memory)
	backend := NewMyCustomBackend()

	// 2. Configure Sampler limits
	cfg := assay.Config{
		MaxDepth:      32,         // Nesting limit to prevent stack exhaustion
		MaxPaths:      1000,       // Max unique paths to prevent cardinality explosion (OOM)
		FlushInterval: 100 * time.Millisecond, // Accumulation window for local writes
	}

	sampler := assay.NewSampler(backend, cfg)
	defer sampler.Close()

	ctx := context.Background()

	// 3. Sample incoming payloads
	jsonBytes := []byte(`{"user": {"name": "Alice", "tags": ["admin"]}}`)
	err := sampler.Sample(ctx, "api-endpoint-v1", jsonBytes)
	if err != nil {
		panic(err)
	}

	// 4. Retrieve computed schema tree on-demand
	schema, err := sampler.GetSchema(ctx, "api-endpoint-v1")
	if err != nil {
		panic(err)
	}
	fmt.Printf("Dominant Type of user.name: %s\n", schema.Children["user"].Children["name"].Type)
}

---

🔌 Decoupled Storage Interface (StatsBackend)

To plug Assay into your cache layer, implement the StatsBackend interface. Because Go utilizes structural typing, any struct implementing MapIncrementBy, MapGetAll, and Delete fits this contract automatically.

type StatsBackend interface {
	MapIncrementBy(ctx context.Context, key, field string, delta float64) (float64, error)
	MapGetAll(ctx context.Context, key string) (map[string]string, error)
	Delete(ctx context.Context, key string) error
}

Direct Integration with Capacitor

Since Capacitor already implements MapIncrementBy and MapGetAll with these exact signatures, a *capacitor.Capacitor instance fits this contract out of the box.

No adapter or wrapper class is needed:

import (
	"context"
	"github.com/cuprite-io/assay"
	"github.com/cuprite-io/capacitor"
)

func main() {
	// Initialize Capacitor client
	capClient, _ := capacitor.New(...)

	// Pass capClient directly as the backend!
	sampler := assay.NewSampler(capClient, assay.Config{})
	
	// ...
}

---

🧮 Mathematical Model

Type Probability

For a path $p$, the probability of type $T$ is computed as: $$P(T \mid p) = \frac{\text{Count}(p, T)}{\sum_{T' \in \mathcal{T}} \text{Count}(p, T')}$$

Field Requirement

A field $p$ is marked as required (relative to its parent) if it is observed every time its parent object is present, and its value is never null: $$\text{IsRequired}(p) = \left( \text{Observations}(p) == \text{Observations}(parent(p)) \right) \land \left( \text{Count}(p, \text{Null}) == 0 \right)$$

---

📄 License

Assay is licensed under the Apache License, Version 2.0. See LICENSE for the full text.

Documentation

Index

• Variables
• type Config
• type DataType
  • func (d DataType) String() string
• type PathStatsSnapshot
• type PathUpdate
• type Sampler
  • func NewSampler(backend StatsBackend, cfg Config) *Sampler
  • func (s *Sampler) Close() error
  • func (s *Sampler) DeleteSchema(ctx context.Context, schemaID string) error
  • func (s *Sampler) GetSchema(ctx context.Context, schemaID string) (*SchemaNode, error)
  • func (s *Sampler) Sample(ctx context.Context, schemaID string, payload any) error
• type SchemaNode
• type StatsBackend

Variables

var ErrMaxDepthExceeded = errors.New("maximum JSON nesting depth exceeded")

ErrMaxDepthExceeded is returned when the payload exceeds the maximum depth limit.

Types

type Config

type Config struct {
	MaxDepth      int           // Prevents Stack Overflow (Default: 32)
	MaxPaths      int           // Prevents Cardinality Explosion (Default: 1000)
	FlushInterval time.Duration // Interval for buffering writes to stats backend (Default: 100ms)
}

Config defines execution and limits for safety.

type DataType

type DataType int

DataType represents the primitive JSON types.

const (
	TypeNull DataType = iota
	TypeString
	TypeNumber
	TypeBoolean
	TypeObject
	TypeArray
)

func (DataType) String

func (d DataType) String() string

String returns the string representation of DataType.

type PathStatsSnapshot

type PathStatsSnapshot struct {
	ObservedCount uint64
	TypeCounts    [6]uint64
}

PathStatsSnapshot represents the snapshot of a path's counters.

type PathUpdate

type PathUpdate struct {
	Path  string
	Type  DataType
	Count uint64
}

PathUpdate represents an aggregated increment update for a path and type.

type Sampler

type Sampler struct {
	// contains filtered or unexported fields
}

Sampler is the entrypoint coordinator.

func NewSampler

func NewSampler(backend StatsBackend, cfg Config) *Sampler

NewSampler initializes a new Sampler with a backend and configuration.

func (*Sampler) Close

func (s *Sampler) Close() error

Close flushes any pending stats and shuts down the background flushing worker. It returns an error if the shutdown or final flush timed out.

func (*Sampler) DeleteSchema

func (s *Sampler) DeleteSchema(ctx context.Context, schemaID string) error

DeleteSchema removes a schema ID and all its accumulated stats from local memory and deletes the schema's metrics from the persistent stats backend.

func (*Sampler) GetSchema

func (s *Sampler) GetSchema(ctx context.Context, schemaID string) (*SchemaNode, error)

GetSchema reconstructs and returns the computed schema tree from the backend metrics.

func (*Sampler) Sample

func (s *Sampler) Sample(ctx context.Context, schemaID string, payload any) error

Sample parses the incoming payload and updates stats in the local buffer. payload can be: []byte (JSON), map[string]any, or a Struct pointer.

type SchemaNode

type SchemaNode struct {
	Name        string                 `json:"name"`
	Path        string                 `json:"path"`
	Type        string                 `json:"type"`        // "string", "number", "mixed", "null", "object", etc.
	Required    bool                   `json:"required"`    // Evaluated against parent observation frequency
	Probability map[string]float64     `json:"probability"` // E.g., {"string": 0.95, "number": 0.05}
	Children    map[string]*SchemaNode `json:"children,omitempty"`
}

SchemaNode represents a reconstructed node in the schema tree.

type StatsBackend

type StatsBackend interface {
	MapIncrementBy(ctx context.Context, key, field string, delta float64) (float64, error)
	MapGetAll(ctx context.Context, key string) (map[string]string, error)
	Delete(ctx context.Context, key string) error
}

StatsBackend represents the abstraction layer for stats storage. It matches Capacitor's MapIncrementBy, MapGetAll, and Delete methods, enabling implicit integration.