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 DecodeJSONColumn(raw []byte) any
- func DecodeValue(v any, isJSON bool) any
- func GetFromMapPath(m map[string]any, path string) any
- func IsJSONColumnType(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 ValidateIdent(name string) error
- type ColumnMapping
- 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 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 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 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 -- there it becomes the value the JSON describes, matching what pgx already produces for PostgreSQL jsonb on its own paths.
typeNames may be nil or shorter than names.
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 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 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.
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