@@ -15,9 +15,11 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import calendar
 import re
 from abc import ABC, ABCMeta, abstractmethod
 from dataclasses import dataclass
+from datetime import datetime
 from decimal import Decimal
 from typing import Any, ClassVar, Dict, List, Optional, Sequence, cast
 
@@ -47,6 +49,19 @@ def _combine_arrow_expressions(expressions, combine):
     return expressions[0]
 
 
+def _timestamp_literal_in_field_units(value, unit):
+    """Express a datetime exactly in column units without a bounded ns cast."""
+    if value is None:
+        return None
+    micros = calendar.timegm(value.timetuple()) * 1000000 + value.microsecond
+    offset = value.utcoffset()
+    if offset is not None:
+        micros -= (offset.days * 86400 + offset.seconds) * 1000000 + offset.microseconds
+    nanos = micros * 1000 + getattr(value, 'nanosecond', 0)
+    scale = {'s': 9, 'ms': 6, 'us': 3, 'ns': 0}[unit]
+    return Decimal((int(nanos < 0), tuple(int(digit) for digit in str(abs(nanos))), -scale))
+
+
 @dataclass
 class Predicate:
     method: str
@@ -204,6 +219,19 @@ def to_arrow(self, schema: Optional[pyarrow.Schema] = None) -> Any:
         literals = self.literals
         if schema is not None and self.field in schema.names:
             field_type = schema.field(self.field).type
+            if (pyarrow.types.is_timestamp(field_type) and literals
+                    and any(getattr(value, 'nanosecond', 0) for value in literals)
+                    and all(value is None or (isinstance(value, datetime)
+                                              and (value.utcoffset() is None) == (field_type.tz is None))
+                            for value in literals)):
+                # Arrow otherwise infers datetime subclasses at microsecond
+                # resolution. Compare exact column units without promoting
+                # wide ms/us columns or companion literals into bounded ns.
+                # Reuse DECIMAL's floor/ceiling and SQL membership semantics.
+                units_type = pyarrow.decimal128(19, 0)
+                return decimal_arrow_expression(
+                    field.cast(pyarrow.int64()).cast(units_type), units_type, self.method,
+                    [_timestamp_literal_in_field_units(value, field_type.unit) for value in literals])
             if (
                 pyarrow.types.is_decimal(field_type)
                 and literals
 
@@ -75,15 +75,55 @@ def _native_blob_view_supported(table, read_names) -> bool:
     return loader is None or _catalog_metastore(loader) == 'rest'
 
 
-def native_split_from_python(split):
-    """Convert a Python-planned DataSplit/IndexedSplit for the Rust reader.
+def _native_read_split(split, ranges_use_row_ids):
+    """Express Python's file slices with the existing indexed-split selector."""
+    from pypaimon.globalindex.indexed_split import IndexedSplit
+    from pypaimon.read.query_auth_split import QueryAuthSplit
+    from pypaimon.read.sliced_split import SlicedSplit
+    from pypaimon.utils.range import Range
+
+    def select_rows(base, ranges):
+        if isinstance(base, QueryAuthSplit):
+            return QueryAuthSplit(select_rows(base.split, ranges), base.auth_result)
+        if isinstance(base, IndexedSplit):
+            return select_rows(base.data_split(), Range.and_(base.row_ranges(), ranges))
+        return IndexedSplit(base, ranges)
+
+    if isinstance(split, QueryAuthSplit):
+        base = _native_read_split(split.split, ranges_use_row_ids)
+        return split if base is split.split else QueryAuthSplit(base, split.auth_result)
+    if isinstance(split, IndexedSplit):
+        base = _native_read_split(split.data_split(), ranges_use_row_ids)
+        return split if base is split.data_split() else select_rows(base, split.row_ranges())
+    if not isinstance(split, SlicedSplit):
+        return split
+    if ranges_use_row_ids is None:
+        raise ValueError('Native file-slice conversion requires the reader row-tracking mode')
+    ranges = []
+    offset = 0
+    slices = split.shard_file_idx_map()
+    for file in split.files:
+        start, end = slices.get(file.file_name, (0, file.row_count))
+        start, end = max(0, start), min(file.row_count, end)
+        if start < end:
+            base = file.first_row_id if ranges_use_row_ids else offset
+            if base is None:
+                raise ValueError("Row-tracked file '%s' is missing first_row_id" % file.file_name)
+            ranges.append(Range(base + start, base + end - 1))
+        offset += file.row_count
+    return select_rows(_native_read_split(split.data_split(), ranges_use_row_ids), ranges)
+
+
+def native_split_from_python(split, *, ranges_use_row_ids=None):
+    """Convert a Python-planned split for the Rust reader.
 
     Vector scores intentionally stay on the Python IndexedSplit. Rust needs
-    only its row ranges to perform the physical read.
+    only its row ranges to perform the physical read. File slices need the
+    reader's tracking mode to map their local offsets to those ranges.
     """
     from pypaimon_rust.datafusion import Split as NativeSplit
-    return NativeSplit.deserialize(
-        serialize_split_v1(split, include_scores=False))
+    return NativeSplit.deserialize(serialize_split_v1(
+        _native_read_split(split, ranges_use_row_ids), include_scores=False))
 
 
 def _partition_fields(table):
 
@@ -454,18 +454,7 @@ def _assign_row_tracking(self, record_batch: RecordBatch) -> RecordBatch:
             # NULL entries from the manifest's max sequence number.
             arrays[idx] = pc.fill_null(arrays[idx], self.max_sequence_number)
 
-        names = record_batch.schema.names
-        table = None
-        for i, name in enumerate(names):
-            field = pa.field(
-                name, arrays[i].type,
-                nullable=record_batch.schema.field(name).nullable
-            )
-            if table is None:
-                table = pa.table({name: arrays[i]}, schema=pa.schema([field]))
-            else:
-                table = table.append_column(field, arrays[i])
-        return table.to_batches()[0]
+        return pa.RecordBatch.from_arrays(arrays, schema=record_batch.schema)
 
     def close(self) -> None:
         self.format_reader.close()
@@ -833,8 +833,8 @@ def _init_bucket_selector(self):
         # Only the default hash function (Math.abs(hash % numBuckets)) is
         # supported for bucket pruning. Non-default functions (mod, hive)
         # use different algorithms and would produce wrong bucket sets.
-        bucket_func = self.table.table_schema.options.get('bucket-function.type', 'default')
-        if bucket_func.lower() != 'default':
+        bucket_func = self.table.table_schema.options.get('bucket-function.type')
+        if bucket_func is not None and bucket_func.lower() != 'default':
             return None
         try:
             bucket_key_fields = self.table.table_schema.logical_bucket_key_fields
 
@@ -1670,6 +1670,7 @@ def _create_union_reader(self, need_merge_files: List[DataFileMeta], deletion_ve
         row_offsets = [-1] * len(all_read_fields)
         field_offsets = [-1] * len(all_read_fields)
         schema_pos = {f.id: p for p, f in enumerate(self.table.fields)}
+        partition_ids = [field.id for field in self.table.fields if field.name in self.table.partition_keys]
 
         for i, bunch in enumerate(fields_files):
             first_file = bunch.files()[0]
@@ -1693,6 +1694,12 @@ def _create_union_reader(self, need_merge_files: List[DataFileMeta], deletion_ve
                 field_ids.append(SpecialFields.ROW_ID.id)
                 field_ids.append(SpecialFields.SEQUENCE_NUMBER.id)
 
+            if isinstance(bunch, DataBunch):
+                # Partition values belong to the split, even when write_cols
+                # omits them from every file. Let the normal reader's existing
+                # PartitionInfo provide them before the merged row is filtered.
+                field_ids.extend(field_id for field_id in partition_ids if field_id not in field_ids)
+
             read_fields = []
             for j, read_field_id in enumerate(read_field_index):
                 for field_id in field_ids:
 
@@ -32,7 +32,6 @@
 from pypaimon.common.options.core_options import ChangelogProducer
 from pypaimon.common.predicate import Predicate
 from pypaimon.consumer.consumer import Consumer
-from pypaimon.consumer.consumer_manager import ConsumerManager
 from pypaimon.manifest.manifest_file_manager import ManifestFileManager
 from pypaimon.manifest.manifest_list_manager import ManifestListManager
 from pypaimon.read.native_plan import _raise_if_native_fork_safety_error
@@ -115,7 +114,7 @@ def __init__(
         self._query_auth_fn = self.table.catalog_environment.table_query_auth(
             self.table.options, self.table.identifier)
         self._consumer_manager = (
-            ConsumerManager(table.file_io, table.table_path)
+            table.consumer_manager()
             if consumer_id else None
         )
 
 
@@ -625,7 +625,8 @@ def _try_native_batches(
             rust_split = getattr(split, '_native_split', None)
             if rust_split is None:
                 try:
-                    rust_split = native_split_from_python(split)
+                    rust_split = native_split_from_python(split, ranges_use_row_ids=(
+                        self.table.options.row_tracking_enabled() or self.table.options.data_evolution_enabled()))
                 except Exception as e:
                     return self._native_fallback(
                         "native split conversion failed: %s" % e,
 
@@ -109,6 +109,10 @@ def merge_into(
     # unrestricted whole-table grant before any user callback or staging.
     from pypaimon.write.table_update import _authorize_update
     _authorize_update(target_table)
+    if any(clause.condition is not None
+           for clauses in (when_matched, when_not_matched) for clause in clauses):
+        from pypaimon.ray.merge_condition import _load_datafusion
+        _load_datafusion()
     target_on, source_on = _normalize_on(on)
     self_merge = _is_self_merge(target_table, source, target_on, source_on)
     from pypaimon.snapshot.time_travel_util import TimeTravelUtil
@@ -222,12 +226,7 @@ def _prepare(
         for c in list(when_matched) + list(when_not_matched)
     )
     if has_condition:
-        from pypaimon.ray.merge_condition import (
-            _load_datafusion,
-            extract_target_columns,
-        )
-
-        _load_datafusion()
+        from pypaimon.ray.merge_condition import extract_target_columns
         for c in when_not_matched:
             if c.condition is not None:
                 t_refs = extract_target_columns(c.condition)
 
@@ -76,15 +76,13 @@ def _parse_type_precision_scale(data_type):
 def _datetime_to_millis_and_nanos(value: datetime):
     """Convert datetime to (epoch_millis, nano_of_millisecond) without float arithmetic.
 
-    Python's datetime is microsecond-resolution, so nano_of_millisecond is
-    always a multiple of 1000 and sub-microsecond precision is lost. For
-    TIMESTAMP(7..9) columns the lower three nano digits will always serialise
-    as zero — round-trip through pypaimon is microsecond-faithful, not
-    nanosecond-faithful.
+    Plain datetime values have microsecond resolution. Arrow returns pandas
+    Timestamp values for nanosecond arrays, carrying the remaining three
+    digits separately in nanosecond.
     """
     epoch_seconds = calendar.timegm(value.timetuple())
     millis = epoch_seconds * 1000 + value.microsecond // 1000
-    nano_of_millisecond = (value.microsecond % 1000) * 1000
+    nano_of_millisecond = (value.microsecond % 1000) * 1000 + getattr(value, 'nanosecond', 0)
     return millis, nano_of_millisecond
 
 
@@ -100,14 +98,21 @@ def _is_ltz_type(type_name: str) -> bool:
 
 
 def _millis_nanos_to_datetime(millis: int, nano_of_millisecond: int = 0) -> datetime:
-    """Convert (epoch_millis, nano_of_millisecond) to datetime. Nanos truncated to micros."""
+    """Retain nanosecond tails while preserving ordinary datetime results."""
     total_micros = millis * 1000 + nano_of_millisecond // 1000
     seconds = total_micros // 1_000_000
     micros = total_micros % 1_000_000
     if micros < 0:
         seconds -= 1
         micros += 1_000_000
-    return _EPOCH + timedelta(seconds=seconds, microseconds=micros)
+    result = _EPOCH + timedelta(seconds=seconds, microseconds=micros)
+    tail = nano_of_millisecond % 1000
+    if tail:
+        import pandas as pd
+        # Construct the final instant directly: the floored microsecond before
+        # Timestamp.min can be out of range even when these nanoseconds fit.
+        return pd.Timestamp(millis * 1000000 + nano_of_millisecond, unit='ns')
+    return result
 
 
 @dataclass
 
@@ -17,6 +17,7 @@
 
 """Index update policy uses the identities in a prepared file's own schema."""
 
+from dataclasses import replace
 from unittest.mock import patch
 
 import pyarrow as pa
@@ -108,3 +109,50 @@ def test_commit_resolves_updated_index_columns_by_file_schema(native_rest_catalo
         read = latest.new_read_builder()
         rows = read.new_read().to_arrow(read.new_scan().plan().splits()).to_pydict()
         assert rows == {'id': [1], index_field.name: [None if evolution == 'drop_readd' else 11]}
+
+
+@pytest.mark.parametrize('native', _ENGINES)
+@pytest.mark.parametrize('action', ['THROW_ERROR', 'DROP_PARTITION_INDEX', 'IGNORE'])
+@pytest.mark.parametrize('deleted', ['affected', 'other', 'both'])
+def test_explicit_index_deletion_does_not_exempt_other_indexes(native_rest_catalog, native, action, deleted):
+    catalog = native_rest_catalog
+    schema = pa.schema([('id', pa.int32()), ('value', pa.int32())])
+    catalog.create_table('default.index_deletes', Schema.from_pyarrow_schema(schema, options={
+        'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+        'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+        'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
+        'global-index.column-update-action': action,
+    }), False)
+    table = catalog.get_table('default.index_deletes')
+    writer = table.new_batch_write_builder().new_write()
+    try:
+        writer.write_arrow(pa.table({'id': [1], 'value': [10]}, schema=schema))
+        _commit(table, writer.prepare_commit())
+    finally:
+        writer.close()
+    affected = _index_entry('affected.index', (), table.fields[1].id)
+    other = _index_entry('other.index', (), table.fields[0].id)
+    _commit(table, [CommitMessage((), 0, [], index_adds=[affected, other])])
+    before = table.snapshot_manager().get_latest_snapshot()
+    source = table.copy({'global-index.column-update-action': 'IGNORE'})
+    messages = source.new_batch_write_builder().new_update().new_update_by_row_id().update_columns(
+        pa.table({'_ROW_ID': [0], 'value': [11]}), ['value'])
+    deletions = [affected] if deleted == 'affected' else [other] if deleted == 'other' else [affected, other]
+    messages.append(CommitMessage((), 0, [], index_deletes=[replace(entry, kind=1) for entry in deletions]))
+    target = table.copy({'commit.native.enabled': str(native).lower()})
+    if action == 'THROW_ERROR' and deleted == 'other':
+        with pytest.raises(Exception, match='globally indexed columns'):
+            _commit(target, messages, native)
+        assert table.snapshot_manager().get_latest_snapshot().id == before.id
+        expected_indexes, expected_value = ['affected.index', 'other.index'], 10
+    else:
+        _commit(target, messages, native)
+        expected_indexes = ['other.index'] if deleted == 'affected' else (
+            ['affected.index'] if deleted == 'other' and action == 'IGNORE' else [])
+        expected_value = 11
+    snapshot = table.snapshot_manager().get_latest_snapshot()
+    actual_indexes = sorted(entry.index_file.file_name for entry in scan_global_index_entries(table, snapshot))
+    assert actual_indexes == expected_indexes
+    read = table.new_read_builder()
+    assert read.new_read().to_arrow(read.new_scan().plan().splits()).to_pydict() == {
+        'id': [1], 'value': [expected_value]}