[flink] Support storing BlobDescriptor in lookup join for BLOB tables

[wwj6591812]

Purpose

When performing lookup join against a Paimon BLOB table (e.g., storing images/videos), the full blob content is materialized via BlobSerializer.serialize() → blob.toData() and written into local RocksDB during bootstrap. For tables with large blob fields, this causes:

• Extremely high local disk usage (e.g., 2 billion images × 200KB = ~400TB total, ~400GB per subtask)
• Page-size overflow errors (single records exceeding the 64KB default page)
• Prolonged bootstrap time leading to TaskManager heartbeat timeouts

This PR introduces a new table option lookup.blob-as-descriptor (default false). When enabled:

1. BLOB fields are stored as their lightweight BlobDescriptor bytes (~130 bytes containing file URI, offset, and length) instead of the full blob content.
2. The RowType used for value serialization replaces BLOB with VARBINARY, so InternalSerializers.create() produces BinarySerializer (calls getBinary()) instead of BlobSerializer (calls getBlob().toData()).
3. BlobRef.toDescriptor() is a pure in-memory operation (no I/O), so bootstrap performance is not affected.

The downstream consumer receives the serialized BlobDescriptor bytes and can resolve the actual blob content on demand via a UDF reading from DFS.

Impact: Per-subtask RocksDB storage drops from ~400GB to ~3.6GB (over 100x reduction), making lookup join feasible for tables with large BLOB columns.

Tests
Existing unit tests pass (no behavioral change when the option is disabled).
TODO: Add integration test for NoPrimaryKeyLookupTable with lookup.blob-as-descriptor = true to verify BlobDescriptor bytes are correctly stored and returned.

Example usage:

CREATE TABLE dim_images (
  url STRING,
  image BLOB
) WITH (
  'lookup.blob-as-descriptor' = 'true'
);

SELECT s.*, resolve_blob(d.image) AS image_data
FROM stream_table s
LEFT JOIN dim_images FOR SYSTEM_TIME AS OF s.proc_time AS d
ON s.url = d.url;

[JingsongLi]

This common bulk-load path now wraps the value row, but the primary-key cache still creates tableState with InternalSerializers.create(projectedType) and its incremental refreshRow stores the raw row. With that serializer, BlobSerializer calls getBlob().toData(), so lookup.blob-as-descriptor=true still materializes and caches the full blob for primary-key and secondary-index lookup tables. Please apply cacheValueRowType() and wrapForCache(row) to the primary-key table state path as well.

[wwj6591812]

Thanks for the review! You're right — for code consistency and defensive completeness, I should apply cacheValueRowType() and wrapForCache(row) to the primary-key and secondary-index paths as well.

That said, I want to note that in practice, BLOB fields can only exist on append-only tables (the schema validation enforces: BLOB → data-evolution → row-tracking → no primary key + bucket=-1), so PrimaryKeyLookupTable and SecondaryIndexLookupTable will never encounter BLOB fields at runtime — blobFieldPositions will always be empty and these calls become no-ops.

But I agree it's better to keep the code paths consistent. I've updated the PR to apply cacheValueRowType() in createTableState() and wrapForCache(row) in refreshRow() for both PrimaryKeyLookupTable and SecondaryIndexLookupTable. Thanks!

[JingsongLi]

This assumes the row entering the lookup cache already carries descriptor-backed BLOB values. The lookup reader still reads table data using the normal blob-as-descriptor option, while this new lookup.blob-as-descriptor option only changes the cache serializer/wrapper. For a normal BLOB table with blob-as-descriptor=false, BlobFormatReader returns BlobData; then this call to blob.toDescriptor() throws Blob data can not convert to descriptor during bootstrap/refresh. Could we either make the lookup reader read BLOB fields in descriptor mode when this option is enabled, or reject/document the unsupported combination before the cache starts loading?

[wwj6591812]

Thanks for the excellent catch @JingsongLi!

You're absolutely right — for a normal BLOB table with blob-as-descriptor=false, BlobFormatReader reads inline data and returns BlobData, and calling toDescriptor() on it throws "Blob data can not convert to descriptor".

I've chosen Option A (making the lookup reader read in descriptor mode): In FullCacheLookupTable.bootstrap(), when lookup.blob-as-descriptor=true, I now copy the table with blob-as-descriptor=true forced in the options before passing it to LookupStreamingReader. This makes BlobFormatReaderFactory close the input stream and pass in=null to BlobFormatReader, which then uses Blob.fromFile(fileIO, path, offset, length) → creates BlobRef → toDescriptor() works correctly.

This approach supports both scenarios:
1、Table written with blob-as-descriptor=true: works as before (reader already in descriptor mode)
2、Table written with blob-as-descriptor=false: now also works — the reader returns BlobRef pointing to the data location in the blob file, and the serialized descriptor enables downstream consumers to range-read the actual content on demand.

[wwj6591812]

@JingsongLi
Hi，Please CC, Thx

[JingsongLi]

+1