fix: unify schema across batches in JSONStreamDatasource to handle null → concrete type evolution

[fengrui-z]

Summary

Fixes #936

JSONStreamDatasource._read_stream locks the schema from the first batch and reuses it for all subsequent batches. When an
early batch infers a nested field as null (e.g. meta.url = null) and a later batch introduces a concrete type (e.g.
string), the forced cast from string to null fails with ArrowInvalid.

This is a correctness bug in DJ's custom JSON streaming ingestion path. Ray's native ray.data.read_json handles the same input correctly.

Root Cause

# Before: first batch locks schema, all subsequent batches forced to it
table = pyarrow.Table.from_batches([batch], schema=schema)
if schema is None:
    schema = table.schema  # locked forever

Fix

1. Remove the first-batch schema lock — create table without forced schema
2. Use pyarrow.unify_schemas to merge schemas across batches, allowing null → concrete type promotion
3. After unification, cast the batch to the unified schema for consistency

  # After: schema evolves across batches
  table = pyarrow.Table.from_batches([batch])
  if schema is None:
      schema = table.schema
  else:
      unified = pyarrow.unify_schemas([schema, table.schema])
      if not unified.equals(schema):
          schema = unified
      table = pyarrow.Table.from_batches([batch], schema=schema)

unify_schemas internally delegates to Arrow C++ UnifyTypes, which promotes null to the concrete type and recursively handles nested structs.

Test Plan

• Verify the minimal repro from [Bug] JSONStreamDatasource locks first-batch schema and fails on later null -> concrete type evolution #936 passes with this fix
• Verify ray.data.read_json and read_json_stream produce consistent results on mixed-null JSONL
• Verify no regression on JSONL files with uniform schema
• Verify no regression on JSONL files with nested structs

See #936 for the minimal reproduction script.

[gemini-code-assist]

Code Review

This pull request updates the _read_stream method in ray_dataset.py to support schema unification when reading batches from a stream. This allows the system to handle batches with varying but compatible schemas. A review comment suggests refactoring the implementation to reduce code duplication by consolidating the pyarrow.Table creation after the final schema has been determined.