Documentation
¶
Index ¶
- Constants
- func AsSlice(v any) ([]any, bool)
- func BuildFirstRowQuery(sourceType, table string) (string, error)
- func BuildIncrementalQuery(sourceType, table, idField string) (string, error)
- func CanonicalDriver(sourceType string) (string, bool)
- func ColumnTypeNames(rows *sql.Rows) []string
- func DecodeColumn(v any, typeName string) any
- func DecodeJSONColumn(raw []byte) any
- func DecodeNumericText(s string) (json.Number, bool)
- func DecodePGArray(raw []byte, typeName string) (any, bool)
- func DecodePGXValue(v any) any
- func DecodeValue(v any, isJSON bool) any
- func GetFromMapPath(m map[string]any, path string) any
- func IsJSONColumnType(databaseTypeName string) bool
- func IsNumericColumnType(databaseTypeName string) bool
- func IsPGArrayTypeName(databaseTypeName string) bool
- func ParameterizeTemplate(driver, tpl string, data map[string]any) (string, []any)
- func Placeholder(driver string, index int) string
- func QuoteIdent(driver, name string) (string, error)
- func RecordFromValues(names, typeNames []string, values []any) map[string]any
- func ScanRows(rows *sql.Rows) ([]map[string]any, error)
- func TemplateArgs(tpl string, data map[string]any) []any
- func TemplateArgsWith(tpl string, resolve Resolver) []any
- func ValidateIdent(name string) error
- type ColumnMapping
- type Resolver
- type TemplateBinding
Constants ¶
const DefaultMaxRows = 1000
DefaultMaxRows is the maximum number of rows ScanRows will fetch to prevent OOM.
const MaxListExpansion = 65535
MaxListExpansion bounds how many placeholders one token may expand into.
The element count comes from message data, which nothing upstream bounds, and each element costs a placeholder in the statement text and an argument in the bind list. PostgreSQL's wire protocol stops at 65535 parameters and SQL Server at 2100, so a list beyond this never reaches a server that would accept it -- but without the cap it is built in this process first, which is where a pathological array does its damage.
Variables ¶
This section is empty.
Functions ¶
func AsSlice ¶ added in v1.8.1
AsSlice coerces v into a slice of any so it can be expanded into a SQL value list. It reports false for anything that is a scalar to a database driver, including []byte and json.RawMessage, which are bytea/blob and JSON column values rather than lists.
func BuildFirstRowQuery ¶
BuildFirstRowQuery returns a dialect-correct query selecting any single row from a table, with no watermark and no ordering.
It is used for schema probes and for the first read of a table that has no cursor yet. Ordering is intentionally omitted: there is no cursor to advance, so "any row" is the intended semantics and a sort would be wasted work on a large table.
func BuildIncrementalQuery ¶
BuildIncrementalQuery returns a dialect-correct query that selects the single next row after a watermark, ordered ascending by idField. The watermark value is bound as parameter 1; callers must pass it as the only query argument.
Use this instead of hand-writing the query in a connector. Watermark polling looks trivial and is not: the row-limiting clause must be applied after the sort, and three of the four dialect families here spell that differently. Every connector that wrote its own got at least the Oracle case wrong.
Both identifiers are validated and quoted, so the result is injection-safe. An unknown sourceType is an error rather than a guess — emitting plausible SQL for the wrong dialect is the failure mode this helper exists to prevent.
func CanonicalDriver ¶
CanonicalDriver maps a user-facing source/sink type (e.g. "mssql", "postgres") to the actual database/sql driver name that must be passed to sql.Open. It is the single source of truth shared by connection opening and placeholder generation so the two can never drift apart. The second return value reports whether the type is backed by a generic SQL driver.
IMPORTANT: this mapping MUST stay in sync with how connections are opened (see registry.getOrOpenDB). For example "mssql" is opened with the microsoft/go-mssqldb "sqlserver" driver, which only accepts @pN placeholders.
func ColumnTypeNames ¶ added in v1.8.1
ColumnTypeNames reads the driver's own name for each column, in order.
Returns nil when the driver cannot say. Callers treat that as "no JSON columns", which is the behaviour every source had before JSON columns were decoded at all -- a driver that withholds type metadata must not change a pipeline's shape.
func DecodeColumn ¶ added in v1.10.0
DecodeColumn is the per-value rule, decided from the database's own name for the column's type.
This is the entry point for any caller that has the driver's type metadata. DecodeValue remains for the one that does not: the MySQL binlog reader knows only a go-mysql column enum, never a type name.
Deliberately narrow, and for the same reason in both branches: only a column the database itself calls JSON or an array is reshaped. Deciding from content would reshape every string that happens to look like one.
numeric becomes a json.Number, not a float64. The pgx-native path carries it exactly -- measured on PostgreSQL 18.4, a numeric(40,20) holding 1.00000000000000000001 marshals from pgtype.Numeric at full precision -- so converting to float64 would have flattened it to 1 and propagated a loss rather than closed a gap. json.Number keeps every digit and serialises as the bare number the native path produces, which is what makes the two agree.
Non-finite numerics stay text: PostgreSQL accepts NaN, Infinity and -Infinity, and DecodeNumericText refuses all three because json.Number does not validate and one such cell would fail the whole message's marshalling.
func DecodeJSONColumn ¶ added in v1.8.1
DecodeJSONColumn gives the bytes of a JSON-typed column the value they represent, so a `jsonb`/`json` column reaches transformations and sinks as an object rather than as a string that happens to contain one.
Only call this for columns the database itself types as JSON -- see IsJSONColumnType, or the source's own type metadata. Applied to a text column it would turn the string "123" into the number 123 and "true" into a boolean.
Bytes that do not parse are returned as a string. A column that cannot be decoded still has to reach the sink; dropping it, or failing the message, is how this class of bug cost data in the first place.
func DecodeNumericText ¶ added in v1.10.0
DecodeNumericText gives an exact decimal column the shape it has on the pgx-native path: a bare JSON number carrying every digit the database sent.
The bool reports whether it could. Text the JSON number grammar does not accept is returned unchanged, and that is not an edge case -- PostgreSQL numeric accepts NaN, Infinity and -Infinity, none of which has a JSON form. json.Number does not validate on construction, so converting one of those would build a value that fails json.Marshal, and a single such cell turns a whole message into a marshalling error rather than a row with one odd column.
Text is also the better answer for those three on the merits: measured on PostgreSQL 18.4, pgtype.Numeric marshals both infinities to 0, so converging on the native shape there would replace a correct string with silent corruption.
func DecodePGArray ¶ added in v1.10.0
DecodePGArray turns a PostgreSQL array literal into a list.
Returns ok=false when the type name is not one pgx knows or the literal does not parse, and the caller keeps the text it already had. A malformed value must stay a malformed value: returning nil would turn it into a missing one, which is the bug being fixed, not a smaller version of it.
func DecodePGXValue ¶ added in v1.10.0
DecodePGXValue is the per-value rule the native pgx read paths apply.
It subsumes the `[]byte -> string` conversion those call sites already did, so it is a drop-in for that line rather than an addition to it.
func DecodeValue ¶ added in v1.8.1
DecodeValue is the per-value rule RecordFromValues applies, exposed for the callers that need to keep hold of a single column as they go. Both carriers matter. database/sql drivers hand text over as []byte, but go-mysql has already converted a MYSQL_TYPE_JSON value to a string by the time the binlog handler sees it -- asserting only on []byte left that column a string, and no unit test built from hand-written []byte could notice.
func GetFromMapPath ¶ added in v1.8.1
GetFromMapPath resolves a dotted path in a nested map[string]any.
func IsJSONColumnType ¶ added in v1.8.1
IsJSONColumnType reports whether a database's own name for a column type means the column holds JSON.
Deliberately narrow. SQLite has no JSON type and SQL Server had none before 2025, so JSON there lives in TEXT/NVARCHAR columns; treating those as JSON would reshape every string that happens to parse, which is a far larger change than the bug being fixed.
func IsNumericColumnType ¶ added in v1.10.0
IsNumericColumnType reports whether a database's own name for a column type means the column holds an exact decimal.
Deliberately narrow, for the same reason IsJSONColumnType is: only a column the database itself names as a decimal is reshaped. Deciding from content would turn every text column that happens to hold "42" into a number.
FLOAT8, REAL and DOUBLE PRECISION are excluded because every driver here already delivers them as a float64 -- they were never text and have no precision left to preserve. MONEY is excluded because its text carries a currency symbol ("$19.99"), which is not a JSON number at all; it already agrees across both read paths as a string.
func IsPGArrayTypeName ¶ added in v1.10.0
IsPGArrayTypeName reports whether a database's own name for a column type means the column holds a PostgreSQL array.
pgx names an array by prefixing its element type with an underscore, so an int[] column reports "_INT4" and a text[] column "_TEXT". Dimensionality is not in the name: int[] and int[][] both report "_INT4".
A type pgx has no registered name for reports its bare OID instead -- money comes back as "790" -- so a name-keyed rule has to tolerate names that describe nothing.
func ParameterizeTemplate ¶ added in v1.8.1
ParameterizeTemplate replaces all {{ ... }} tokens in the SQL template with driver-specific placeholders and returns the parameterized SQL text and a corresponding args slice. Token content should be either a path like `source.foo` or a quoted literal. Paths are resolved against `data`.
func Placeholder ¶
Placeholder returns a bound-parameter placeholder suitable for the driver and 1-based index. The driver argument may be either a user-facing type label (e.g. "mssql") or an actual driver name (e.g. "sqlserver"); both are normalized through CanonicalDriver so the placeholder style always matches the driver that will ultimately execute the query.
func QuoteIdent ¶
QuoteIdent validates and quotes an SQL identifier (optionally schema-qualified) according to the target driver. It supports dot-separated identifiers like schema.table. Drivers: pgx/postgres -> "name", mysql/mariadb/sqlite -> `name`, mssql -> [name].
It returns an error for any malformed or unsafe identifier. Callers MUST check the error and never splice the result on failure, otherwise an empty identifier would be interpolated straight into the SQL (e.g. "WHERE = $1").
NOTE: quoting makes identifiers case-sensitive. On engines that fold unquoted identifiers (e.g. PostgreSQL lower-cases them) a column physically named "id" must be referenced as "id", not "ID".
func RecordFromValues ¶ added in v1.8.1
RecordFromValues builds one row map from a scanned row.
Driver []byte becomes a string, as it always has, except where typeNames says the column is JSON or a PostgreSQL array -- there it becomes the value the literal describes, matching what pgx already produces for those columns on its own paths.
typeNames may be nil or shorter than names.
func ScanRows ¶
ScanRows scans sql.Rows into a slice of maps. It is hard-limited to DefaultMaxRows.
Values go through RecordFromValues, so a column the driver names JSON or JSONB arrives as the value its document describes and every other column keeps the shape it has always had -- driver []byte becomes a string.
This is the generic database/sql read path: db_lookup uses it in both key-column and query-template mode, and the editor's SQL builder falls back to it whenever a query binds arguments. Without the JSON half, a `jsonb` column reached transformations, sinks and the message trace as a string holding JSON, while the very same column fetched through pgx natively (PostgresSource.ExecuteSQL, and the source's snapshot and polling paths) arrived as a map. One document, two shapes, decided by which path fetched it.
func TemplateArgs ¶ added in v1.8.1
TemplateArgs returns the values a template's {{ ... }} tokens bind to, in placeholder order, resolved against data.
It is the same walk ParameterizeTemplateEx performs, minus the statement text, so a caller that needs to know "what does this template depend on in this message" cannot drift from what the executed query actually binds. The driver only decides placeholder syntax, which no such caller cares about, so there is no driver argument.
The use that motivated it is cache identity: a templated lookup's per-message input lives entirely in these values, and a cache keyed on the template text alone is the same key for every message.
func TemplateArgsWith ¶ added in v1.12.0
TemplateArgsWith is TemplateArgs with the path resolution rule supplied by the caller.
Anything deriving a cache key from a template has to use the same rule the statement is built with, or the key can describe a different query from the one that runs.
func ValidateIdent ¶
ValidateIdent verifies that an identifier (optionally schema/keyspace-qualified) contains only safe characters, has no empty segments, and respects the common length limit, without altering its quoting. Use it for engines (e.g. CQL) where identifiers are interpolated as-is, to prevent SQL/CQL injection.
Types ¶
type ColumnMapping ¶
type ColumnMapping struct {
SourceField string `json:"source_field"`
TargetColumn string `json:"target_column"`
DataType string `json:"data_type"` // Optional, used for auto-creation
IsPrimaryKey bool `json:"is_primary_key"` // Optional
IsNullable bool `json:"is_nullable"` // Optional
IsIdentity bool `json:"is_identity"` // Optional, auto-increment/sequence
}
ColumnMapping defines how a source field maps to a sink column.
func ParseColumnMappings ¶
func ParseColumnMappings(s string) ([]ColumnMapping, error)
ParseColumnMappings parses a JSON string into a slice of ColumnMapping.
A blank source_field is treated as "use the column's own name": it falls back to target_column. Without this normalization an empty source_field resolves to a nil value at write time (evaluator.GetMsgValByPath returns nil for an empty path), so every such column is bound as NULL and rows fail on NOT NULL / PRIMARY KEY constraints. UIs commonly leave source_field blank to mean "same name as the target column", so this default makes that intent work safely.
type Resolver ¶ added in v1.12.0
Resolver answers what one token's path resolves to.
It exists because a SQL template used to be strictly weaker than every other template in Hermod. Tokens here resolved through GetFromMapPath, a literal walk of the data map, while a condition, a sink mapping or a db_lookup keyField resolved through evaluator.GetMsgValByPath, which also answers the CDC envelope (after., before.), the virtual fields (operation, table, schema) and meta.. The data map of a CDC message *is* its after-image, so `after.x` had nothing to walk and bound NULL -- while the editor's SQL builder, whose sample still carries the envelope, ran the same text and returned rows.
The rule is supplied by the caller rather than imported, because sqlutil sits deliberately below the transformer packages so that sources and sinks can use it; importing the evaluator here would invert that.
func MapResolver ¶ added in v1.12.0
MapResolver resolves paths as a dotted walk of a plain map -- the behaviour every map-taking entry point below still has. It is the right rule for a caller that genuinely has no message: batch_sql binds from a `parameters` object on the source config, where an envelope path would mean nothing.
type TemplateBinding ¶ added in v1.8.1
type TemplateBinding struct {
SQL string
Args []any
// Unresolved holds the paths of tokens that resolved to nothing. Those
// tokens are still bound -- as NULL -- because an optional message field is
// a legitimate reason for a path to be empty. Callers whose variables come
// from configuration rather than from a message should treat a non-empty
// Unresolved as an error instead: binding NULL there turns a typo into a
// query that silently matches no rows.
Unresolved []string
// Err is set when the template could not be turned into a statement at all
// -- currently only when a list is too long to expand. The SQL and Args are
// not usable when it is set.
Err error
}
TemplateBinding is the result of turning a SQL template into a parameterized statement: the rewritten SQL, the bound arguments in placeholder order, and the token paths that resolved to nothing.
func ParameterizeTemplateEx ¶ added in v1.8.1
func ParameterizeTemplateEx(driver, tpl string, data map[string]any) TemplateBinding
ParameterizeTemplateEx is ParameterizeTemplate with the full binding result, resolving paths as a walk of data. See ParameterizeTemplateWith for how a list token is expanded.
func ParameterizeTemplateWith ¶ added in v1.12.0
func ParameterizeTemplateWith(driver, tpl string, resolve Resolver) TemplateBinding
ParameterizeTemplateWith is ParameterizeTemplateEx with the path resolution rule supplied by the caller. Every entry point above reaches the statement through here, so this is where the list rule lives.
A token whose value is a slice and which sits directly in an `IN (...)` list expands to one placeholder per element. Binding a slice to a single placeholder is not a list -- it is an encoding error on every driver ("unsupported type []interface {}, a slice of interface" through database/sql, "cannot find encode plan" through pgx) -- so `id IN ({{.ids}})` had no working form before this.
The expansion is deliberately confined to IN lists. `= ANY({{.ids}})` is the native Postgres array form and already works by binding the slice whole; expanding it would produce `= ANY($1, $2)`, a syntax error.
Source Files