Java: Add blob v2 write support with external blob references

Summary

Java SDK currently only supports reading blob v2 datasets and writing legacy blob v1 (via lance-encoding:blob metadata). We need full blob v2 write support, including external blob references with byte ranges.

Current State

Reading (✅ Supported)

• Dataset.takeBlobs() and Dataset.takeBlobsByIndices() work with blob v2
• BlobFile provides file-like access with read(), seek(), tell(), size()

Writing (⚠️ Legacy Only)

Java tests use legacy blob format:

// TestUtils.java:642-645
private static final String BLOB_META_KEY = "lance-encoding:blob";
private static final String BLOB_META_TRUE = "true";
Field blobField = new Field("blobs",
    new FieldType(true, ArrowType.LargeBinary.INSTANCE, null, 
        Map.of("lance-encoding:blob", "true")),
    Collections.emptyList());

Proposed API

1. Schema Helpers

import org.lance.schema.BlobField;

// Create a blob v2 field
Field blobField = BlobField.create("video");
Field nullableBlobField = BlobField.nullable("thumbnail");

Schema schema = new Schema(Arrays.asList(
    Field.notNullable("id", new ArrowType.Int(64, true)),
    BlobField.nullable("blob")
));

2. Blob Value Builder

import org.lance.blob.Blob;
import org.lance.blob.BlobArray;

// Build blob values - supports inline bytes, external URIs, and byte ranges
BlobArray blobs = BlobArray.builder()
    .add(new byte[] {0x01, 0x02, 0x03})                    // inline bytes
    .add("s3://bucket/path/video.mp4")                      // external URI (full file)
    .add(Blob.fromUri("s3://bucket/archive.tar")            // external with byte range
              .position(4096)
              .size(8192))
    .addNull()                                              // null value
    .build(allocator);

// Or from lists
BlobArray blobs = BlobArray.fromBytes(allocator, listOfByteArrays);
BlobArray blobs = BlobArray.fromUris(allocator, listOfUriStrings);

3. Writing to Dataset (Simple Path)

try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
    // ... populate id column ...
    
    // Set blob column using BlobArray
    BlobArray blobArray = BlobArray.builder()
        .add(Blob.fromUri("s3://bucket/video1.mp4"))
        .add(Blob.fromUri("s3://bucket/archive.tar", 4096, 8192))
        .build(allocator);
    root.setVector("blob", blobArray.toVector());
    root.setRowCount(2);
    
    WriteParams params = new WriteParams.Builder()
        .withDataStorageVersion("2.2")
        .withAllowExternalBlobOutsideBases(true)  // for absolute external URIs
        .build();
    
    Dataset ds = Dataset.write(allocator, root, "/path/to/dataset", params);
}

4. Writing via Fragment (Distributed Engine Integration)

This is the preferred path for Spark/Trino/distributed engines where workers write fragments independently and a coordinator commits them.

// Worker: Create fragment with blob data
try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
    BigIntVector idVec = (BigIntVector) root.getVector("id");
    
    BlobArray blobArray = BlobArray.builder()
        .add(Blob.fromUri("s3://bucket/video1.mp4"))
        .add(Blob.fromUri("s3://bucket/video2.mp4"))
        .add(Blob.fromUri("s3://bucket/archive.tar", 4096, 8192))  // byte range
        .build(allocator);
    
    for (int i = 0; i < 3; i++) {
        idVec.setSafe(i, i);
    }
    root.setRowCount(3);
    
    WriteParams params = new WriteParams.Builder()
        .withDataStorageVersion("2.2")
        .withAllowExternalBlobOutsideBases(true)
        .build();
    
    // Worker creates fragment, returns serializable metadata
    List<FragmentMetadata> fragments = Fragment.create(
        datasetPath, allocator, root, params);
    
    // Send fragment metadata to coordinator (e.g., via Spark driver)
    return fragments;
}

// Coordinator: Commit fragments from all workers
List<FragmentMetadata> allFragments = collectFromWorkers();
FragmentOperation.Append appendOp = new FragmentOperation.Append(allFragments);
Dataset ds = Dataset.commit(allocator, datasetPath, appendOp, Optional.of(readVersion));

5. Spark Integration Example

// Spark DataSourceV2 Write implementation
public class LanceBlobWriterFactory implements DataWriterFactory {
    
    @Override
    public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
        return new LanceBlobDataWriter(partitionId, schema, datasetPath);
    }
}

public class LanceBlobDataWriter implements DataWriter<InternalRow> {
    private final List<Blob> pendingBlobs = new ArrayList<>();
    
    @Override
    public void write(InternalRow row) {
        // Row contains: id (long), blob_uri (string), offset (long), size (long)
        String uri = row.getString(1);
        long offset = row.getLong(2);
        long size = row.getLong(3);
        
        pendingBlobs.add(Blob.fromUri(uri, offset, size));
    }
    
    @Override
    public WriterCommitMessage commit() {
        BlobArray blobArray = BlobArray.fromBlobs(allocator, pendingBlobs);
        // ... build VectorSchemaRoot with blob column ...
        
        List<FragmentMetadata> fragments = Fragment.create(
            datasetPath, allocator, root, writeParams);
        
        return new LanceWriterCommitMessage(fragments);
    }
}

// Driver commits all fragments
public class LanceBlobBatchWrite implements BatchWrite {
    @Override
    public void commit(WriterCommitMessage[] messages) {
        List<FragmentMetadata> allFragments = Arrays.stream(messages)
            .flatMap(m -> ((LanceWriterCommitMessage) m).getFragments().stream())
            .collect(Collectors.toList());
        
        Dataset.commit(allocator, datasetPath, 
            new FragmentOperation.Append(allFragments), 
            Optional.of(readVersion));
    }
}

Implementation Notes

Blob V2 Struct Schema

The blob v2 Arrow schema is a struct with these fields:

• data (LargeBinary, nullable) - inline bytes
• uri (Utf8, nullable) - external URI
• position (UInt64, nullable) - byte offset for external
• size (UInt64, nullable) - byte length for external

JNI Considerations

• BlobArray.build() should produce an Arrow StructArray matching the blob v2 schema
• The Rust BlobPreprocessor handles routing to inline/packed/dedicated/external based on size
• External URIs are validated and normalized by ExternalBaseResolver on the Rust side

WriteParams Additions

public class WriteParams {
    // Existing fields...
    
    // New blob v2 fields
    private String dataStorageVersion;           // "2.2" for blob v2
    private boolean allowExternalBlobOutsideBases;  // allow absolute external URIs
}

Acceptance Criteria

• BlobField.create() / BlobField.nullable() helpers for schema construction
• Blob class with fromBytes(), fromUri(uri), fromUri(uri, position, size)
• BlobArray builder for constructing blob column values
• WriteParams.withDataStorageVersion("2.2") to enable blob v2
• WriteParams.withAllowExternalBlobOutsideBases(true) for external URIs
• Dataset.write() works with blob v2 columns
• Fragment.create() works with blob v2 columns (distributed write path)
• Dataset.commit() correctly commits fragments with blob v2 data
• Integration tests covering inline, external URI, and external byte range cases
• Documentation with examples for both simple and distributed write paths

Start with the Java tests and TestUtils.java:642-645, then trace the existing Dataset.takeBlobs() and Dataset.takeBlobsByIndices() implementations and BlobFile API. Define the BlobField, Blob, BlobArray, WriteParams, Dataset.write(), Fragment.create(), and Dataset.commit() entry points against the listed acceptance criteria. Add integration coverage for inline values, external URIs, byte ranges, and both simple and distributed write paths.