Repository: ray-project/ray
Issue category: Bug Report

Issue:
[Data] Inconsistent behavior with Ray Data and timestamps
### What happened + What you expected to happen

*This example has been provided as a gist instead of in-line code to make it easier to reproduce and trace.

Two odd behaviors are identified:
1. When working with flat-structured datasets, ray data performs destructive type conversions when using the `.map()` API.
2. When working with nest-structed datasets (ie. data with sub-structs), ray data performs destructive type conversions when using the `.map_batches()` (or `.take_batches()`) API.

The first issue is isolated to just `ray.data`, while the second is demonstrated to fail with the `pandas` API as well.

This currently blocks me from using ray data as I cannot extract timestamps consistently across my data sources.

### Versions / Dependencies

Name: numpy
Version: 1.24.4

Name: pandas
Version: 2.0.3

Name: pyarrow
Version: 17.0.0

Name: ray
Version: 2.10.0

Name: host
Version: Darwin [hostname-redacted] 23.5.0 Darwin Kernel Version 23.5.0: Wed May  1 20:13:18 PDT 2024; root:xnu-10063.121.3~5/RELEASE_ARM64_T6030 arm64

### Reproduction script

Gist: https://gist.github.com/NellyWhads/fdfb261a027be7e7bc87bec91d9e9035

### Issue Severity

High: It blocks me from completing my task.


Hints:
@jcotant1 the crux for both issues is around how we handle nanoseconds

1. for the 1st behavior: `.map()` + `.take_batch(..., batch_format="pandas")` truncates the time information (see the **ns** field)
2. for the 2nd behavior (timestamps within a struct): 
    - currently it casts timestamps with nanoseconds to integers while those with lower precisions are preserved as Python datetime objects. This prohibits any caller from relying on the type, and requires them to coerce the value back to the appropriate type before every use
    - since Python's datetime and timedelta objects don't support nanoseconds, @NellyWhads asked if Ray can standardize around np.datetime64 and/or pd.Timestamp (perhaps depending on the batch format)?
Based on the warnings logged from ray data, `map_batch()` / `take_batch()` is the API users should target since it is more consistent with the interface to the ray training runner. If this is the case, the 2nd issue is of much higher priority than the first.
Test case for this.

```
import ray
import pandas as pd

# Initialize Ray
ray.init(ignore_reinit_error=True)

# Sample flat-structured dataset (pandas DataFrame)
df = pd.DataFrame({
    'id': [1, 2, 3],
    'timestamp': pd.to_datetime(['2024-01-01 00:00:00.123456789',
                                 '2024-01-02 00:00:00.987654321',
                                 '2024-01-03 00:00:00.111222333']),
    'value': [10.123456789, 20.987654321, 30.111222333]
})

# Ray data setup
ray_data = ray.data.from_pandas(df)

# Define a function to apply with .map()
def process_data(row):
    # Example of applying a transformation
    row['timestamp'] = row['timestamp'] + pd.Timedelta(1, 'ns')  # add 1ns to timestamp
    return row

# Apply the .map() function
result = ray_data.map(process_data)

# Convert back to pandas for inspection
processed_df = result.to_pandas()

# Print the original and processed data
print("Original DataFrame:")
print(df)
print("\nProcessed DataFrame after .map():")
print(processed_df)

```

```
Original DataFrame:
   id                     timestamp      value
0   1 2024-01-01 00:00:00.123456789  10.123457
1   2 2024-01-02 00:00:00.987654321  20.987654
2   3 2024-01-03 00:00:00.111222333  30.111222

Processed DataFrame after .map():
   id                  timestamp      value
0   1 2024-01-01 00:00:00.123456  10.123457
1   2 2024-01-02 00:00:00.987654  20.987654
2   3 2024-01-03 00:00:00.111222  30.111222
```
for the 2nd issue, i believe it's because Ray Data relies on PyArrow's Pandas integration. PyArrow doesn't seem to respect the conversion table [here](https://arrow.apache.org/docs/python/pandas.html#arrow-pandas-conversion) if the timestamps are inside a StructArray 

Repro:
```
import pandas as pd
import pyarrow as pa

base_timestamp = pd.Timestamp("2024-01-01")

start_times = pa.array([base_timestamp + pd.Timedelta(nanoseconds=i) for i in range(0, 5)], type=pa.timestamp('ns'))
end_times = pa.array([base_timestamp + pd.Timedelta(nanoseconds=j) for j in range(1, 6)], type=pa.timestamp('ns'))

struct_array = pa.StructArray.from_arrays(
    [start_times, end_times], 
    fields=[
        pa.field("start_time", pa.timestamp('ns')), 
        pa.field("end_time", pa.timestamp('ns'))
        ]
    )
# when not using StructArray, conversion works as expected per https://arrow.apache.org/docs/python/pandas.html#arrow-pandas-conversion
table_without_struct = pa.Table.from_arrays([start_times, end_times], ["start_timestamp", "end_timestamp"])
df_without_struct = table_without_struct.to_pandas()
print(df_without_struct.iloc[0]) # returns pd.Series with dtype=datetime64[ns]

print("--------")
# however when using pyarrow StructArray, pyarrow converts timestamps with nanosecond precision to ints rather than respecting the conversion table
# this is true even without using Ray
table_with_struct = pa.Table.from_arrays([struct_array], ["timestamps"])
df_with_struct = table_with_struct.to_pandas()
print(df_with_struct["timestamps"].iloc[0]) # returns a dict of ints
I believe that's the exact issue. I'm unfortunately not sure if pandas will be able to iterate on a fix quickly in this case. Perhaps the ray team has a better relationship with the pandas devs than I (a relative stranger) may have?

A pandas-centric fix also means that users are forced to use a specific version of pandas in their environments.
Would you be able to convert the timestamp to int64? This should handle nanosecs precision.

```
import ray
import pandas as pd

# Initialize Ray
ray.init(ignore_reinit_error=True)

# Sample flat-structured dataset (pandas DataFrame)
df = pd.DataFrame({
    'id': [1, 2, 3],
    'timestamp': pd.to_datetime(['2024-01-01 00:00:00.123456789',
                                 '2024-01-02 00:00:00.987654321',
                                 '2024-01-03 00:00:00.111222333']),
    'value': [10.123456789, 20.987654321, 30.111222333]
})

# Convert timestamp to nanoseconds (int64) before passing to Ray
df['timestamp_ns'] = df['timestamp'].astype('int64')  # Convert to nanoseconds since epoch
df = df.drop(columns=['timestamp'])  # Drop original timestamp if not needed

# Ray data setup
ray_data = ray.data.from_pandas(df)

# Define a function to apply with .map()
def process_data(row):
    # Add 1 nanosecond
    row['timestamp_ns'] += 1
    return row

# Apply the .map() function
result = ray_data.map(process_data)

# Convert back to pandas for inspection
processed_df = result.to_pandas()

# Restore timestamp from nanoseconds
processed_df['timestamp'] = pd.to_datetime(processed_df['timestamp_ns'])  # Convert back to timestamp
processed_df = processed_df.drop(columns=['timestamp_ns'])  # Drop intermediate column if not needed

# Print the original and processed DataFrames
print("Original DataFrame:")
print(df)
print("\nProcessed DataFrame after .map():")
print(processed_df)
```


```
Original DataFrame:
   id      value         timestamp_ns
0   1  10.123457  1704067200123456789
1   2  20.987654  1704153600987654321
2   3  30.111222  1704240000111222333

Processed DataFrame after .map():
   id      value                     timestamp
0   1  10.123457 2024-01-01 00:00:00.123456790
1   2  20.987654 2024-01-02 00:00:00.987654322
2   3  30.111222 2024-01-03 00:00:00.111222334
```
Unfortunately, this means that whoever uses the data is now responsible for remembering to convert it back to a timestamp (in each transform they `map_batch()` onto the data).

This is how we currently "patch" the behavior, but it requires an extra layer of API that is rather expensive to maintain instead of just using ray data as a product directly.
I actually run into the same bug, however, I only became aware when ray data raised an error during `pa.concat_tables` while unifying the schemas of two blocks. 

This happens in cases there the type of the timestamp columns of two blocks are inferred differently:
`2024-01-01 00:00:00.123456790 -> 2024-01-01 00:00:00.123456 -> timestamp[us] (microseconds)`, however,
`2024-01-01 00:00:00.123000090 -> 2024-01-01 00:00:00.123 -> timestamp[ms] (milliseconds)`

Patch:
diff --git a/python/ray/data/_internal/numpy_support.py b/python/ray/data/_internal/numpy_support.py
index d04060fc831e8..28cb0e537fa76 100644
--- a/python/ray/data/_internal/numpy_support.py
+++ b/python/ray/data/_internal/numpy_support.py
@@ -45,28 +45,94 @@ def validate_numpy_batch(batch: Union[Dict[str, np.ndarray], Dict[str, list]]) -
 
 
 def _detect_highest_datetime_precision(datetime_list: List[datetime]) -> str:
-    highest_precision = "D"
+    """Detect the highest precision for a list of datetime objects.
+
+    Args:
+        datetime_list: List of datetime objects.
+
+    Returns:
+        A string representing the highest precision among the datetime objects
+        ('D', 's', 'ms', 'us', 'ns').
+    """
+    # Define precision hierarchy
+    precision_hierarchy = ["D", "s", "ms", "us", "ns"]
+    highest_precision_index = 0  # Start with the lowest precision ("D")
 
     for dt in datetime_list:
-        if dt.microsecond != 0 and dt.microsecond % 1000 != 0:
-            highest_precision = "us"
+        # Safely get the nanosecond value using getattr for backward compatibility
+        nanosecond = getattr(dt, "nanosecond", 0)
+        if nanosecond != 0:
+            current_precision = "ns"
+        elif dt.microsecond != 0:
+            # Check if the microsecond precision is exactly millisecond
+            if dt.microsecond % 1000 == 0:
+                current_precision = "ms"
+            else:
+                current_precision = "us"
+        elif dt.second != 0 or dt.minute != 0 or dt.hour != 0:
+            # pyarrow does not support h or m, use s for those cases to
+            current_precision = "s"
+        else:
+            current_precision = "D"
+
+        # Update highest_precision_index based on the hierarchy
+        current_index = precision_hierarchy.index(current_precision)
+        highest_precision_index = max(highest_precision_index, current_index)
+
+        # Stop early if highest possible precision is reached
+        if highest_precision_index == len(precision_hierarchy) - 1:
             break
-        elif dt.microsecond != 0 and dt.microsecond % 1000 == 0:
-            highest_precision = "ms"
-        elif dt.hour != 0 or dt.minute != 0 or dt.second != 0:
-            # pyarrow does not support h or m, use s for those cases too
-            highest_precision = "s"
 
-    return highest_precision
+    return precision_hierarchy[highest_precision_index]
+
+
+def _convert_to_datetime64(dt: datetime, precision: str) -> np.datetime64:
+    """
+    Converts a datetime object to a numpy datetime64 object with the specified
+    precision.
+
+    Args:
+        dt: A datetime object to be converted.
+        precision: The desired precision for the datetime64 conversion. Possible
+        values are 'D', 's', 'ms', 'us', 'ns'.
+
+    Returns:
+        np.datetime64: A numpy datetime64 object with the specified precision.
+    """
+    if precision == "ns":
+        # Calculate nanoseconds from microsecond and nanosecond
+        microseconds_as_ns = dt.microsecond * 1000
+        # Use getattr for backward compatibility where nanosecond attribute may not
+        # exist
+        nanoseconds = getattr(dt, "nanosecond", 0)
+        total_nanoseconds = microseconds_as_ns + nanoseconds
+        # Create datetime64 from base datetime with microsecond precision
+        base_dt = np.datetime64(dt, "us")
+        # Add remaining nanoseconds as timedelta
+        return base_dt + np.timedelta64(total_nanoseconds - microseconds_as_ns, "ns")
+    else:
+        return np.datetime64(dt).astype(f"datetime64[{precision}]")
 
 
 def _convert_datetime_list_to_array(datetime_list: List[datetime]) -> np.ndarray:
+    """Convert a list of datetime objects to a NumPy array of datetime64 with proper
+    precision.
+
+    Args:
+        datetime_list (List[datetime]): A list of `datetime` objects to be converted.
+        Each `datetime` object represents a specific point in time.
+
+    Returns:
+        np.ndarray: A NumPy array containing the `datetime64` values of the datetime
+        objects from the input list, with the appropriate precision (e.g., nanoseconds,
+        microseconds, milliseconds, etc.).
+    """
+    # Detect the highest precision for the datetime objects
     precision = _detect_highest_datetime_precision(datetime_list)
 
-    return np.array(
-        [np.datetime64(dt, precision) for dt in datetime_list],
-        dtype=f"datetime64[{precision}]",
-    )
+    # Convert each datetime to the corresponding numpy datetime64 with the appropriate
+    # precision
+    return np.array([_convert_to_datetime64(dt, precision) for dt in datetime_list])
 
 
 def convert_to_numpy(column_values: Any) -> np.ndarray:


Test patch:
diff --git a/python/ray/data/tests/test_arrow_block.py b/python/ray/data/tests/test_arrow_block.py
index ec6342c7e2b4c..80c916a22b36b 100644
--- a/python/ray/data/tests/test_arrow_block.py
+++ b/python/ray/data/tests/test_arrow_block.py
@@ -6,6 +6,7 @@
 from typing import Union
 
 import numpy as np
+import pandas as pd
 import pyarrow as pa
 import pytest
 from pyarrow import parquet as pq
@@ -355,6 +356,32 @@ def test_build_block_with_null_column(ray_start_regular_shared):
     assert np.array_equal(rows[1]["array"], np.zeros((2, 2)))
 
 
+def test_arrow_block_timestamp_ns(ray_start_regular_shared):
+    # Input data with nanosecond precision timestamps
+    data_rows = [
+        {"col1": 1, "col2": pd.Timestamp("2023-01-01T00:00:00.123456789")},
+        {"col1": 2, "col2": pd.Timestamp("2023-01-01T01:15:30.987654321")},
+        {"col1": 3, "col2": pd.Timestamp("2023-01-01T02:30:15.111111111")},
+        {"col1": 4, "col2": pd.Timestamp("2023-01-01T03:45:45.222222222")},
+        {"col1": 5, "col2": pd.Timestamp("2023-01-01T05:00:00.333333333")},
+    ]
+
+    # Initialize ArrowBlockBuilder
+    arrow_builder = ArrowBlockBuilder()
+    for row in data_rows:
+        arrow_builder.add(row)
+    arrow_block = arrow_builder.build()
+
+    assert arrow_block.schema.field("col2").type == pa.timestamp("ns")
+    for i, row in enumerate(data_rows):
+        result_timestamp = arrow_block["col2"][i].as_py()
+        # Convert both values to pandas Timestamp to preserve nanosecond precision for
+        # comparison.
+        assert pd.Timestamp(row["col2"]) == pd.Timestamp(
+            result_timestamp
+        ), f"Timestamp mismatch at row {i} in ArrowBlockBuilder output"
+
+
 def test_arrow_nan_element():
     ds = ray.data.from_items(
         [
diff --git a/python/ray/data/tests/test_map.py b/python/ray/data/tests/test_map.py
index 5df6ff3066d2f..8321864801f83 100644
--- a/python/ray/data/tests/test_map.py
+++ b/python/ray/data/tests/test_map.py
@@ -330,6 +330,144 @@ def map_generator(item: dict) -> Iterator[int]:
     ]
 
 
+# Helper function to process timestamp data in nanoseconds
+def process_timestamp_data(row):
+    # Convert numpy.datetime64 to pd.Timestamp if needed
+    if isinstance(row["timestamp"], np.datetime64):
+        row["timestamp"] = pd.Timestamp(row["timestamp"])
+
+    # Add 1ns to timestamp
+    row["timestamp"] = row["timestamp"] + pd.Timedelta(1, "ns")
+
+    # Ensure the timestamp column is in the expected dtype (datetime64[ns])
+    row["timestamp"] = pd.to_datetime(row["timestamp"], errors="raise")
+
+    return row
+
+
+def process_timestamp_data_batch_arrow(batch: pa.Table) -> pa.Table:
+    # Convert pyarrow Table to pandas DataFrame to process the timestamp column
+    df = batch.to_pandas()
+
+    df["timestamp"] = df["timestamp"].apply(
+        lambda x: pd.Timestamp(x) if isinstance(x, np.datetime64) else x
+    )
+
+    # Add 1ns to timestamp
+    df["timestamp"] = df["timestamp"] + pd.Timedelta(1, "ns")
+
+    # Convert back to pyarrow Table
+    return pa.table(df)
+
+
+def process_timestamp_data_batch_pandas(batch: pd.DataFrame) -> pd.DataFrame:
+    # Add 1ns to timestamp column
+    batch["timestamp"] = batch["timestamp"] + pd.Timedelta(1, "ns")
+    return batch
+
+
+@pytest.mark.parametrize(
+    "df, expected_df",
+    [
+        pytest.param(
+            pd.DataFrame(
+                {
+                    "id": [1, 2, 3],
+                    "timestamp": pd.to_datetime(
+                        [
+                            "2024-01-01 00:00:00.123456789",
+                            "2024-01-02 00:00:00.987654321",
+                            "2024-01-03 00:00:00.111222333",
+                        ]
+                    ),
+                    "value": [10.123456789, 20.987654321, 30.111222333],
+                }
+            ),
+            pd.DataFrame(
+                {
+                    "id": [1, 2, 3],
+                    "timestamp": pd.to_datetime(
+                        [
+                            "2024-01-01 00:00:00.123456790",
+                            "2024-01-02 00:00:00.987654322",
+                            "2024-01-03 00:00:00.111222334",
+                        ]
+                    ),
+                    "value": [10.123456789, 20.987654321, 30.111222333],
+                }
+            ),
+            id="nanoseconds_increment",
+        )
+    ],
+)
+def test_map_batches_timestamp_nanosecs(df, expected_df, ray_start_regular_shared):
+    """Verify handling timestamp with nanosecs in map_batches"""
+    ray_data = ray.data.from_pandas(df)
+
+    # Using pyarrow format
+    result_arrow = ray_data.map_batches(
+        process_timestamp_data_batch_arrow, batch_format="pyarrow"
+    )
+    processed_df_arrow = result_arrow.to_pandas()
+    processed_df_arrow["timestamp"] = processed_df_arrow["timestamp"].astype(
+        "datetime64[ns]"
+    )
+    pd.testing.assert_frame_equal(processed_df_arrow, expected_df)
+
+    # Using pandas format
+    result_pandas = ray_data.map_batches(
+        process_timestamp_data_batch_pandas, batch_format="pandas"
+    )
+    processed_df_pandas = result_pandas.to_pandas()
+    processed_df_pandas["timestamp"] = processed_df_pandas["timestamp"].astype(
+        "datetime64[ns]"
+    )
+    pd.testing.assert_frame_equal(processed_df_pandas, expected_df)
+
+
+@pytest.mark.parametrize(
+    "df, expected_df",
+    [
+        pytest.param(
+            pd.DataFrame(
+                {
+                    "id": [1, 2, 3],
+                    "timestamp": pd.to_datetime(
+                        [
+                            "2024-01-01 00:00:00.123456789",
+                            "2024-01-02 00:00:00.987654321",
+                            "2024-01-03 00:00:00.111222333",
+                        ]
+                    ),
+                    "value": [10.123456789, 20.987654321, 30.111222333],
+                }
+            ),
+            pd.DataFrame(
+                {
+                    "id": [1, 2, 3],
+                    "timestamp": pd.to_datetime(
+                        [
+                            "2024-01-01 00:00:00.123456790",
+                            "2024-01-02 00:00:00.987654322",
+                            "2024-01-03 00:00:00.111222334",
+                        ]
+                    ),
+                    "value": [10.123456789, 20.987654321, 30.111222333],
+                }
+            ),
+            id="nanoseconds_increment_map",
+        )
+    ],
+)
+def test_map_timestamp_nanosecs(df, expected_df, ray_start_regular_shared):
+    """Verify handling timestamp with nanosecs in map"""
+    ray_data = ray.data.from_pandas(df)
+    result = ray_data.map(process_timestamp_data)
+    processed_df = result.to_pandas()
+    processed_df["timestamp"] = processed_df["timestamp"].astype("datetime64[ns]")
+    pd.testing.assert_frame_equal(processed_df, expected_df)
+
+
 def test_add_column(ray_start_regular_shared):
     """Tests the add column API."""
 
diff --git a/python/ray/data/tests/test_numpy_support.py b/python/ray/data/tests/test_numpy_support.py
index ec67bcf689bb8..adca6e4ab5aa4 100644
--- a/python/ray/data/tests/test_numpy_support.py
+++ b/python/ray/data/tests/test_numpy_support.py
@@ -1,6 +1,7 @@
 from datetime import datetime
 
 import numpy as np
+import pandas as pd
 import pytest
 import torch
 
@@ -58,6 +59,7 @@ def test_list_of_objects(ray_start_regular_shared, restore_data_context):
     assert_structure_equals(output, np.array([1, 2, 3, UserObj()]))
 
 
+# Define datetime values with different precisions
 DATETIME_DAY_PRECISION = datetime(year=2024, month=1, day=1)
 DATETIME_HOUR_PRECISION = datetime(year=2024, month=1, day=1, hour=1)
 DATETIME_MIN_PRECISION = datetime(year=2024, month=1, day=1, minute=1)
@@ -65,14 +67,28 @@ def test_list_of_objects(ray_start_regular_shared, restore_data_context):
 DATETIME_MILLISEC_PRECISION = datetime(year=2024, month=1, day=1, microsecond=1000)
 DATETIME_MICROSEC_PRECISION = datetime(year=2024, month=1, day=1, microsecond=1)
 
+# Define pandas values for different precisions
+PANDAS_DAY_PRECISION = pd.Timestamp(year=2024, month=1, day=1)
+PANDAS_HOUR_PRECISION = pd.Timestamp(year=2024, month=1, day=1, hour=1)
+PANDAS_MIN_PRECISION = pd.Timestamp(year=2024, month=1, day=1, minute=1)
+PANDAS_SEC_PRECISION = pd.Timestamp(year=2024, month=1, day=1, second=1)
+PANDAS_MILLISEC_PRECISION = pd.Timestamp(year=2024, month=1, day=1, microsecond=1000)
+PANDAS_MICROSEC_PRECISION = pd.Timestamp(year=2024, month=1, day=1, microsecond=1)
+PANDAS_NANOSEC_PRECISION = pd.Timestamp(
+    year=2024, month=1, day=1, hour=1, minute=1, second=1
+) + pd.Timedelta(nanoseconds=1)
+
+# Define numpy.datetime64 values for comparison
 DATETIME64_DAY_PRECISION = np.datetime64("2024-01-01")
 DATETIME64_HOUR_PRECISION = np.datetime64("2024-01-01T01:00", "s")
 DATETIME64_MIN_PRECISION = np.datetime64("2024-01-01T00:01", "s")
 DATETIME64_SEC_PRECISION = np.datetime64("2024-01-01T00:00:01")
 DATETIME64_MILLISEC_PRECISION = np.datetime64("2024-01-01T00:00:00.001")
 DATETIME64_MICROSEC_PRECISION = np.datetime64("2024-01-01T00:00:00.000001")
+DATETIME64_NANOSEC_PRECISION = np.datetime64("2024-01-01T01:01:01.000000001", "ns")
 
 
+# Parametrized test to validate datetime values and expected numpy.datetime64 results
 @pytest.mark.parametrize(
     "data,expected_output",
     [
@@ -85,17 +101,59 @@ def test_list_of_objects(ray_start_regular_shared, restore_data_context):
         (
             [DATETIME_MICROSEC_PRECISION, DATETIME_MILLISEC_PRECISION],
             np.array(
-                [DATETIME64_MICROSEC_PRECISION, DATETIME_MILLISEC_PRECISION],
+                [DATETIME64_MICROSEC_PRECISION, DATETIME64_MILLISEC_PRECISION],
                 dtype="datetime64[us]",
             ),
         ),
         (
             [DATETIME_SEC_PRECISION, DATETIME_MILLISEC_PRECISION],
             np.array(
-                [DATETIME64_SEC_PRECISION, DATETIME_MILLISEC_PRECISION],
+                [DATETIME64_SEC_PRECISION, DATETIME64_MILLISEC_PRECISION],
                 dtype="datetime64[ms]",
             ),
         ),
+        (
+            [DATETIME_DAY_PRECISION, DATETIME_SEC_PRECISION],
+            np.array(
+                [DATETIME64_DAY_PRECISION, DATETIME64_SEC_PRECISION],
+                dtype="datetime64[s]",
+            ),
+        ),
+        ([PANDAS_DAY_PRECISION], np.array([DATETIME64_DAY_PRECISION])),
+        ([PANDAS_HOUR_PRECISION], np.array([DATETIME64_HOUR_PRECISION])),
+        ([PANDAS_MIN_PRECISION], np.array([DATETIME64_MIN_PRECISION])),
+        ([PANDAS_SEC_PRECISION], np.array([DATETIME64_SEC_PRECISION])),
+        ([PANDAS_MILLISEC_PRECISION], np.array([DATETIME64_MILLISEC_PRECISION])),
+        ([PANDAS_MICROSEC_PRECISION], np.array([DATETIME64_MICROSEC_PRECISION])),
+        ([PANDAS_NANOSEC_PRECISION], np.array([DATETIME64_NANOSEC_PRECISION])),
+        (
+            [PANDAS_NANOSEC_PRECISION, PANDAS_MICROSEC_PRECISION],
+            np.array(
+                [DATETIME64_NANOSEC_PRECISION, DATETIME64_MICROSEC_PRECISION],
+                dtype="datetime64[ns]",
+            ),
+        ),
+        (
+            [PANDAS_MICROSEC_PRECISION, PANDAS_MILLISEC_PRECISION],
+            np.array(
+                [DATETIME64_MICROSEC_PRECISION, DATETIME64_MILLISEC_PRECISION],
+                dtype="datetime64[us]",
+            ),
+        ),
+        (
+            [PANDAS_SEC_PRECISION, PANDAS_MILLISEC_PRECISION],
+            np.array(
+                [DATETIME64_SEC_PRECISION, DATETIME64_MILLISEC_PRECISION],
+                dtype="datetime64[ms]",
+            ),
+        ),
+        (
+            [PANDAS_DAY_PRECISION, PANDAS_SEC_PRECISION],
+            np.array(
+                [DATETIME64_DAY_PRECISION, DATETIME64_SEC_PRECISION],
+                dtype="datetime64[s]",
+            ),
+        ),
     ],
 )
 def test_list_of_datetimes(
diff --git a/python/ray/data/tests/test_pandas_block.py b/python/ray/data/tests/test_pandas_block.py
index f7e98f4bd1eb1..4e2e1a93c9d40 100644
--- a/python/ray/data/tests/test_pandas_block.py
+++ b/python/ray/data/tests/test_pandas_block.py
@@ -9,7 +9,7 @@
 
 import ray
 import ray.data
-from ray.data._internal.pandas_block import PandasBlockAccessor
+from ray.data._internal.pandas_block import PandasBlockAccessor, PandasBlockBuilder
 from ray.data.extensions.object_extension import _object_extension_type_allowed
 
 # Set seed for the test for size as it related to sampling
@@ -28,6 +28,32 @@ def test_append_column(ray_start_regular_shared):
     assert actual_block.equals(expected_block)
 
 
+def test_pandas_block_timestamp_ns(ray_start_regular_shared):
+    # Input data with nanosecond precision timestamps
+    data_rows = [
+        {"col1": 1, "col2": pd.Timestamp("2023-01-01T00:00:00.123456789")},
+        {"col1": 2, "col2": pd.Timestamp("2023-01-01T01:15:30.987654321")},
+        {"col1": 3, "col2": pd.Timestamp("2023-01-01T02:30:15.111111111")},
+        {"col1": 4, "col2": pd.Timestamp("2023-01-01T03:45:45.222222222")},
+        {"col1": 5, "col2": pd.Timestamp("2023-01-01T05:00:00.333333333")},
+    ]
+
+    # Initialize PandasBlockBuilder
+    pandas_builder = PandasBlockBuilder()
+    for row in data_rows:
+        pandas_builder.add(row)
+    pandas_block = pandas_builder.build()
+
+    assert pd.api.types.is_datetime64_ns_dtype(pandas_block["col2"])
+
+    for original_row, result_row in zip(
+        data_rows, pandas_block.to_dict(orient="records")
+    ):
+        assert (
+            original_row["col2"] == result_row["col2"]
+        ), "Timestamp mismatch in PandasBlockBuilder output"
+
+
 @pytest.mark.skipif(
     _object_extension_type_allowed(), reason="Objects can be put into Arrow"
 )