Documentation
¶
Overview ¶
Package schema implements a common standard for describing data schemas within the domain of benthos. The intention for these schemas is to encourage schema conversion between multiple common formats such as avro, parquet, and so on.
Schema Identification and Caching ¶
To optimize performance when converting schemas between formats, this package provides fingerprinting and caching mechanisms:
- SchemaCache: A thread-safe cache for storing converted schemas
This allows downstream components to lazily perform conversions only once per unique schema identifier, avoiding redundant ToAny/FromAny serialization and expensive format translations.
Example usage:
// Create a cache for Parquet schema conversions
cache := schema.NewSchemaCache(func(c schema.Common) (ParquetSchema, error) {
return convertToParquet(c)
})
// First access converts and caches
parquet1, err := cache.GetOrConvert(mySchema)
// Second access uses cached result (no conversion)
parquet2, err := cache.GetOrConvert(mySchema)
Optimized Cache Lookups with Any Format ¶
When schemas are serialized to Any format (map[string]any), a fingerprint field is automatically included. This enables optimized cache lookups:
// Producer side: export schema (fingerprint included automatically)
schema := schema.Common{Type: schema.String, Name: "id"}
anySchema := schema.ToAny()
// ... send anySchema over network or store it ...
// Consumer side: optimized cache lookup
cache := schema.NewSchemaCache(convertFunc)
result, err := cache.GetOrConvertFromAny(anySchema)
// Fast path: if cached, avoids ParseFromAny and Fingerprint calculation
This optimization is particularly useful in scenarios where schemas are transmitted over the network or stored in external systems, as it eliminates the need to parse and recalculate fingerprints on cache hits.
Parameterised logical types ¶
Some types carry parameters beyond what the type identifier conveys. For example, a Decimal requires a precision and a scale. These parameters are attached to the Common schema via the Common.Logical field, which holds a LogicalParams struct. Only the field within LogicalParams that corresponds to the schema's Common.Type should be set.
Use Common.Validate to confirm a schema's parameters are internally consistent before relying on it.
Index ¶
- Constants
- func FormatBigDecimal(unscaled *big.Int, scale int32) (string, error)
- func FormatDecimal(unscaled *big.Int, scale int32) (string, error)
- func ParseBigDecimal(s string) (*big.Int, int32, error)
- func ParseDecimal(s string, scale int32) (*big.Int, error)
- type Cache
- type Common
- type CommonType
- type ConvertFunc
- type DecimalParams
- type LogicalParams
- type TimeOfDayParams
- type TimeUnit
- type TimestampParams
Examples ¶
Constants ¶
const ( DecimalMinPrecision int32 = 1 DecimalMaxPrecision int32 = 38 )
Decimal precision bounds. The upper bound matches the widest precision that can be represented losslessly across Avro, Parquet and Oracle NUMBER.
Variables ¶
This section is empty.
Functions ¶
func FormatBigDecimal ¶ added in v4.72.0
FormatBigDecimal renders an unscaled integer at the given scale as a canonical BigDecimal string. Output rules match FormatDecimal: leading minus for negatives only, no leading plus, no leading zeros aside from a single "0" before the decimal point, decimal point present iff scale > 0, exactly scale fractional digits emitted.
Unlike DecimalParams.Format, the BigDecimal schema imposes no fixed scale; callers pick the scale that matches the source value's natural precision. The scale parameter must be non-negative.
func FormatDecimal ¶ added in v4.72.0
FormatDecimal renders an unscaled integer as the canonical decimal string described by the package's value contract: a leading minus sign for negatives, no leading plus, no leading zeros aside from a single "0" before the decimal point, exactly scale fractional digits, and no scientific notation. The scale parameter must be non-negative.
Precision is not enforced here; use DecimalParams.Format when both precision and scale must be checked.
Examples for scale=4:
FormatDecimal(big.NewInt(12345), 4) // "1.2345" FormatDecimal(big.NewInt(0), 4) // "0.0000" FormatDecimal(big.NewInt(-1), 4) // "-0.0001" FormatDecimal(big.NewInt(12345), 0) // "12345"
func ParseBigDecimal ¶ added in v4.72.0
ParseBigDecimal interprets s as a decimal-shaped string and returns the unscaled integer alongside the scale recovered from the number of fractional digits in the input.
The accepted form matches ParseDecimal: lenient on non-canonical-but- unambiguous inputs (leading plus, leading zeros, missing integer part as in ".5"), strict on ambiguous or malformed inputs (scientific notation, multiple decimal points, whitespace, thousands separators, non-digit characters). Canonical form is enforced when values are re-emitted via FormatBigDecimal. Unlike ParseDecimal, the scale is recovered from the input rather than supplied by the caller.
The parser does not bound the input length. The underlying big.Int parse is super-linear, so callers exposing this directly to untrusted input should impose their own length cap.
func ParseDecimal ¶ added in v4.72.0
ParseDecimal interprets s as a decimal-shaped string and returns the unscaled integer at the given scale. Inputs with fewer fractional digits than scale are right-padded with zeros; inputs with more fractional digits than scale are rejected.
The parser is lenient: leading plus signs, leading zeros, and inputs missing the integer part (e.g. ".5") are accepted and normalised. The parser is strict about ambiguous or malformed inputs — scientific notation, multiple decimal points, whitespace, thousands separators, and non-digit characters are rejected. Canonical form is enforced when values are re-emitted via FormatDecimal. The scale parameter must be non-negative.
Precision is not enforced here; use DecimalParams.Parse when both precision and scale must be checked.
The parser does not bound the input length. The underlying big.Int parse is super-linear, so callers exposing this directly to untrusted input should impose their own length cap.
Types ¶
type Cache ¶ added in v4.65.0
type Cache[T any] struct { // contains filtered or unexported fields }
Cache provides a thread-safe cache for storing converted schemas. It uses schema fingerprints as keys to ensure that conversions only happen once per unique schema structure, avoiding redundant ToAny/FromAny serialization and expensive format translations (e.g., to Parquet).
Example usage:
type ParquetSchema struct { /* ... */ }
cache := schema.NewCache(func(c schema.Common) (ParquetSchema, error) {
// Convert Common schema to Parquet format
return convertToParquet(c)
})
schema := schema.Common{ /* ... */ }
parquetSchema, err := cache.GetOrConvert(schema)
Example ¶
ExampleCache demonstrates using Cache to avoid redundant conversions
package main
import (
"fmt"
"github.com/redpanda-data/benthos/v4/public/schema"
)
// ParquetSchema represents a target schema format (simplified example)
type ParquetSchema struct {
Name string
Fields []ParquetField
}
type ParquetField struct {
Name string
Type string
}
// convertToParquet simulates converting a Common schema to Parquet format
func convertToParquet(c schema.Common) (ParquetSchema, error) {
ps := ParquetSchema{Name: c.Name}
for _, child := range c.Children {
field := ParquetField{
Name: child.Name,
Type: child.Type.String(),
}
ps.Fields = append(ps.Fields, field)
}
return ps, nil
}
func main() {
// Create a cache for Parquet schema conversions
cache := schema.NewCache(convertToParquet)
// Define a common schema
userSchema := schema.Common{
Type: schema.Object,
Name: "user",
Children: []schema.Common{
{Type: schema.String, Name: "id"},
{Type: schema.String, Name: "email"},
{Type: schema.Int64, Name: "age", Optional: true},
},
}
// First conversion - will call convertToParquet
parquet1, err := cache.GetOrConvert(userSchema)
if err != nil {
panic(err)
}
fmt.Printf("First conversion: %s with %d fields\n", parquet1.Name, len(parquet1.Fields))
// Second conversion - will use cached result (no conversion)
parquet2, err := cache.GetOrConvert(userSchema)
if err != nil {
panic(err)
}
fmt.Printf("Cached result: %s with %d fields\n", parquet2.Name, len(parquet2.Fields))
// Check cache size
fmt.Printf("Cache size: %d\n", cache.Size())
}
Output: First conversion: user with 3 fields Cached result: user with 3 fields Cache size: 1
func NewCache ¶ added in v4.65.0
func NewCache[T any](convert ConvertFunc[T]) *Cache[T]
NewCache creates a new Cache with the provided conversion function. The conversion function will be called lazily when a schema is first accessed.
func (*Cache[T]) Clear ¶ added in v4.65.0
func (sc *Cache[T]) Clear()
Clear removes all entries from the cache.
func (*Cache[T]) GetOrConvert ¶ added in v4.65.0
GetOrConvert retrieves a converted schema from the cache, or converts and caches it if not already present. This method is thread-safe and ensures that the conversion function is only called once per unique schema structure.
The fingerprint is computed automatically from the provided schema.
func (*Cache[T]) GetOrConvertFromAny ¶ added in v4.65.0
GetOrConvertFromAny retrieves a converted schema from the cache, or converts and caches it if not already present. This method optimizes cache lookups when working with schemas in Any format (map[string]any).
Since ToAny() automatically includes a "fingerprint" field at the top level, this method:
- First attempts to retrieve the cached value using that fingerprint (fast path)
- Only parses the Any format and recalculates the fingerprint on cache miss
This optimization is particularly useful when schemas are frequently received in Any format, as it avoids the expensive ParseFromAny conversion and Fingerprint calculation on cache hits.
Usage example:
// Producer side: export schema (fingerprint included automatically)
schema := schema.Common{Type: schema.String, Name: "id"}
anySchema := schema.ToAny()
// ... send anySchema over network or store it ...
// Consumer side: optimized cache lookup
cache := schema.NewCache(convertFunc)
result, err := cache.GetOrConvertFromAny(anySchema)
// Fast path: if cached, avoids ParseFromAny and Fingerprint calculation
If the Any format does not include a fingerprint (e.g., from an older version), this method falls back to parsing the schema and calling GetOrConvert normally.
Example ¶
ExampleCache_GetOrConvertFromAny demonstrates optimized cache usage with Any format
package main
import (
"fmt"
"github.com/redpanda-data/benthos/v4/public/schema"
)
// ParquetSchema represents a target schema format (simplified example)
type ParquetSchema struct {
Name string
Fields []ParquetField
}
type ParquetField struct {
Name string
Type string
}
// convertToParquet simulates converting a Common schema to Parquet format
func convertToParquet(c schema.Common) (ParquetSchema, error) {
ps := ParquetSchema{Name: c.Name}
for _, child := range c.Children {
field := ParquetField{
Name: child.Name,
Type: child.Type.String(),
}
ps.Fields = append(ps.Fields, field)
}
return ps, nil
}
func main() {
cache := schema.NewCache(convertToParquet)
// Define a schema
userSchema := schema.Common{
Type: schema.Object,
Name: "user",
Children: []schema.Common{
{Type: schema.String, Name: "id"},
{Type: schema.String, Name: "email"},
{Type: schema.Int64, Name: "age"},
},
}
// Convert to Any format (fingerprint included automatically)
// This is typically done by the producer/sender
anySchema := userSchema.ToAny()
// First call: converts and caches (requires parsing Any -> Common)
result1, err := cache.GetOrConvertFromAny(anySchema)
if err != nil {
panic(err)
}
fmt.Printf("First call: %s with %d fields\n", result1.Name, len(result1.Fields))
// Second call: uses cached result (optimized - skips parsing and fingerprint calculation)
result2, err := cache.GetOrConvertFromAny(anySchema)
if err != nil {
panic(err)
}
fmt.Printf("Second call (cached): %s with %d fields\n", result2.Name, len(result2.Fields))
fmt.Printf("Cache size: %d\n", cache.Size())
}
Output: First call: user with 3 fields Second call (cached): user with 3 fields Cache size: 1
type Common ¶
type Common struct {
Name string
Type CommonType
Optional bool
Children []Common
// Logical holds parameters for parameterised types (e.g. Decimal). Only
// the field within LogicalParams that corresponds to Type should be
// populated; setting parameters that do not apply to the type is a
// validation error.
Logical *LogicalParams
}
Common schema is a neutral form that can be converted to and from other schemas. This is not intended to be a superset of all schema capabilites and instead focuses on compatibility and minimum viable translations between schemas.
func InferFromAny ¶
InferFromAny attempts to infer a common schema from any Go value. This process fails if the value, or any children of a provided map/slice, are not within the following subset of Go types: bool, int, int32, int64, float32, float64, encoding/json.Number, []byte, string, map[string]any, []any.
encoding/json.Number values are inferred as Int64 when they parse as an integer and as Float64 otherwise.
Decimal types (both Decimal and BigDecimal) cannot be inferred from generic Go values and must be constructed explicitly via NewDecimal or NewBigDecimal.
All values will be recorded as non-optional.
func NewBigDecimal ¶ added in v4.72.0
NewBigDecimal constructs a Common schema for a BigDecimal column — an arbitrary-precision decimal with no schema-level precision or scale commitment. Use this for sources where the column type does not carry fixed precision and scale (e.g. unparameterised Postgres NUMERIC, Oracle NUMBER without DATA_PRECISION, MongoDB Decimal128).
func NewDecimal ¶ added in v4.72.0
NewDecimal constructs a Common schema for a fixed-precision decimal column and validates the precision and scale bounds. It exists so callers can avoid the triple-nested LogicalParams/DecimalParams struct literal at the call site.
func ParseFromAny ¶
ParseFromAny deserializes a common schema from a generic Go value. The resulting schema is validated via Common.Validate before being returned.
func (*Common) EffectiveTimestamp ¶ added in v4.73.0
func (c *Common) EffectiveTimestamp() TimestampParams
EffectiveTimestamp returns the timestamp parameters for c, applying the legacy default ({Unit: TimeUnitMillis, AdjustToUTC: true}) when c.Logical is unset. It is only meaningful when c.Type == Timestamp; for other types the returned value should be ignored.
Format adapters that need to honour both pre-parameterised legacy schemas and richer schemas produced by newer decoders should consult this rather than peeking at c.Logical directly.
func (*Common) ToAny ¶
ToAny serializes the common schema into a generic Go value, with structured schemas being represented as map[string]any and []any. This could be further manipulated using generic mapping tools such as bloblang, before either bringing back into a Common representation or serializing into another format.
The serialized format includes a "fingerprint" field at the top level, which can be used to optimize cache lookups via SchemaCache.GetOrConvertFromAny, avoiding the need to parse the Any format and recalculate the fingerprint.
NOTE: Ironically, the schema for this serialization is not something that can be precisely represented as a Common schema. The Children field requires an Array of structured schema objects, which cannot be described accurately within the Common type system.
Example ¶
ExampleCommon_ToAny demonstrates that ToAny includes fingerprints
package main
import (
"fmt"
"github.com/redpanda-data/benthos/v4/public/schema"
)
func main() {
userSchema := schema.Common{
Type: schema.Object,
Name: "user",
Children: []schema.Common{
{Type: schema.String, Name: "id"},
{Type: schema.String, Name: "email"},
},
}
// Export to Any format (fingerprint included automatically)
anySchema := userSchema.ToAny()
// The result includes a "fingerprint" field at the top level
if m, ok := anySchema.(map[string]any); ok {
if fp, ok := m["fingerprint"].(string); ok {
fmt.Printf("Has fingerprint: %t\n", len(fp) > 0)
fmt.Printf("Fingerprint length: %d\n", len(fp))
}
}
// This format can be sent over the network, stored, etc.
// When received, GetOrConvertFromAny can use the fingerprint
// to optimize cache lookups
}
Output: Has fingerprint: true Fingerprint length: 64
func (*Common) Validate ¶ added in v4.72.0
Validate enforces the parameter invariants of parameterised types — Decimal precision/scale bounds — and the structural invariant that leaf types do not carry children. The container types (Object, Map, Array, Union) are the only types permitted to populate Common.Children. Validate recurses into Children.
Other structural invariants — for example that an Object has children, or that a Union has more than one child — are not currently enforced; the validation surface may grow as new logical types arrive.
Schemas constructed via ParseFromAny are validated automatically. Schemas constructed by struct literal should call Validate before being passed to converters or caches.
type CommonType ¶
type CommonType int
CommonType represents types supported by common schemas.
const ( Boolean CommonType = 1 Int32 CommonType = 2 Int64 CommonType = 3 Float32 CommonType = 4 Float64 CommonType = 5 String CommonType = 6 ByteArray CommonType = 7 Object CommonType = 8 Map CommonType = 9 Array CommonType = 10 Null CommonType = 11 Union CommonType = 12 Timestamp CommonType = 13 Any CommonType = 14 Decimal CommonType = 15 BigDecimal CommonType = 16 Date CommonType = 17 TimeOfDay CommonType = 18 UUID CommonType = 19 )
Supported common types
func (CommonType) String ¶
func (t CommonType) String() string
String returns a human readable string representation of the type.
type ConvertFunc ¶ added in v4.65.0
ConvertFunc is a function that converts a Common schema to a target format T. This function is called lazily when a schema is first accessed in the cache.
type DecimalParams ¶ added in v4.72.0
DecimalParams describes a fixed-precision decimal number.
Precision is the total number of significant digits and must be in [DecimalMinPrecision, DecimalMaxPrecision]. Scale is the number of digits to the right of the decimal point and must be in [0, Precision]. These constraints describe the lossless intersection across Avro, Parquet and Oracle NUMBER.
func (DecimalParams) Format ¶ added in v4.72.0
func (p DecimalParams) Format(unscaled *big.Int) (string, error)
Format renders the unscaled integer as a canonical decimal string at the configured scale, and rejects values whose magnitude exceeds the configured precision.
func (DecimalParams) Parse ¶ added in v4.72.0
func (p DecimalParams) Parse(s string) (*big.Int, error)
Parse interprets s as a canonical decimal string at the configured scale, and rejects values whose magnitude exceeds the configured precision.
func (DecimalParams) ValidateValue ¶ added in v4.72.0
func (p DecimalParams) ValidateValue(v *big.Int) error
ValidateValue confirms that the magnitude of v has no more significant digits than the configured precision. The configured precision and scale are not themselves validated by this method; use Common.Validate for that.
type LogicalParams ¶ added in v4.72.0
type LogicalParams struct {
Decimal *DecimalParams
Timestamp *TimestampParams
TimeOfDay *TimeOfDayParams
}
LogicalParams groups the optional parameter blocks that parameterised CommonType values may carry. Each parameterised type has its own field; at most one is expected to be non-nil for any given Common schema.
type TimeOfDayParams ¶ added in v4.73.0
TimeOfDayParams describes the precision and timezone semantics of a TimeOfDay schema (a wall-clock time with no date component). Unit selects the resolution; AdjustToUTC parallels the equivalent Parquet TIME flag and is rare outside Parquet/Postgres timetz.
Unlike TimestampParams, a TimeOfDay-typed schema must have non-nil LogicalParams.TimeOfDay — there is no historical default to fall back to.
type TimeUnit ¶ added in v4.73.0
type TimeUnit int
TimeUnit names the precision at which a Timestamp or TimeOfDay value is expressed. The zero value is invalid; use one of the named constants.
const ( TimeUnitSeconds TimeUnit = 1 TimeUnitMillis TimeUnit = 2 TimeUnitMicros TimeUnit = 3 TimeUnitNanos TimeUnit = 4 )
Supported time units.
func (TimeUnit) String ¶ added in v4.73.0
String returns a human-readable representation of the time unit, suitable for serialisation via Common.ToAny.
type TimestampParams ¶ added in v4.73.0
TimestampParams describes the precision and timezone semantics of a Timestamp schema. Unit selects the resolution at which the timestamp is expressed; AdjustToUTC distinguishes a UTC instant (true) from a civil / "local" datetime that carries no timezone offset (false).
A nil LogicalParams.Timestamp on a Timestamp-typed schema is permitted for backwards compatibility and is treated as {Unit: TimeUnitMillis, AdjustToUTC: true}; see Common.EffectiveTimestamp.
Source Files