feat(structured-ingestion) DBI-1096: Generic flattening schema projection for structured ingestion

[pfcoperez]

This PR adds the generic tools to implement structured ingestion (using QValue system) for any unstructured (dynamic schema) data source.

It provided two abstractions for this purpose:

• SchemaProjector (db9ce41): Upon initialization, it receives the target schema ([]*protos.ColumnSetting) and a function to interpret these raw mappings column settings as an ordered array of QValue. Through ProjectRecord method it transforms generic dynamic schema records (abstracted behind a walk iterator so it could be JSON, BSON, or anything that a lazy walker function can take) into an ordered array of QValue instances matching the schema plus an extra QValueJSON column (malformed_data) reporting records not matching the schema (see next point). In the ouput QValues record all fields are nullable as missing fields are considered NULL for structured logging.

One example of lazy iterator applied for MongDB document flattening using SchemaProjector is:

peerdb/flow/connectors/mongo/qvalue_convert.go

Lines 246 to 273 in 859e699

	func DocumentQValueIterator(raw bson.Raw, converter BsonToQValueConverter) (iter.Seq2[string, types.QValue], func() error) {
	var walkErr error
	return func(yield func(string, types.QValue) bool) {
	elements, err := raw.Elements()
	if err != nil {
	walkErr = fmt.Errorf("failed to read document fields: %w", err)
	return
	}
	for _, element := range elements {
	field, err := element.KeyErr()
	if err != nil {
	walkErr = fmt.Errorf("failed to read document field name: %w", err)
	return
	}
	if field == DefaultDocumentKeyColumnName {
	continue
	}
	value, err := converter.QValueFromBsonValue(element.Value(), types.QValueKindInvalid)
	if err != nil {
	walkErr = fmt.Errorf("failed to convert document field to QValue %s: %w", field, err)
	return
	}
	if !yield(field, value) {
	return
	}
	}
	}, func() error { return walkErr }
	}

• MalformedData (898e5d0): While processing each individual unstructured document, it is the tracker of schema violations. Its JSON marshalling method implementation generates a JSON structure that makes it possible and easy to query data that failed to fit into the schema at destination CH table:

{
 "<offending_field_i>": {
    "unexpected": true|false,        // True iff this field was not in the projected schema
    "type_mismatch": true|false, // True iff this field was in the schema but with an incompatible type
 }
}

e.g:

{
    "year": {
        "unexpected": true,
        "value": 2019
      }
}

{
    "year": {
        "type_mismatch": true,
        "value": "two thousand nineteen" 
      }
}

Query example at destination CH table:

SELECT name, coalesce(age, malformed_data.age.value) FROM test_people;

Carmen	seven
Sofía	4

Given that it provides a default CH types to QKind schema interpreter (Used for MongDB structured ingestion) and that it matches the equivalence function already present in the ClickHouse target connector implementation, this PR also factors out the conversion table so it's shared: 60a80b9

Part of: https://linear.app/clickhouse/issue/DBI-1096

[claude]

Code review

No issues found. Checked for bugs and CLAUDE.md compliance.

[pfcoperez]

@fable do a deep review with explanations of the changes.

[claude]

Claude finished @pfcoperez's task in 4m 6s —— View job

---

Deep review: generic flattening schema projection for structured ingestion

• Gather context (diff, changed files, related connector code)
• Review flow/pkg/clickhouse/qvalue_kind.go (factored-out CH type → QKind conversion)
• Review flow/connectors/clickhouse/clickhouse.go changes
• Review flow/connectors/utils/structured/malformed_data.go + tests
• Review flow/connectors/utils/structured/schema.go + tests
• Post detailed review with explanations

---

What the PR does

The PR introduces a source-agnostic pipeline for ingesting dynamic-schema documents into a fixed columnar schema, in three layers:

1. flow/pkg/clickhouse/qvalue_kind.go — extracts the ClickHouse-type → QValueKind switch that previously lived inline in GetTableSchemaForTable (flow/connectors/clickhouse/clickhouse.go) into a shared QValueKindForType(string) (string, error). It returns a string rather than types.QValueKind, consistent with flow/pkg/clickhouse deliberately staying free of flow-internal imports (it only depends on flow/pkg/common). I diffed the new switch against the one on main case-by-case and cross-checked every returned literal against the constants in flow/shared/types/kind.go:11-67 — the mapping is behavior-identical, with one intentional-looking addition, see note 1 below.

2. MalformedData (flow/connectors/utils/structured/malformed_data.go) — a per-record tracker of schema violations. AddField records a field name, a reason (unexpected / type_mismatch / not_a_number), and optionally the offending value. MarshalJSON produces the query-friendly {"<field>": {"<reason>": true, "value": ...}} shape, and AsQValue wraps it as a QValueJSON destined for the reserved malformed_data column. Non-finite floats (which encoding/json rejects) are caught by isJSONRepresentable and downgraded to a value-less not_a_number entry. Output is deterministic because encoding/json sorts map keys — and the tests assert that explicitly.

3. SchemaProjector (flow/connectors/utils/structured/schema.go) — resolves the target schema once at construction ([]*protos.ColumnSetting + a pluggable schemaToQKind resolver; NewSchemaProjectorFromCHtoQValue plugs in the shared CH resolver from layer 1). Construction rejects unresolvable types, duplicate columns, and a column named malformed_data. ProjectRecord consumes a record as a lazy iter.Seq2[string, QValue] walk — so JSON, BSON, or anything else can feed it — and emits an ordered []QValue: schema values in declared order (all nullable; missing fields become typed nulls, and a null value of any kind fits any column), plus the malformed_data JSON column. Unknown fields and kind-mismatched values are diverted into MalformedData instead of failing the record. ApplyRecordSchema adapts the same output into model.RecordItems for CDC consumers.

Nothing outside the tests calls SchemaProjector yet — the MongoDB walker shown in the PR description lives in a follow-up branch, so this PR is pure infrastructure plus the ClickHouse refactor.

Findings

1. Behavior change hidden in the "pure refactor": Nullable(JSON) is now accepted — flow/pkg/clickhouse/qvalue_kind.go:66 maps "JSON", "Nullable(JSON)", while the old switch on main only had case "JSON". This means GetTableSchemaForTable (and therefore GetTableSchema for ClickHouse source peers) no longer errors on Nullable(JSON) columns. Almost certainly an improvement, but it's an untracked behavior change inside a commit presented as factoring — worth a mention in the commit/PR text so it doesn't surprise anyone bisecting later.

2. Non-finite floats nested inside compound values fail the whole record (medium) — isJSONRepresentable (flow/connectors/utils/structured/malformed_data.go:117) only inspects top-level float32/float64. A malformed field whose value is, say, a QValueArrayFloat64 containing a NaN (perfectly reachable from a BSON array) passes that check, then json.Marshal fails, AsQValue errors, and ProjectRecord (schema.go:149-155) returns an error for the entire record. That inverts the feature's purpose: the machinery built to make ingestion resilient to bad data becomes the thing that rejects the record. Suggest a fallback in MarshalJSON: if marshalling a value fails (or per-element checking detects non-finite entries), degrade that field to a value-less entry (as already done for scalar NaN) instead of propagating the error. Fix this →

3. shouldRecordValues=false still persists values of unexpected fields (medium, PII question) — schema.go:126 records the value for ReasonUnexpected unconditionally; only the ReasonTypeMismatch path (schema.go:139-141) honors the flag. The test documents this asymmetry with a NOTE, so it may be intentional — but if the flag exists so operators can avoid persisting raw source values (compliance/PII), unexpected fields are precisely the ones most likely to contain data nobody vetted, since they're by definition outside the declared schema. If intentional, a doc comment on shouldRecordValues stating the scope ("gates mismatched values only") would prevent misuse; if not, gate line 126 too.

4. Duplicate record fields produce contradictory output (low) — BSON legally allows repeated keys and JSON tolerates them, and ProjectRecord consumes whatever the walker yields. If a field appears twice — first with a valid value, then with a mismatched one — the column keeps the first value and malformed_data reports a type_mismatch for the same field; two valid occurrences silently last-write-win into the slot. Related: AddField (malformed_data.go:55-60) updates the reason but keeps a stale value when the new value is nil, so repeated adds can pair a value with a reason it didn't belong to. Cheapest fix is documenting the iterator contract (field names must be unique, e.g. dedup in the walker); alternatively treat a repeated field as malformed itself.

5. Drift risk in the string mapping (nit) — qvalue_kind.go returns raw literals ("array_bool", "timestamp", …) that must stay in sync with flow/shared/types/kind.go. All match today (verified each one), but nothing enforces it. A tiny test in the flow module asserting each reachable return value is a known types.QValueKind would lock this in without giving flow/pkg/clickhouse a new dependency.

6. Pre-existing quirks now shared (nits, no action needed) — the strings.Contains(columnType, "Decimal") fallback (qvalue_kind.go:69) would classify any type merely containing "Decimal" (e.g. Map(String, Decimal(10,2))) as numeric; that's carried over verbatim from the old switch. Also Date, DateTime (non-64) and DateTime64(3) remain unmapped, as before. And a typo: schema.go:136 "canbe" → "can be".

7. NaN downgrade discards the original reason (design nit) — malformed_data.go:92 replaces e.g. {"type_mismatch": true} with {"not_a_number": true} rather than combining them. The tests pin this behavior, so it's deliberate; just noting that {"type_mismatch": true, "not_a_number": true} would preserve more diagnostic signal at no cost.

What's good

• Both test files are thorough and table-driven, cover the error invariants (unknown reason, value-without-reason), and pin JSON determinism — the require.Equal alongside require.JSONEq in malformed_data_test.go:94 is a nice touch.
• Resolving column kinds once at construction and pre-filling typed nulls keeps ProjectRecord a single pass over the walk with O(1) per field — the right shape for a per-record hot path.
• The iter.Seq2 walker abstraction cleanly decouples flattening (source-specific) from projection (generic), and the reserved-column collision check at construction prevents a silent footgun.
• Per .claude/REVIEW.md: no new stdout/stderr/log output at all in this PR, so no secret/PII logging risk; not a dependency bump.

None of the findings are blockers for merging infrastructure-only code; 2 and 3 are the ones I'd resolve (or explicitly rule intentional) before the MongoDB walker starts feeding real data through this path.

[dtunikov]

hm
why can't we return QValueKind type from here?
because you want it to stay in pkg and pkg can't refer to flow/shared/types package?

[pfcoperez]

Because flow/pgk doesn't have that type defined. I think that's OK since it's just a rename of string (from now) but it is less than ideal.

Once in the past I attempted bringing Q abstractions to flow/pkg but that's a rabbit hole for a PR like this one.

[dtunikov]

yeah, but I'm wondering why do you want to put it in pkg at all?
afaiu we put things in pkg when we want to re-use them from the other repos.
this function refers to internal peerdb qvalue/qkind concept which is unknown to the outside world

i guess you didn't want to put it in flow/connectors/clickhouse/clickhouse.go because you need to reference it from utils/structured?

i'd rather keep it closer to QValueKind itself so that it would be hard to forget to update it when a new kind is added

[pfcoperez]

I am preparing to potentially use them in Discovery validations.

[dtunikov]

nit: can wrap float validation into helper:

func isJSONRepresentable(v any) bool {
	validateFloat := func(f float64) bool {
		return !math.IsNaN(f) && !math.IsInf(f, 0)
	}
	switch f := v.(type) {
	case float64:
		return validateFloat(f)
	case float32:
		return validateFloat(float64(f))
	default:
		return true
	}
}

[dtunikov]

i wouldn't add 3 NewProjectorFromX implementations prematurely
imo just one NewSchemaProjector is fine until we see the need for other ways to create this object

[dtunikov]

i'm not sure i understand why we'd need these two methods - Columns and QRecordSchema.
can we maybe add them once we see the need for that in the code that is going to use this SchemaProjector?
kinda the same concern as 3 constructors.

[pfcoperez]

This was used in the context of #4774 at a certain point but it's no longer. We can remove it indeed.