feat: add postgres toast column resolution when unavailable during cdc

constants/constants.go

@@ -42,6 +42,9 @@ const (
 	UpdateMethodStandalone = "Standalone"
 	// DestinationDatabasePrefix is used as prefix for destination database name
 	DestinationDatabasePrefix = "DESTINATION_DATABASE_PREFIX"
+	// TableIndexRequired is set by a source that reads rows back through the destination's
+	// table index (Postgres preserve_toast_values), so only update types that keep the index are offered.
+	TableIndexRequired = "TABLE_INDEX_REQUIRED"
 	// EffectiveParquetSize is the effective size in bytes considering 256mb targeted parquet size, compression ratio as 8
 	EffectiveParquetSize        = int64(256) * 1024 * 1024 * int64(8)
 	DB2StateTimestampFormat     = "2006-01-02 15:04:05.000000"
@@ -79,6 +82,10 @@ const (
 	DefaultCacheSize          = 128 * 1024 * 1024 // 128 MB default block cache
 	DefaultMemTableSize       = 64 * 1024 * 1024  // 64 MB default memtable
 	DefaultMaxOpenFiles       = 1000
+
+	// UnavailableValue is written for a column the source could not send
+	// (Postgres unchanged TOAST when the replica identity is not FULL).
+	UnavailableValue = "__olake_unavailable_value__"
 )
 
 // DriverType identifies a source/destination driver.

destination/iceberg/arrow-writer/utils.go

@@ -62,6 +62,10 @@ func getDefaultWriterProps() []parquet.WriterProperty {
 
 		// iceberg writes root name as "table" in parquet's meta
 		parquet.WithRootName("table"),
+
+		// Page index: lets a reader fetch the page holding one row instead of the whole
+		// column. parquet-java writes it by default; query engines use it to skip pages.
+		parquet.WithPageIndexEnabled(true),
 	}
 }
 

destination/iceberg/arrow-writer/writer.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"strings"
 	"time"
 
@@ -78,7 +79,7 @@ type PositionalDelete struct {
 	Position int64
 }
 
-func New(ctx context.Context, options *destination.Options, partitionInfo []internal.PartitionInfo, schema map[string]string, stream types.StreamInterface, server internal.ServerClient, upsertMode bool) (*ArrowWriter, error) {
+func New(ctx context.Context, options *destination.Options, partitionInfo []internal.PartitionInfo, schema map[string]string, stream types.StreamInterface, server internal.ServerClient, upsertMode bool, indexThread *types.StreamIndexThread) (*ArrowWriter, error) {
 	writer := &ArrowWriter{
 		options:       options,
 		partitionInfo: partitionInfo,
@@ -89,7 +90,7 @@ func New(ctx context.Context, options *destination.Options, partitionInfo []inte
 		writers:       make(map[string]*Writer),
 		createdFiles:  make(map[string]*PartitionFiles),
 		upsertMode:    upsertMode,
-		indexThread:   types.NewStreamIndexThread(options.TableIndex),
+		indexThread:   indexThread,
 		deleteMode:    stream.GetUpdateType(),
 
 		pendingVectors: make(map[string]*pendingVector),
@@ -341,6 +342,12 @@ func (w *ArrowWriter) checkAndFlush(ctx context.Context, rw *RollingWriter, part
 		return rw, nil
 	}
 
+	return w.roll(ctx, rw, partitionKey)
+}
+
+// roll uploads rw's file and returns a writer on a new file. Positions already handed
+// out stay valid: the uploaded file keeps its path.
+func (w *ArrowWriter) roll(ctx context.Context, rw *RollingWriter, partitionKey string) (*RollingWriter, error) {
 	if err := w.flush(ctx, rw, partitionKey); err != nil {
 		return nil, err
 	}
@@ -359,6 +366,26 @@ func (w *ArrowWriter) checkAndFlush(ctx context.Context, rw *RollingWriter, part
 	return newWriter, nil
 }
 
+// EnsureReadable uploads any file in paths that is still being written, so its rows
+// can be read before the commit.
+func (w *ArrowWriter) EnsureReadable(ctx context.Context, paths []string) error {
+	for partitionKey, writer := range w.writers {
+		// completeWriters (schema evolution) leaves a partition without a data writer until
+		// its next record; its files are already uploaded.
+		if writer.dataWriter == nil || !slices.Contains(paths, writer.dataWriter.filePath) {
+			continue
+		}
+
+		rolled, err := w.roll(ctx, writer.dataWriter, partitionKey)
+		if err != nil {
+			return fmt.Errorf("failed to flush open data file to read unavailable column values: %w", err)
+		}
+		writer.dataWriter = rolled
+	}
+
+	return nil
+}
+
 func (w *ArrowWriter) flush(ctx context.Context, rw *RollingWriter, partitionKey string) error {
 	if rw.currentRowCount == 0 {
 		_ = rw.currentWriter.Close()

destination/iceberg/iceberg.go

@@ -33,6 +33,7 @@ type Iceberg struct {
 	server        *serverInstance          // shared Java server instance (per-process singleton)
 	schema        map[string]string        // schema for current thread associated with Java writer (col -> type)
 	writer        Writer                   // writer instance
+	toast         *toastResolver           // recovers values the source could not send; nil unless the stream keeps a row index (pos or dv)
 	// Why Schema On Thread Level?
 	// Schema on thread level is identical to the writer instance available in the Java server.
 	// It defines when to complete the Java writer and when schema evolution is required.
@@ -143,13 +144,22 @@ func (i *Iceberg) Setup(ctx context.Context, stream types.StreamInterface, _ any
 		return schema, nil, err
 	}
 
+	// nil in equality mode; shared by the writer, which buffers this thread's row locations
+	// in it, and the toast resolver, which looks rows up in it
+	indexThread := types.NewStreamIndexThread(options.TableIndex)
 	if i.config.UseArrowWrites {
-		i.writer, err = arrowwriter.New(ctx, i.options, i.partitionInfo, i.schema, i.stream, i.server, upsertMode)
+		i.writer, err = arrowwriter.New(ctx, i.options, i.partitionInfo, i.schema, i.stream, i.server, upsertMode, indexThread)
 		if err != nil {
 			return nil, nil, destination.WriteFailure(fmt.Errorf("failed to create arrow writer: %w", err))
 		}
 	} else {
-		i.writer = legacywriter.New(i.options, i.schema, i.stream, i.server, upsertMode)
+		i.writer = legacywriter.New(i.options, i.schema, i.stream, i.server, upsertMode, indexThread)
+	}
+
+	// Recovery needs the row's previous location, which only an upsert thread with a
+	// stream index has. Backfill and equality mode keep the placeholder.
+	if indexThread != nil && upsertMode {
+		i.toast = newToastResolver(options.ThreadID, i.stream, i.writer, indexThread, i.server.tableIndexClient)
 	}
 
 	return schema, &metadataState, nil
@@ -180,6 +190,10 @@ func (i *Iceberg) Close(ctx context.Context, finalMetadataState any) (err error)
 		}
 	}()
 
+	if i.toast != nil {
+		i.toast.Close()
+	}
+
 	select {
 	case <-ctx.Done():
 		// skip commit in case of context cancellation
@@ -261,6 +275,14 @@ func (i *Iceberg) Check(ctx context.Context) error {
 
 // validate schema change & evolution and removes null records
 func (i *Iceberg) FlattenAndCleanData(ctx context.Context, records []types.RawRecord) (bool, []types.RawRecord, any, error) {
+	// Resolve first, so schema detection, filters, partition values and the JSON row
+	// (normalization off) all see the recovered values.
+	if i.toast != nil {
+		if err := i.toast.Resolve(ctx, records); err != nil {
+			return false, nil, nil, fmt.Errorf("failed to recover unavailable column values: %w", err)
+		}
+	}
+
 	// extractSchemaFromRecords detects difference in current thread schema and the batch that being received
 	// Also extracts current batch schema
 	extractSchemaFromRecords := func(ctx context.Context, records []types.RawRecord) (bool, map[string]string, error) {

destination/iceberg/interface.go

@@ -13,4 +13,8 @@ type Writer interface {
 	// Abort discards whatever the writer staged outside Iceberg, for the paths
 	// that give up before Close can commit.
 	Abort()
+	// EnsureReadable closes any file in paths the Go side still has open, without
+	// committing it, so its rows can be read (an open file has no footer). Files written
+	// by Java are closed by Java itself when ReadRows asks for them.
+	EnsureReadable(ctx context.Context, paths []string) error
 }

destination/iceberg/legacy-writer/writer.go

@@ -24,17 +24,24 @@ type LegacyWriter struct {
 	upsertMode  bool
 }
 
-func New(options *destination.Options, schema map[string]string, stream types.StreamInterface, server internal.ServerClient, upsertMode bool) *LegacyWriter {
+func New(options *destination.Options, schema map[string]string, stream types.StreamInterface, server internal.ServerClient, upsertMode bool, indexThread *types.StreamIndexThread) *LegacyWriter {
 	return &LegacyWriter{
 		options:     options,
 		schema:      schema,
 		stream:      stream,
 		server:      server,
 		upsertMode:  upsertMode,
-		indexThread: types.NewStreamIndexThread(options.TableIndex),
+		indexThread: indexThread,
 	}
 }
 
+// EnsureReadable has nothing to do: Java writes the data files and knows which are still
+// open, so ReadRows closes those itself. Tracking them here would go stale whenever Java
+// rolls a file on its own.
+func (w *LegacyWriter) EnsureReadable(context.Context, []string) error {
+	return nil
+}
+
 func (w *LegacyWriter) Write(ctx context.Context, records []types.RawRecord) error {
 	protoSchema := make([]*proto.IcebergPayload_SchemaField, 0, len(w.schema))
 	for field, dType := range w.schema {

destination/iceberg/olake-iceberg-java-writer/pom.xml

@@ -34,6 +34,9 @@
     <version.googlebigdataoss>2.2.28</version.googlebigdataoss>
     <version.testcontainers>1.20.4</version.testcontainers>
     <version.protobuf>3.25.5</version.protobuf>
+    <!-- matches the parquet iceberg-parquet pulls in; needed at compile time to read
+         individual rows back by position (see tableIndex/RowReader) -->
+    <version.parquet>1.16.0</version.parquet>
     <version.log4j2>2.26.1</version.log4j2>
     <!-- deliberate family pins; see DEPENDENCY POLICY above -->
     <version.netty>4.1.137.Final</version.netty>
@@ -563,6 +566,12 @@
       <version>3.12.14</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-hadoop</artifactId>
+      <version>${version.parquet}</version>
+    </dependency>
+
     <dependency>
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java</artifactId>