feat: add postgres toast column resolution when unavailable during cdc

[vaibhav-datazip]

Description

Problem

With Postgres CDC (pgoutput), an UPDATE does not send large column values that did not change (unchanged TOAST values), unless the table uses REPLICA IDENTITY FULL. OLake currently writes __olake_unavailable_value__ for those columns, which overwrites the real value in the destination.

Solution

A new Postgres CDC option, preserve_toast_values. When it is on:

1. The Postgres reader marks which columns of an update were left out.
2. The Iceberg destination fills those columns with the row's previous value:
   • from memory, if this sync already wrote the row;
   • otherwise from the destination table: the table index gives the row's file and position, and the Java writer reads just that row.

If a value cannot be recovered, it is written as __olake_unavailable_value__, as before.

Configuration

"update_method": {
  "replication_slot": "olake_slot",
  "publication": "olake_pub",
  "preserve_toast_values": true
}

• Default: false. Off means no behaviour change.
• Choose it when you create the source. With it on, discover only offers pos / dv update types.

Supported

Setup	Recovered?
Iceberg, update type pos or dv, CDC	✅
Legacy (Java) writer and arrow writer	✅
Normalization on or off, partitioned or not, with filters	✅
Update type eq	❌ placeholder kept (no table index)
Append mode	❌ placeholder kept (no table index)
Parquet / S3 destination	❌ placeholder kept
Backfill / full refresh / incremental	not required, these read whole rows
REPLICA IDENTITY FULL tables	not required, Postgres already sends the value
wal2json	❌ not supported (deprecated)

Fixes #1176

Type of change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• This change requires a documentation update

How has this been tested ?

• CDC sync with preserve_toast_values: pos and dv, legacy and arrow writers
• CDC sync without preserve_toast_values: pos and dv, legacy and arrow writers (output identical to staging)
• Non-TOAST column change: unchanged TOAST values recovered
• TOAST value change, then a non-TOAST change in the next sync: new value kept (pos and dv)
• Normalization off: unchanged TOAST values recovered]
• Partition on a non-TOAST column
• Filter on a TOAST column
• Filter on a non-TOAST column
• Selected columns: a TOAST column left unselected is never read; the other TOAST columns are still recovered
• Small batch size (10): updates spread across many batches in one sync
• Multiple syncs: recovered value read back again in the next sync

Screenshots or Recordings

Documentation

• Documentation Link: [link to README, olake.io/docs, or olake-docs]
• N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

[github-actions]

Setting this global flag constrains discover's available_update_types to index-based modes (pos/dv) for every stream. When --target-query-engines is not passed (the default), mergeUpdateType returns early and an existing update_type=eq (or a blank one, which it records as eq) survives re-discovery without any warning; at sync, ValidateUpdateType then rejects those streams against the narrowed list and they are skipped with only a warn log (or the sync fails with 'no valid streams found'). Users enabling post_read get no discover-time signal that their CDC streams must switch to pos/dv. Consider logging an explicit warning here when PostRead is enabled, or ensuring the discover merge surfaces the invalidated choice regardless of whether engines were specified.

Suggestion:

Suggested change

	// post_read reads rows back through the destination's table index
	viper.Set(constants.TableIndexRequired, cdc.PostRead)
	// post_read reads rows back through the destination's table index
	viper.Set(constants.TableIndexRequired, cdc.PostRead)
	if cdc.PostRead {
	logger.Warn("post_read enabled: CDC streams must use positional (pos) or deletion-vector (dv) delete mode; re-run discover and update update_type, otherwise existing equality-delete streams are skipped at sync")
	}

[github-actions]

This makes AvailableUpdateTypes depend on process-global viper state that is set late: viper.Set(constants.TableIndexRequired, ...) only happens in Postgres.Setup (drivers/postgres/internal/postgres.go:169), which discover invokes in RunE (protocol/discover.go:56) — while resolveTargetQueryEngines() already calls this function from the same command's PreRunE (protocol/discover.go:42). On that pre-Setup call the flag always reads false, so post_read is silently ignored there. Today the only pre-Setup consumer is the 'no readable delete format' precondition, whose outcome happens to be flag-independent (every engine that supports eq also supports pos, and every engine supports pos or dv, so dropping eq can never empty the set) — but any future caller or engine-matrix change inherits this hidden ordering constraint. Prefer passing the flag explicitly from the caller (the source driver knows it) instead of a global side channel, or at least document that the value is only valid after connector.Setup.

Suggestion:

Suggested change

	indexRequired := viper.GetBool(constants.TableIndexRequired)
	func AvailableUpdateTypes(engines []QueryEngine, indexRequired bool) []UpdateType {
	available := make([]UpdateType, 0, len(writableUpdateTypes))
	for _, updateType := range writableUpdateTypes {
	if indexRequired && !updateType.NeedsTableIndex(Iceberg) {
	continue
	}
	if readableByAll(engines, updateType) {
	available = append(available, updateType)
	}
	}
	return available
	}

[github-actions]

UnavailableValue is a plain, user-representable string used as an in-band marker: the toast resolver distinguishes markers from real data purely by string equality (isUnavailable in destination/iceberg/toast.go). With post_read enabled, a genuine TEXT/JSON source value exactly equal to this constant is misclassified, and I traced a concrete silent-corruption path: (1) a row is written with a normal value, which is remembered into carry; (2) the row is updated so this column's real new value equals the constant — the resolver's remember path skips it because isUnavailable(value) matches, so carry keeps the stale value; (3) a later UPDATE leaves the column as unchanged TOAST, and the resolver recovers the stale carry value, permanently replacing the true data in the destination. Prefer discriminating via the record's UnavailableColumns metadata (slices.Contains) wherever record context exists (Resolve/remember), and/or make the marker collision-proof (e.g., embed a UUID component). If the constant is changed, note destinations already written by earlier versions contain the old literal, which then stops being recognized as a placeholder.

[github-actions]

isUnavailable only matches a plain Go string, but the non-normalized path of rowValues returns every stored value as json.RawMessage, so a stored placeholder (written by an earlier sync before recovery existed) is never detected here: it is remembered into carry as if it were real data and counted in recovered/fromTable, contradicting the documented intent ("the stored version has the placeholder too") and the normalized-mode-only test "a stored placeholder is not counted as recovered". Output bytes happen to stay identical (the RawMessage embeds as the same JSON string), but the resolver reports unrecovered values as recovered and caches the placeholder in carry, suppressing later read retries for that column. Making isUnavailable RawMessage-aware — or decoding JSON string tokens to plain strings in rowValues, which also fixes the partition-value issue — resolves this.

Suggestion:

Suggested change

	for column, value := range values {
	if isUnavailable(value) {
	continue // the stored version has the placeholder too
	}
	r.remember(pending.olakeID, column, value)
	r.fromTable++
	}
	for column, value := range values {
	if text, ok := value.(string); ok && text == constants.UnavailableValue {
	continue // the stored version has the placeholder too
	}
	if raw, ok := value.(json.RawMessage); ok && string(raw) == strconv.Quote(constants.UnavailableValue) {
	continue // non-normalized: the stored version has the placeholder too
	}
	r.remember(pending.olakeID, column, value)
	r.fromTable++
	}

[github-actions]

In non-normalized mode the recovered value is a json.RawMessage (the raw JSON token, quotes included for text columns). FlattenAndCleanData copies partition values straight out of record.Data into Data[pInfo.SchemaField] (iceberg.go:397-398), so a recovered TOASTed partition column reaches the writers as json.RawMessage instead of the plain string the source produced: the arrow writer's identityTransform fails its string assertion ("expected string for colType "string", got []uint8") and fails the whole batch, while the legacy writer's toProtoFieldValue default branch stringifies the []byte as a numeric list (e.g. "[34 118 97 108 ...]"), silently landing the row in a wrong partition. This defeats the stated intent that "partition values ... see the recovered values". Decoding JSON string tokens into plain Go strings here (keeping compound tokens as RawMessage so jsonb values round-trip verbatim) makes recovered values byte-identical to what the source would have produced, both inside the StringifiedData document and for partition extraction.

Suggestion:

Suggested change

	for _, column := range columns {
	// a key missing from the stored row keeps the placeholder rather than become NULL
	if raw, present := row[column]; present {
	values[column] = raw
	}
	}
	for _, column := range columns {
	// a key missing from the stored row keeps the placeholder rather than become NULL
	if raw, present := row[column]; present {
	// decode JSON string tokens to plain strings so recovered values match what the
	// source produced (placeholder detection, partition values); compound JSON stays raw
	var text string
	if err := json.Unmarshal(raw, &text); err == nil {
	values[column] = text
	} else {
	values[column] = raw
	}
	}
	}

[github-actions]

This makes AvailableUpdateTypes depend on process-global viper state that only the postgres driver sets during connector.Setup(). Within one discover run the function is now evaluated with different effective inputs: resolveTargetQueryEngines() calls it in PreRunE (protocol/discover.go:42/101) BEFORE connector.Setup() (RunE, discover.go:56) sets TABLE_INDEX_REQUIRED, while LogCatalog -> GetWrappedCatalog/mergeUpdateType evaluates it after. Today this is masked because every engine in the matrix supports positional deletes, so the PreRunE "no delete format is readable by all of the selected query engines" fail-fast can never diverge; but if an engine lacking positional support is ever added, discover would pass the precondition (unfiltered) and then emit a catalog whose index-filtered available set is empty, silently bypassing the check. Prefer threading the flag as an explicit parameter (computed after Setup, or derived from the already-unmarshalled source config in PreRunE) so both call sites observe the same input.

Suggestion:

Suggested change

	indexRequired := viper.GetBool(constants.TableIndexRequired)
	available := make([]UpdateType, 0, len(writableUpdateTypes))
	for _, updateType := range writableUpdateTypes {
	if indexRequired && !updateType.NeedsTableIndex(Iceberg) {
	continue
	}
	// AvailableUpdateTypes returns the writable delete formats every engine can read, cheapest
	// first. indexRequired drops formats that keep no destination table index. No engines means
	// the choice is unconstrained.
	func AvailableUpdateTypes(engines []QueryEngine, indexRequired bool) []UpdateType {
	available := make([]UpdateType, 0, len(writableUpdateTypes))
	for _, updateType := range writableUpdateTypes {
	if indexRequired && !updateType.NeedsTableIndex(Iceberg) {
	continue
	}
	if readableByAll(engines, updateType) {
	available = append(available, updateType)
	}
	}
	return available
	}

[github-actions]

PreserveToastValues is honored only by pgoutputReplicator, but Postgres.Setup publishes viper.Set(constants.TableIndexRequired, cdc.PreserveToastValues) unconditionally for any CDC config. For a slot using the wal2json plugin (this fallback path), the source never marks UnavailableColumns (FilterWalJsChange builds CDCChange without it, and wal2json omits unchanged-TOAST columns from its output), so enabling preserve_toast_values silently steers discover-time update types toward index-keeping formats and costs destination row-index maintenance while no TOAST value is ever recovered. Consider warning (or failing) when the slot's plugin is not pgoutput — the plugin is known after GetSlotPosition in StreamCDC — or noting the pgoutput-only requirement in the spec description.

Suggestion:

Suggested change

	return &pgoutputReplicator{socket: socket, publication: config.Publication, preserveToastValues: config.PreserveToastValues, relationIDToMsgMap: make(map[uint32]*pglogrepl.RelationMessage)}, nil
	default:
	if config.PreserveToastValues {
	logger.Warnf("preserve_toast_values has no effect with the %q plugin (pgoutput only); unchanged-TOAST column values will not be recovered", plugin)
	}
	return &wal2jsonReplicator{socket: socket}, nil

[vaibhav-datazip]

wal2json has been deprecated, we are explicitly not handling in that case.

[mihir-datazip]

Small change: can we move this waljs package inside the postgres driver code? Currently changes in this package detects change in all the drivers. As we are touching this code, better to move it where it should be

[mihir-datazip]

Can we run memory benchmark to have a metric on how much RAM olake will need with given average toast value size?

Also I think we should add a new test case in our Performance Tests where we have toast values as well. That way in future we can catch change that can cause OOM to customers before release.

[mihir-datazip]

Can you confirm if we can show any info tooltip for this in UI?

Also make sure the documentation clearly shows that without this toast will land as unavailable value in destination with a clear screenshot

[vaibhav-datazip]

had added about correcting the documentation , will also ask the team to add screenshot.

[mihir-datazip]

Logic of setting record.UnavailableColumns should be kept in types.CreateRawRecord function only

[vaibhav-datazip]

NewCDCChange function for postgres will get the unavailable columns in a cleaner way , but this will create an unnecessary argument passing for other drivers which have cdc.

[mihir-datazip]

If the updateType doesn't need index doesn't meant it cannot have one. Why are we adding this restriction?

[mihir-datazip]

We should not use viper here, just update the function params and pass it from the driver side

[mihir-datazip]

TableIndexRequired flag should be set outside the cdc specific codeblock as it might not always just depend on cdc specific settings.

Keep it at an appropriate common place

[mihir-datazip]

How was the code working before this? It would panic

[mihir-datazip]

Same here, unchanged column setting should be the part of abstract.NewCDCChange itself. That way we will always be aware in the caller sites and explicitly pass nil if not applicable

[mihir-datazip]

We should not name things in destination based on postgres only terminology.
From destination code, this is basically resolver for values that come as unavailable so name it accordingly

[mihir-datazip]

we will need to rename this file as well

[mihir-datazip]

This doesn't really set with Writer definition
We are doing it because the writer we use in toastResolver function needs it. See if that can be handled without change in interface, if this is not possible then just make a new interface with Writer embedded and we can use that in the toast Resolver.
This will also automatically handle the case of legacy writer without adding new nil function

[mihir-datazip]

No test reaches this page-skipping path or an actual closeOpen close: IT files have one page per column and are already committed.
Add Java unit tests for multi-page reads across page/row-group boundaries and for write → closeOpen → write.

[mihir-datazip]

Every non-double column goes through getString here (binary garbles, int/long throw), and toFieldValue sends any other type as NULL.
Decode by column type straight into FieldValue (no Row/toFieldValue hop) and fail on unsupported types; only string/double reach it today.

[mihir-datazip]

Offset indexes are read here and again inside readFilteredRowGroup, so every row group fetches them twice.
Take them from reader.getColumnIndexStore(blockIndex), which is cached and reused by readFilteredRowGroup.

[mihir-datazip]

Files are read one at a time, each costing several storage round trips, which stalls the flush when rows span many files.
Read files concurrently on a small bounded pool; Go matches rows by file and position, so order doesn't matter.

[mihir-datazip]

lastRouted/lastDataWriter can still point at a writer closed here, whose currentPath() throws.
Reset both, as close() does.

[mihir-datazip]

This hand pin overrides Iceberg's Parquet on a future bump, silently mixing versions.
Add a maven-enforcer requireUpperBoundDeps rule for org.apache.parquet so the drift fails the build.

[mihir-datazip]

Added some review comments after some local testing

[mihir-datazip]

Just to confirm if preserve_toast_values is true and someone sets one stream to update_type: eq and syncs.
Currently we silently skip the stream with warning. Is this behaviour discussed with product?

[mihir-datazip]

Verify the following test case once:

1. Create a table with two large text columns named BigDoc and big doc: one has capital letters, the other has a space.
2. turn on use_source_column_names, so the Iceberg columns keep exactly those names. By default OLake would rename them bigdoc and big_doc.
3. It copies the table, then updates only a small number column. Postgres says "unchanged" for both large columns, so OLake has to fetch both from Iceberg.
4. It reads the table with Spark and checks that both columns still hold the original text.

Why it fails

Iceberg keeps its data in Parquet files, and a column's name in the table isn't always its name inside those files. The legacy (Java) writer is built on Iceberg's Java library, which renames columns containing spaces or symbols when it writes Parquet. So big doc is stored in the files as big_x20doc (20 is the hex code for a space). Iceberg links the two names through a column ID, which is why Spark still shows the column as big doc.

When OLake looks up the old value, it searches the Parquet file for a column named exactly big doc, not for the ID. That name doesn't exist in the file, so the lookup finds nothing and the placeholder stays.

[mihir-datazip]

If a toast column is deselected, then reselected then that column is stored as NULL instead of placeholder I think. Ensure we handle this case by using unavailable placeholder instead of NULL