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, err := assay.NewSampler(backend, cfg)
	if err != nil {
		panic(err)
	}
	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)
}

---

📊 Observability & Metrics

Assay provides built-in metrics and error hooks to monitor the sampler's performance and health:

Config.OnError Callback

Provide an optional OnError callback in Config to handle background flush errors asynchronously:

cfg := assay.Config{
	OnError: func(err error) {
		log.Printf("Assay background error: %v", err)
	},
}

Retrieval of Live Stats

Call sampler.Stats() to query real-time execution statistics:

stats := sampler.Stats()
fmt.Printf("Active Schemas: %d\n", stats.ActiveSchemas)
fmt.Printf("Ingested Samples: %d\n", stats.IngestedSamples)
fmt.Printf("Dropped Samples (Resource Limits): %d\n", stats.DroppedSamples)
fmt.Printf("Rejected Samples (Parse Errors): %d\n", stats.RejectedSamples)
fmt.Printf("Flush Successes: %d\n", stats.FlushSuccesses)
fmt.Printf("Flush Failures: %d\n", stats.FlushFailures)

Manual Flushing

Use sampler.Flush() to force a synchronous flush of all accumulated stats in memory to the storage backend:

if err := sampler.Flush(); err != nil {
	log.Printf("Flush failed: %v", err)
}

---

🔌 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, err := assay.NewSampler(capClient, assay.Config{})
	if err != nil {
		panic(err)
	}
	defer sampler.Close()
	
	// ...
}

---

🧮 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

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

Constants

const Version = "1.0.0"

Version is the current version of the assay library.

Variables

var ErrClosed = errors.New("assay: sampler is closed")

ErrClosed is returned when an operation is performed on a closed sampler.

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

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

var ErrMaxSchemasExceeded = errors.New("maximum schema count exceeded")

ErrMaxSchemasExceeded is returned when the maximum number of active schemas is reached.

Types

type Config

type Config struct {
	MaxDepth         int           // Prevents Stack Overflow (Default: 32)
	MaxPaths         int           // Prevents Cardinality Explosion (Default: 1000)
	MaxSchemas       int           // Prevents unbounded schema creation (Default: 1000)
	MaxArrayElements int           // Max array/slice elements profiled (Default: 10)
	FlushInterval    time.Duration // Interval for buffering writes to stats backend (Default: 100ms. Set to -1 to disable periodic background flushing)
	FlushTimeout     time.Duration // Timeout for individual backend writes (Default: 3s)
	OnError          func(error)   // Optional callback to handle flush or background errors
}

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. Sampler must not be copied after creation.

func NewSampler

func NewSampler(backend StatsBackend, cfg Config) (*Sampler, error)

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 or failed.

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) Flush

func (s *Sampler) Flush() error

Flush forces an immediate synchronous flush of all locally accumulated statistics to the backend. It returns an error if any flush operation failed.

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.

func (*Sampler) Stats

func (s *Sampler) Stats() SamplerStats

Stats returns a snapshot of the Sampler's metrics.

type SamplerStats

type SamplerStats struct {
	ActiveSchemas   int64
	IngestedSamples uint64
	DroppedSamples  uint64
	RejectedSamples uint64
	FlushSuccesses  uint64
	FlushFailures   uint64
}

SamplerStats contains metrics and observability data for the Sampler.

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. Note: This structure is not safe for concurrent mutation. If multiple goroutines write to the returned tree, external synchronization is required.

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.