discovery

Documentation

Index

• func CanonicalizeTimestamps(schema *TableSchema, data map[string]any)
• func IsNumericType(chType string) bool
• func IsStringType(chType string) bool
• func Validate(schema *TableSchema, data map[string]any) error
• type Column
  • func (c *Column) TimeParser() func(v any) (time.Time, bool)
• type NumericStorage
  • func NumericStorageOf(chType string) (NumericStorage, bool)
• type SchemaRegistry
  • func NewSchemaRegistry(conn driver.Conn, database string, refreshInterval time.Duration, ...) *SchemaRegistry
  • func (sr *SchemaRegistry) Get(name string) *TableSchema
  • func (sr *SchemaRegistry) List() []*TableSchema
  • func (sr *SchemaRegistry) Refresh(ctx context.Context) error
  • func (sr *SchemaRegistry) RetryRefresh(ctx context.Context, initialBackoff, maxBackoff time.Duration, ...) error
  • func (sr *SchemaRegistry) StartAutoRefresh(ctx context.Context)
• type TableSchema
  • func (ts *TableSchema) ColumnNames() []string

Functions

func CanonicalizeTimestamps

func CanonicalizeTimestamps(schema *TableSchema, data map[string]any)

CanonicalizeTimestamps rewrites every DateTime/DateTime64 column value in data to the canonical RFC 3339 UTC form, in place: zone-less values are read in the column's zone, else the server default — ClickHouse's own rule, so only the spelling changes, never the instant — and the fraction truncates to the column's precision to byte-match /v1/query. Everything else passes through verbatim (fail-open): absent/null values, unparseable values, columns without a spec, and instants outside the column kind's range (see rewritable).

func IsNumericType

func IsNumericType(chType string) bool

IsNumericType reports whether chType is a ClickHouse numeric type (integer, float, or decimal), unwrapping Nullable/LowCardinality modifiers first. The stream row-filter evaluator classifies such columns numeric-comparable, so ordering predicates (>, <) on numbers match ClickHouse (9 < 100).

func IsStringType

func IsStringType(chType string) bool

IsStringType reports whether chType is a ClickHouse String (unwrapping Nullable/LowCardinality). For String columns, byte comparison IS ClickHouse comparison — equality and lexicographic order alike — so the stream row-filter evaluator can compare them exactly. FixedString is deliberately excluded: its stored values are zero-padded to the declared width, so a byte comparison of an ingested value against a filter constant would not match ClickHouse.

func Validate

func Validate(schema *TableSchema, data map[string]any) error

Validate checks that the given data matches the table schema. It rejects unknown fields and checks type compatibility.

Types

type Column

type Column struct {
	Name       string `json:"name"`
	Type       string `json:"type"`
	IsNullable bool   `json:"is_nullable"`
	HasDefault bool   `json:"has_default"`
	// contains filtered or unexported fields
}

Column describes a single ClickHouse column.

func (*Column) TimeParser

func (c *Column) TimeParser() func(v any) (time.Time, bool)

TimeParser returns the mapping from one rendering of this DateTime/DateTime64 column's value to the instant ClickHouse would store: the same grammar and zone rule ingest canonicalization applies (parseTimestamp), the same range guard (rewritable — an out-of-range operand, which insert-time saturation would move, is refused), truncated to the column's precision exactly like the canonical wire form. nil when the column isn't a timestamp column with a resolved spec; such columns keep byte-equality semantics on the stream. The stream row-filter uses it (policy.ColumnSpec.ParseTime) so a filter constant in any accepted spelling — zone-less read in the column's zone, RFC 3339, Unix seconds — and the canonicalized payload compare as instants (#381), through one grammar that can't drift from ingest's.

type NumericStorage

type NumericStorage struct {
	Integer   bool
	IntBits   int
	Unsigned  bool
	FloatBits int
	Precision int
	Scale     int
}

NumericStorage describes how a ClickHouse numeric column stores a value — the narrowing AND range the stream row-filter must apply to BOTH comparison operands so its verdicts match the query path, where ClickHouse narrows the stored value at insert and the bound constant at compare, and errors the query outright on a constant outside the column's range. Each family carries its parameters: Integer + IntBits/Unsigned for Int*/UInt* (exact within the width's range), FloatBits (32/64) for Float*, Precision+Scale for Decimal*.

func NumericStorageOf

func NumericStorageOf(chType string) (NumericStorage, bool)

NumericStorageOf classifies chType's numeric storage model, unwrapping Nullable/LowCardinality. ok=false for non-numeric types AND for a Decimal whose precision/scale cannot be parsed — the caller must then refuse numeric comparison rather than compare under a guessed model (fail closed). system.columns always reports the two-argument canonical Decimal(P, S) form (Decimal32(4) is stored as Decimal(9, 4)); the shorthand widths are handled anyway for robustness, and a bare single-argument Decimal(P) is refused rather than misread.

type SchemaRegistry

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

SchemaRegistry discovers and caches ClickHouse table schemas.

func NewSchemaRegistry

func NewSchemaRegistry(conn driver.Conn, database string, refreshInterval time.Duration, logger *slog.Logger) *SchemaRegistry

NewSchemaRegistry creates a registry that discovers schemas from system.columns.

func (*SchemaRegistry) Get

func (sr *SchemaRegistry) Get(name string) *TableSchema

Get returns the schema for a table, or nil if not found.

func (*SchemaRegistry) List

func (sr *SchemaRegistry) List() []*TableSchema

List returns all discovered table schemas.

func (*SchemaRegistry) Refresh

func (sr *SchemaRegistry) Refresh(ctx context.Context) error

Refresh rebuilds the in-memory schema cache: it discovers the server's default time zone, queries system.columns, and precomputes timestamp column specs.

func (*SchemaRegistry) RetryRefresh

func (sr *SchemaRegistry) RetryRefresh(ctx context.Context, initialBackoff, maxBackoff time.Duration, onAttempt func(err error)) error

RetryRefresh repeatedly calls Refresh with exponential backoff until it succeeds or ctx is cancelled. onAttempt is invoked after each failed attempt with the resulting error, letting callers surface the latest diagnostic (e.g. via /livez) while the registry is still degraded.

The first attempt fires immediately. After a failure the loop sleeps for initialBackoff, then doubles up to maxBackoff between attempts. Returns nil on success or ctx.Err() on cancellation. Zero/negative bounds are clamped via clampBackoff rather than busy-looping.

func (*SchemaRegistry) StartAutoRefresh

func (sr *SchemaRegistry) StartAutoRefresh(ctx context.Context)

StartAutoRefresh runs a background goroutine that refreshes schemas at the configured interval. Blocks until ctx is cancelled.

type TableSchema

type TableSchema struct {
	Name    string   `json:"name"`
	Columns []Column `json:"columns"`
}

TableSchema holds the discovered schema for one ClickHouse table.

func (*TableSchema) ColumnNames

func (ts *TableSchema) ColumnNames() []string

ColumnNames returns the table's column names in their discovered order (system.columns position; see Refresh). The query builder uses this to expand a SELECT * into a role's allowed projection, so the order is stable and matches the physical column order. Returns an empty (non-nil) slice for a schema with no columns.