feat: preserve blob data through Spark shuffle during JOIN + INSERT INTO

[beinan]

Summary

• When blob columns flow through Spark's shuffle (e.g., INSERT INTO target SELECT ... FROM source_a JOIN source_b), the actual blob data was previously lost because getBinary() returned empty byte arrays (new byte[0]). The target table ended up with zero-length blobs.
• This PR introduces a blob reference mechanism that preserves blob data through Spark's shuffle without materializing the full blob bytes (which could be MBs/GBs per row and would kill shuffle performance).

How it works

Read side — compact blob references flow through shuffle:

• Instead of returning byte[0], blob columns now serialize compact ~100-byte BlobReference descriptors containing the source dataset URI, column name, and row address
• The scanner automatically requests _rowaddr when blob columns are present, and strips it from the output after extracting the row addresses
• These references are small enough to flow through Spark shuffle without overhead

Write side — blob references are resolved to actual bytes:

• A new BlobResolvingLargeBinaryWriter detects BlobReference magic headers in incoming binary values
• It opens the source dataset (with caching), calls Dataset.takeBlobs() to fetch the actual blob bytes
• The real blob data is then written to the target table through the normal blob write path

Key files

File	Status	Description
BlobReference.java	NEW	Compact serializable blob reference format (LBRF magic + dataset URI + column name + row address)
BlobReferenceResolver.java	NEW	Write-side resolver: opens source dataset → takeBlobs() → returns actual bytes
BaseBlobJoinTest.java	NEW	Tests for blob preservation through simple INSERT INTO SELECT, JOIN + INSERT, and mixed blob/non-blob JOINs
BlobJoinTest.java (3.4 + 3.5)	NEW	Concrete test implementations
BlobStructAccessor.java	Modified	Added blob reference context (dataset URI, column name, row addresses) and getBlobReference()
LanceArrowColumnVector.java	Modified	getBinary() returns blob reference instead of byte[0]
LanceFragmentScanner.java	Modified	Detects blob columns, requests _rowaddr, exposes dataset URI and blob column names
LanceFragmentColumnarBatchScanner.java	Modified	Extracts row addresses from _rowaddr, sets blob context on BlobStructAccessor, strips implicit _rowaddr
LanceArrowWriter.scala	Modified	Added BlobResolvingLargeBinaryWriter for blob-encoded columns
BaseBlobCreateTableTest.java	Modified	Updated assertions for new blob reference behavior

Test plan

• BlobJoinTest.testBlobPreservedDuringInsertIntoSelect — Simple INSERT INTO target SELECT FROM source preserves blob data
• BlobJoinTest.testBlobPreservedDuringJoinAndInsert — JOIN of two blob tables preserves both blob columns in target
• BlobJoinTest.testNonBlobColumnsPreservedDuringJoinWithBlobs — Non-blob columns survive JOIN+INSERT alongside blob columns
• All existing BlobCreateTableTest tests pass (9/9)

🤖 Generated with Claude Code

[hamersaw]

Thanks for the PR! Should be a pretty significant help in reducing memory utilization. A few comments to iron things out, but I think I need to dive through the regular read paths to make sure this doesn't negatively affect performance there as well.

[hamersaw]

Diving through this a bit more. I'm not sure there's a reason to have a LargeBinaryWriter (existing) and a BlobResolvingLargeBinaryWriter (new). Right now, if we're reading from a blob column and writing to a regular binary we're going to dump a ton of BlobReferences rather than the actual bytes because it uses the LargeBinaryWriter which doesn't resolve the BlobReference. I think we should collapse these into a single writer to mitigate future issues.

[beinan]

Good point — collapsed BlobResolvingLargeBinaryWriter into LargeBinaryWriter so there is now a single writer that always resolves blob references via resolveIfNeeded(). This means blob references are correctly resolved regardless of whether the target column is blob-encoded or regular binary, preventing silent data corruption.

[beinan]

@hamersaw All feedback has been addressed and CI is fully green. Ready for another look when you get a chance!

[hamersaw]

Thanks for the iteration here, I think we're getting close!

[hamersaw]

Per-row BigInteger allocation extracting row addresses: extractRowAddresses() runs once per Arrow batch and loops over every row calling rowAddrVector.getObjectNoOverflow(i), which returns a freshly allocated BigInteger that is immediately reduced to a long. For a multi-million-row scan with a blob column this is millions of throwaway BigInteger allocations purely to read a 64-bit value, adding avoidable GC pressure on the scan hot path.

Fix: Use UInt8Vector.get(i), which returns a primitive long directly, instead of getObjectNoOverflow(i).longValue().

[beinan]

Fixed — now uses UInt8Vector.get(i) which returns a primitive long directly, avoiding per-row BigInteger allocation.

[hamersaw]

Blob resolve batching is defeated — one takeBlobs() round-trip per row: resolve() adds the pending blob to its group, and only batch-flushes when group.size() >= batchSize (64). But the very next statement if (pending.resolved == null) flushes immediately whenever the batch threshold was NOT hit — which is every call in steady state, because flushGroup() also removes the group from pendingBlobs so it never accumulates. resolveIfNeeded()/resolve() must return bytes synchronously to LargeBinaryWriter.setValue(), so deferral is impossible by construction. Net effect: dataset.takeBlobs() is invoked with exactly one row address per blob row instead of once per 64. On a JOIN+INSERT INTO carrying millions of blob cells through shuffle, this is ~64x the JNI crossings and object-store read round-trips the batching was designed to eliminate, making the feature impractical at scale.

Fix: Make resolution genuinely deferred: have getBinary()/the writer accumulate references and resolve them in true batches (e.g. a two-pass write, or buffer references and back-fill the vector after a single batched takeBlobs per group), rather than resolving synchronously per row.

[beinan]

Completely reworked — the writer now does genuine two-pass batched resolution. setValue() buffers blob references with empty placeholders in the vector. finish() groups all pending references by (datasetUri, columnName) and calls takeBlobs() once per group, then back-fills the vector with resolved bytes. This reduces JNI+I/O from O(N) to O(groups).

[hamersaw]

Every non-empty large-binary value written to a blob column is now passed through resolver.resolveIfNeeded(), which calls BlobReference.isBlobReference() — a 4-byte magic-header ('L','B','R','F') sniff plus a >=17 byte length check. Legitimate user-supplied blob bytes whose first four bytes happen to be 0x4C424652 and whose length is >=17 (e.g. an arbitrary file/binary payload) will be misclassified as a serialized BlobReference. deserialize() then either throws IllegalArgumentException (unknown version byte) — failing the INSERT with a RuntimeException — or, on a coincidental version match, parses bogus length-prefixed strings and a garbage dataset URI, causing a takeBlobs() failure or silently wrong data being written. This affects first-time user binary data on a plain INSERT into a blob column, not reference round-trips.

Fix: Disambiguate references from real payloads instead of content sniffing — e.g. carry an out-of-band flag/column indicating the value is a reference, or make the magic long/structured enough (and validate full structure) that collision with real data is negligible, and treat a failed parse as 'not a reference' rather than throwing.

[beinan]

Addressed in two ways: (1) Extended magic from 4 bytes (LBRF) to 8 bytes (LANCEREF), making random collision probability ~1 in 2^64. (2) isBlobReference() now does full structural validation — checks version byte, verifies both length-prefixed strings fit within the total byte count, and confirms the remaining bytes exactly equal colLen + 8 (the row address). Random binary data that happens to start with LANCEREF will fail the structural check.

[hamersaw]

reset() closes resolver, discarding the source-dataset cache: LargeBinaryWriter.reset() calls resolver.close(), which closes and clears BlobReferenceResolver.datasetCache — the cache whose stated purpose is 'avoid re-opening the same dataset for every row'. Arrow field writers are reset between batches, so the source Lance dataset (manifest + metadata I/O, JNI open) is re-opened on the first blob of every subsequent batch instead of once per task. Combined with the broken batching above, the write path repeatedly re-opens the source dataset and issues per-row blob fetches.

Fix: Release the resolver (and its dataset cache) once at writer finish/close, not on every reset(); reset() should only flush pending state, keeping cached datasets open for the task lifetime.

[beinan]

Fixed — reset() now only clears the pending blob reference buffer. The resolver (and its dataset cache) lives for the lifetime of the writer and is never closed/recreated between batches. Datasets opened on the first batch stay cached for all subsequent batches in the same task.