[Data] Write Delta timestamps as microseconds in write_delta

[HirokiNariyoshi]

Description

DeltaDatasink writes Parquet with PyArrow directly and never converts
timestamps to microseconds, which deltalake's own write_deltalake does.
Any s, ms, or ns timestamp, whether top-level, nested in a struct,
list, or map, or dictionary-encoded, therefore breaks the write:

• deltalake 1.5.0 (Ray's pinned version): the commit fails with
  Invalid data type for Delta Lake after the workers have already written
  their Parquet files, leaving those files orphaned.
• deltalake 1.6.x: ns commits a log schema of us over ns files, so the
  table's schema disagrees with its Parquet files. Timezone-aware ns
  commits a table with the timestampNanos reader feature, which the
  deltalake reader rejects.

This PR casts every timestamp, including nested and dictionary-encoded ones,
to timestamp[us] (timezone-aware ones in UTC) on each worker before
writing. It also converts the schema captured in on_write_start, which is
what an empty write commits. Sub-microsecond precision is truncated, since
Delta can't store it; out-of-range values still raise. Nanosecond timestamps are always
truncated to us, even if deltalake.enable_nanosecond_timestamps() (experimental, deltalake ≥ 1.6)
has been called, since Delta's stable timestamp type is microseconds.

The change is limited to DeltaDatasink and its tests.

Repro on master with deltalake 1.5.0:

import pyarrow as pa, ray
t = pa.table({"t": pa.array([1704110400_123456789], pa.timestamp("ns"))})
ray.data.from_arrow(t).write_delta("/tmp/delta_ts")
# Schema error: Invalid data type for Delta Lake: Timestamp(ns)

Related issues

None. I found this while reading DeltaDatasink. I searched open and closed
issues and PRs for write_delta, DeltaDatasink, delta_datasink,
timestampNanos, and "Invalid data type for Delta Lake", and found no
duplicate.

Additional information

Tests run locally:

• pytest python/ray/data/tests/datasource/test_write_delta.py: 112 passed
  on deltalake 1.5.0 and on 1.6.1.
• The 11 new tests (every unit with and without a timezone; struct, list,
  large_list, fixed_size_list, map, and dictionary-encoded columns; an ns
  append onto a us table; an empty dataset; out-of-range values; timestamp map keys)
  fail on master and pass with this change.
• The new helpers also work on pyarrow 15.0.0, the lowest version the test
  file allows.
• pre-commit run --files <changed files>: pass.

AI assistance was used to draft the change.
I reviewed every changed line and ran the tests above.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[gemini-code-assist]

Code Review

This pull request introduces automatic conversion of PyArrow timestamp types (including nested ones in structs, lists, and maps) to microsecond precision (and UTC for timezone-aware ones) when writing to Delta Lake, ensuring compatibility with Delta Lake's supported timestamp formats. The reviewer suggested extending this conversion to handle dictionary-encoded timestamp types to make the implementation more robust.

[gemini-code-assist]

Code Review

This pull request introduces automatic conversion of PyArrow timestamp types, including nested ones, to microsecond precision (and UTC for timezone-aware ones) when writing to Delta Lake. This ensures compatibility with Delta Lake's supported timestamp formats. Feedback points out a bug in the fixed-size list conversion logic where PyArrow would raise a ValueError if a Field object is passed with a specified list size, and suggests using the resolved DataType directly.

[Yin-1022]

I've tested _cast_to_delta_schema() with an out-of-range timestamp[s] value (9223372036855), and it correctly raises ArrowInvalid instead of silently overflowing. Just sharing the result in case it is helpful.

Since the docstring explicitly guarantees this behavior, would it be worth adding a regression test for this boundary case? It could help prevent accidental changes to the overflow behavior in future refactoring.

Here's my attempt on that regression test

def test_timestamp_overflow_rejected():
    import pyarrow as pa

    from ray.data._internal.datasource.delta_datasink import (
        _cast_to_delta_schema,
    )

    table = pa.table({
        "t": pa.array(
            [9_223_372_036_855],
            type=pa.timestamp("s"),
        ),
    })

    with pytest.raises(pa.ArrowInvalid, match="out of bounds"):
        _cast_to_delta_schema(table)

Also, _to_delta_type() recursively converts both map keys and values, but the nested timestamp test only covers timestamp values with string keys. Would a timestamp-keyed map also be a supported case for this conversion? If so, could a test for that path be added, or clarify whether Delta Lake supports it?

None of them are blockers, just suggestion for test coverage.

[HirokiNariyoshi]

I've tested _cast_to_delta_schema() with an out-of-range timestamp[s] value (9223372036855), and it correctly raises ArrowInvalid instead of silently overflowing. Just sharing the result in case it is helpful.

Since the docstring explicitly guarantees this behavior, would it be worth adding a regression test for this boundary case? It could help prevent accidental changes to the overflow behavior in future refactoring.

Here's my attempt on that regression test

def test_timestamp_overflow_rejected():
    import pyarrow as pa

    from ray.data._internal.datasource.delta_datasink import (
        _cast_to_delta_schema,
    )

    table = pa.table({
        "t": pa.array(
            [9_223_372_036_855],
            type=pa.timestamp("s"),
        ),
    })

    with pytest.raises(pa.ArrowInvalid, match="out of bounds"):
        _cast_to_delta_schema(table)

Also, _to_delta_type() recursively converts both map keys and values, but the nested timestamp test only covers timestamp values with string keys. Would a timestamp-keyed map also be a supported case for this conversion? If so, could a test for that path be added, or clarify whether Delta Lake supports it?

None of them are blockers, just suggestion for test coverage.

Thanks for testing this! Both addressed in d7a5a2e

test_timestamp_conversion_rejects_out_of_range_values, based on your test, using 9,223,372,036,855 s, the smallest whole second whose microsecond value overflows int64.

Map keys are converted, and timezone-aware keys commit fine, so I added a test for that. Naive timestamp keys can’t be written yet, independent of this PR. when a naive timestamp appears only as a map key, deltalake doesn’t add the timestampNtz feature, so the commit fails even with microsecond keys. write_deltalake fails the same way on 1.5.0 and 1.6.1.