Documentation
¶
Index ¶
- Variables
- 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
- type NumericStorage
- type SchemaRegistry
- func (sr *SchemaRegistry) Get(name string) *TableSchema
- func (sr *SchemaRegistry) List() []*TableSchema
- func (sr *SchemaRegistry) Loaded() bool
- func (sr *SchemaRegistry) Lookup(name string) (*TableSchema, error)
- func (sr *SchemaRegistry) Refresh(ctx context.Context) error
- func (sr *SchemaRegistry) RetryRefresh(ctx context.Context, initialBackoff, maxBackoff time.Duration, ...) error
- func (sr *SchemaRegistry) ServerVersion() string
- func (sr *SchemaRegistry) StartAutoRefresh(ctx context.Context)
- type Source
- type TableSchema
Constants ¶
This section is empty.
Variables ¶
var ( // ErrNotLoaded is Lookup's answer before the first successful Refresh: // the table may well exist, the registry just cannot say yet. ErrNotLoaded = errors.New("schema not loaded yet") // ErrUnknownTable is Lookup's answer for a table the loaded schema lacks. ErrUnknownTable = errors.New("unknown table") // ErrNoConnection is Refresh's answer when the connection getter yields // none: the tenant has no open ClickHouse pool. ErrNoConnection = errors.New("no ClickHouse connection") )
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 ¶
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 ¶
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.
Types ¶
type Column ¶
type Column struct {
Name string `json:"name"`
Type string `json:"type"`
IsNullable bool `json:"is_nullable"`
HasDefault bool `json:"has_default"`
// DefaultKind is how the column's default is declared, verbatim from
// system.columns.default_kind: "" (none), "DEFAULT", "MATERIALIZED",
// "ALIAS", or "EPHEMERAL". It is what decides whether a record may carry a
// value for the column — see IsInsertable. HasDefault is the boolean
// reading of the same field, so the two never disagree about whether a
// default exists.
DefaultKind string `json:"default_kind,omitempty"`
// DefaultExpression is the column's DEFAULT/MATERIALIZED/ALIAS expression
// as ClickHouse stores it (system.columns.default_expression), empty when
// the column declares none.
DefaultExpression string `json:"default_expression,omitempty"`
// Position is the column's 1-based ordinal in the table's declaration order
// (system.columns.position). Columns is already ordered by it; the field
// carries the ordinal itself so a caller holding a lone Column still knows
// where it sits.
Position uint64 `json:"position"`
// contains filtered or unexported fields
}
Column describes a single ClickHouse column.
func (Column) IsInsertable ¶ added in v0.2.0
IsInsertable reports whether a record may carry a value for this column — whether naming it in an INSERT's column list is legal.
ClickHouse refuses exactly two kinds, verified against a live server (26.6.3): a MATERIALIZED column is `Cannot insert column …, because it is MATERIALIZED column` (code 44, and insert_allow_materialized_columns defaults to 0), and an ALIAS column is `No such column …` (code 16) — it has no storage to write. Everything else takes a value: a plain column, a DEFAULT column, and an EPHEMERAL one, which exists precisely to be inserted into (it is insert-only — never stored, never selected).
func (*Column) TimeParser ¶
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 one tenant's ClickHouse table schemas.
func NewSchemaRegistry ¶
func NewSchemaRegistry(source Source, id tenant.ID, refreshInterval func(tenant.ID) time.Duration) *SchemaRegistry
NewSchemaRegistry creates the registry of tenant id, which discovers schemas from system.columns over the connection and database source yields.
func (*SchemaRegistry) Get ¶
func (sr *SchemaRegistry) Get(name string) *TableSchema
Get returns the schema for a table, or nil if not found — before the first refresh as much as for a table the schema lacks, which is the fail-closed reading the stream hub wants. A handler that answers 404 uses Lookup.
func (*SchemaRegistry) List ¶
func (sr *SchemaRegistry) List() []*TableSchema
List returns all discovered table schemas.
func (*SchemaRegistry) Loaded ¶ added in v0.2.0
func (sr *SchemaRegistry) Loaded() bool
Loaded reports whether a Refresh has ever succeeded: until one has, the registry cannot tell an unknown table from one it has not seen yet.
func (*SchemaRegistry) Lookup ¶ added in v0.2.0
func (sr *SchemaRegistry) Lookup(name string) (*TableSchema, error)
Lookup is Get for a caller that answers the two misses differently: ErrNotLoaded before the first successful Refresh (the table may exist — a 503 with Retry-After, not a 404), ErrUnknownTable for a table the loaded schema lacks.
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 and version, queries system.columns, attaches each table's DDL from system.tables, 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 a uniformly random time below the backoff ("full jitter"), which starts at initialBackoff and doubles up to maxBackoff, so instances retrying against one recovering ClickHouse spread over the whole window rather than firing in lockstep (#141). Returns nil on success or ctx.Err() on cancellation. Zero/negative bounds are clamped via clampBackoff rather than busy-looping.
func (*SchemaRegistry) ServerVersion ¶ added in v0.2.0
func (sr *SchemaRegistry) ServerVersion() string
ServerVersion returns the ClickHouse version string captured by the last successful Refresh, or "" before the first one.
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. The first refresh fires at a random point within the interval, so tenants adopted together do not refresh together; the cadence runs from there. The interval is re-read after every tick, so a changed setting applies from the next cycle — an in-flight wait finishes at the old cadence rather than resetting, which keeps a reload from ever deferring an imminent refresh.
A validated setting is at least a second, so a non-positive interval is a tenant the settings registry could not resolve, read as the zero value. NewTicker and Reset panic on one, so the loop keeps the cadence it has — unresolvedRefreshInterval when it has none yet — and picks the setting up on the first tick that resolves.
type Source ¶ added in v0.2.0
Source yields a tenant's connection and the database it discovers from, one snapshot: the database is the one the connection's own pool was opened for, so the schema discovered always describes the database the tenant's queries and inserts run against. A nil connection is a tenant with no open pool, which Refresh reports as ErrNoConnection.
type TableSchema ¶
type TableSchema struct {
Name string `json:"name"`
Columns []Column `json:"columns"`
// DDL is the table's CREATE TABLE statement as ClickHouse renders it
// (system.tables.create_table_query). Deliberately NOT serialized: the
// schema HTTP endpoint marshals TableSchema straight to the client, and an
// external-engine table (S3, MySQL, PostgreSQL, Kafka) renders its wiring
// there unconditionally — endpoint, bucket or host, database, username, S3
// access key id, ZooKeeper replica paths. The password specifically is
// rendered `[HIDDEN]` by ClickHouse >= ~23.9 (verified on 26.7.3), so the
// leak is the topology rather than the secret — except on an older server,
// or one with `display_secrets_in_show_and_select` enabled, where the
// password is in there too. Internal consumers read the field directly.
// Empty when the table was discovered from system.columns but had gone from
// system.tables by the time the second query ran — the two scans are not one
// snapshot, so a consumer must not treat an empty DDL as "no such table".
DDL string `json:"-"`
// contains filtered or unexported fields
}
TableSchema holds the discovered schema for one ClickHouse table. Columns is in declaration order (system.columns.position; see Refresh), which is the order positional row formats such as JSONCompactEachRow use.
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.
func (*TableSchema) InsertableColumnNames ¶ added in v0.2.0
func (ts *TableSchema) InsertableColumnNames() []string
InsertableColumnNames is InsertableColumns reduced to names, for the wire envelope's column list. Returns an empty (non-nil) slice when none qualify.
func (*TableSchema) InsertableColumns ¶ added in v0.2.0
func (ts *TableSchema) InsertableColumns() []Column
InsertableColumns returns the columns a record may carry, in declaration order — the positional contract for a row on the ingest path. Computed columns are left out because naming one in an INSERT is an error, not because they are uninteresting: they stay in Columns, so the schema endpoint and the query path still see the whole table.
func (*TableSchema) Lookup ¶ added in v0.2.0
func (ts *TableSchema) Lookup(name string) (Column, bool)
Lookup returns the named column and whether the table declares it. Matching is exact, as ClickHouse's own column resolution is. Linear over Columns, which is the right shape for the per-record call sites: schemas are small and the caller asks about one or two columns.
It returns the Column rather than a bool because "does the table have it" is rarely the whole question — a caller on the ingest path also has to know whether a record may carry a value for it (IsInsertable).
Source Files