SNOW-1025489 enforce ms scale for timestamps (#2758)

Author: sfc-gh-pczajka

DESCRIPTION.md

@@ -14,6 +14,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
   - Log a warning when using http protocol for OAuth urls.
   - Deprecated support for custom revocation error classes in OCSP response cache deserialization. By default, only `RevocationCheckError` exceptions are deserialized from OCSP cache. Custom exception classes can be temporarily enabled by setting the `SNOWFLAKE_ENABLE_CUSTOM_REVOCATION_ERRORS` environment variable to `true` or `1`, but this support will be removed in a future release.
   - Bumped up vendored `urllib3` to `2.6.3`
+  - Added `force_microseconds_precision` to `cursor.fetch_arrow_all` and `cursor.fetch_pandas_all` to avoid PyArrow schema incosistency between batches.
 
 - v4.2.0(January 07,2026)
   - Added `SnowflakeCursor.stats` property to expose granular DML statistics (rows inserted, deleted, updated, and duplicates) for operations like CTAS where `rowcount` is insufficient.

src/snowflake/connector/cursor.py

@@ -1337,7 +1337,21 @@ def query_result(self, qid: str) -> SnowflakeCursor:
             )
         return self
 
-    def fetch_arrow_batches(self) -> Iterator[Table]:
+    def fetch_arrow_batches(
+        self,
+        force_microsecond_precision: bool = False,
+    ) -> Iterator[Table]:
+        """Fetch Arrow Tables in batches.
+
+        Args:
+            force_microsecond_precision: When True, all timestamp columns are converted
+                to microsecond precision, ensuring consistent schema across all batches.
+                This is useful when your data contains timestamps outside the nanosecond
+                range (1677-2262), such as '9999-12-31' or '0001-01-01'. When False
+                (default), precision is determined per-batch based on the data, which
+                may cause pyarrow schema mismatch errors when combining batches.
+                Note: enabling this truncates sub-microsecond precision (scale 7-9).
+        """
         self.check_can_use_arrow_resultset()
         if self._prefetch_hook is not None:
             self._prefetch_hook()
@@ -1346,20 +1360,43 @@ def fetch_arrow_batches(self) -> Iterator[Table]:
         self._log_telemetry_job_data(
             TelemetryField.ARROW_FETCH_BATCHES, TelemetryData.TRUE
         )
-        return self._result_set._fetch_arrow_batches()
+        return self._result_set._fetch_arrow_batches(
+            force_microsecond_precision=force_microsecond_precision
+        )
 
     @overload
-    def fetch_arrow_all(self, force_return_table: Literal[False]) -> Table | None: ...
+    def fetch_arrow_all(
+        self,
+        force_return_table: Literal[False] = ...,
+        force_microsecond_precision: bool = ...,
+    ) -> Table | None: ...
 
     @overload
-    def fetch_arrow_all(self, force_return_table: Literal[True]) -> Table: ...
+    def fetch_arrow_all(
+        self,
+        force_return_table: Literal[True],
+        force_microsecond_precision: bool = ...,
+    ) -> Table: ...
+
+    def fetch_arrow_all(
+        self,
+        force_return_table: bool = False,
+        force_microsecond_precision: bool = False,
+    ) -> Table | None:
+        """Fetch all results as a single Arrow Table.
 
-    def fetch_arrow_all(self, force_return_table: bool = False) -> Table | None:
-        """
         Args:
             force_return_table: Set to True so that when the query returns zero rows,
-                an empty pyarrow table will be returned with schema using the highest bit length for each column.
-                Default value is False in which case None is returned in case of zero rows.
+                an empty pyarrow table will be returned with schema using the highest
+                bit length for each column. Default value is False in which case None
+                is returned in case of zero rows.
+            force_microsecond_precision: When True, all timestamp columns are converted
+                to microsecond precision, ensuring consistent schema across all batches.
+                This is useful when your data contains timestamps outside the nanosecond
+                range (1677-2262), such as '9999-12-31' or '0001-01-01'. When False
+                (default), precision is determined per-batch based on the data, which
+                may cause pyarrow schema mismatch errors when combining batches.
+                Note: enabling this truncates sub-microsecond precision (scale 7-9).
         """
         self.check_can_use_arrow_resultset()
 
@@ -1368,7 +1405,10 @@ def fetch_arrow_all(self, force_return_table: bool = False) -> Table | None:
         if self._query_result_format != "arrow":
             raise NotSupportedError
         self._log_telemetry_job_data(TelemetryField.ARROW_FETCH_ALL, TelemetryData.TRUE)
-        return self._result_set._fetch_arrow_all(force_return_table=force_return_table)
+        return self._result_set._fetch_arrow_all(
+            force_return_table=force_return_table,
+            force_microsecond_precision=force_microsecond_precision,
+        )
 
     def fetch_pandas_batches(self, **kwargs: Any) -> Iterator[DataFrame]:
         """Fetches a single Arrow Table."""

src/snowflake/connector/nanoarrow_cpp/ArrowIterator/CArrowTableIterator.cpp

@@ -335,10 +335,12 @@ void CArrowTableIterator::reconstructRecordBatches_nanoarrow() {
 
 CArrowTableIterator::CArrowTableIterator(PyObject* context, char* arrow_bytes,
                                          int64_t arrow_bytes_size,
-                                         const bool number_to_decimal)
+                                         const bool number_to_decimal,
+                                         const bool force_microsecond_precision)
     : CArrowIterator(arrow_bytes, arrow_bytes_size),
       m_context(context),
-      m_convert_number_to_decimal(number_to_decimal) {
+      m_convert_number_to_decimal(number_to_decimal),
+      m_force_microsecond_precision(force_microsecond_precision) {
   if (py::checkPyError()) {
     return;
   }
@@ -745,8 +747,11 @@ void CArrowTableIterator::convertTimestampColumn_nanoarrow(
   // Find epoch and fraction arrays for overflow detection
   ArrowArrayView* epochArray = nullptr;
   ArrowArrayView* fractionArray = nullptr;
-  bool has_overflow_to_downscale = false;
-  if (scale > 6 && field->type == NANOARROW_TYPE_STRUCT) {
+  // When m_force_microsecond_precision is true, always use microsecond
+  // precision to ensure consistent schema across all batches
+  bool has_overflow_to_downscale = m_force_microsecond_precision;
+  if (!m_force_microsecond_precision && scale > 6 &&
+      field->type == NANOARROW_TYPE_STRUCT) {
     for (int64_t i = 0; i < field->schema->n_children; i++) {
       ArrowSchema* c_schema = field->schema->children[i];
       if (std::strcmp(c_schema->name, internal::FIELD_NAME_EPOCH.c_str()) ==
@@ -977,9 +982,10 @@ void CArrowTableIterator::convertTimestampTZColumn_nanoarrow(
     }
   }
 
-  // Check for timestamp overflow and determine if downscaling is needed
-  bool has_overflow_to_downscale = false;
-  if (scale > 6 && byteLength == 16) {
+  // When m_force_microsecond_precision is true, always use microsecond
+  // precision to ensure consistent schema across all batches
+  bool has_overflow_to_downscale = m_force_microsecond_precision;
+  if (!m_force_microsecond_precision && scale > 6 && byteLength == 16) {
     has_overflow_to_downscale = _checkNanosecondTimestampOverflowAndDownscale(
         columnArray, epochArray, fractionArray);
   }

src/snowflake/connector/nanoarrow_cpp/ArrowIterator/CArrowTableIterator.hpp

@@ -23,7 +23,8 @@ class CArrowTableIterator : public CArrowIterator {
    * Constructor
    */
   CArrowTableIterator(PyObject* context, char* arrow_bytes,
-                      int64_t arrow_bytes_size, bool number_to_decimal);
+                      int64_t arrow_bytes_size, bool number_to_decimal,
+                      bool force_microsecond_precision = false);
 
   /**
    * Destructor
@@ -49,6 +50,8 @@ class CArrowTableIterator : public CArrowIterator {
   /** local time zone */
   char* m_timezone;
   const bool m_convert_number_to_decimal;
+  /** force microsecond precision for timestamps to ensure consistent schema */
+  const bool m_force_microsecond_precision;
 
   /**
    * Reconstruct record batches with type conversion in place

src/snowflake/connector/nanoarrow_cpp/ArrowIterator/nanoarrow_arrow_iterator.pyx

@@ -64,6 +64,7 @@ cdef extern from "CArrowTableIterator.hpp" namespace "sf":
             char* arrow_bytes,
             int64_t arrow_bytes_size,
             bint number_to_decimal,
+            bint force_microsecond_precision,
         ) except +
 
 
@@ -100,6 +101,7 @@ cdef class PyArrowIterator(EmptyPyArrowIterator):
     cdef object check_error_on_every_column
     cdef object number_to_decimal
     cdef object pyarrow_table
+    cdef bint force_microsecond_precision
 
     def __cinit__(
             self,
@@ -109,7 +111,8 @@ cdef class PyArrowIterator(EmptyPyArrowIterator):
             object use_dict_result,
             object numpy,
             object number_to_decimal,
-            object check_error_on_every_column
+            object check_error_on_every_column,
+            object force_microsecond_precision=False,
     ):
         self.context = arrow_context
         self.cIterator = NULL
@@ -122,6 +125,7 @@ cdef class PyArrowIterator(EmptyPyArrowIterator):
         self.table_returned = False
         self.arrow_bytes = <char*>arrow_bytes
         self.arrow_bytes_size = len(arrow_bytes)
+        self.force_microsecond_precision = force_microsecond_precision
 
     def __dealloc__(self):
         del self.cIterator
@@ -202,9 +206,10 @@ cdef class PyArrowTableIterator(PyArrowIterator):
         object use_dict_result,
         object numpy,
         object number_to_decimal,
-        object check_error_on_every_column
+        object check_error_on_every_column,
+        object force_microsecond_precision=False,
     ):
-        super().__init__(cursor, py_inputstream, arrow_context, use_dict_result, numpy, number_to_decimal, check_error_on_every_column)
+        super().__init__(cursor, py_inputstream, arrow_context, use_dict_result, numpy, number_to_decimal, check_error_on_every_column, force_microsecond_precision)
         if not INSTALLED_PYARROW:
             raise Error.errorhandler_make_exception(
                 ProgrammingError,
@@ -225,6 +230,7 @@ cdef class PyArrowTableIterator(PyArrowIterator):
             self.arrow_bytes,
             self.arrow_bytes_size,
             self.number_to_decimal,
+            self.force_microsecond_precision,
         )
         cdef ReturnVal cret = self.cIterator.checkInitializationStatus()
         if cret.exception:

src/snowflake/connector/result_batch.py

@@ -61,6 +61,7 @@ def _create_nanoarrow_iterator(
     number_to_decimal: bool,
     row_unit: IterUnit,
     check_error_on_every_column: bool = True,
+    force_microsecond_precision: bool = False,
 ):
     from .nanoarrow_arrow_iterator import PyArrowRowIterator, PyArrowTableIterator
 
@@ -84,6 +85,7 @@ def _create_nanoarrow_iterator(
             numpy,
             number_to_decimal,
             check_error_on_every_column,
+            force_microsecond_precision,
         )
     )
 
@@ -669,7 +671,10 @@ def __repr__(self) -> str:
         return f"ArrowResultChunk({self.id})"
 
     def _load(
-        self, response: Response, row_unit: IterUnit
+        self,
+        response: Response,
+        row_unit: IterUnit,
+        force_microsecond_precision: bool = False,
     ) -> Iterator[dict | Exception] | Iterator[tuple | Exception]:
         """Creates a ``PyArrowIterator`` from a response.
 
@@ -683,13 +688,15 @@ def _load(
             self._numpy,
             self._number_to_decimal,
             row_unit,
+            force_microsecond_precision=force_microsecond_precision,
         )
 
     def _from_data(
         self,
         data: str | bytes,
         iter_unit: IterUnit,
         check_error_on_every_column: bool = True,
+        force_microsecond_precision: bool = False,
     ) -> Iterator[dict | Exception] | Iterator[tuple | Exception]:
         """Creates a ``PyArrowIterator`` files from a str.
 
@@ -710,6 +717,7 @@ def _from_data(
             self._number_to_decimal,
             iter_unit,
             check_error_on_every_column,
+            force_microsecond_precision=force_microsecond_precision,
         )
 
     @classmethod
@@ -741,7 +749,10 @@ def from_data(
         return new_chunk
 
     def _create_iter(
-        self, iter_unit: IterUnit, connection: SnowflakeConnection | None = None
+        self,
+        iter_unit: IterUnit,
+        connection: SnowflakeConnection | None = None,
+        force_microsecond_precision: bool = False,
     ) -> Iterator[dict | Exception] | Iterator[tuple | Exception] | Iterator[Table]:
         """Create an iterator for the ResultBatch. Used by get_arrow_iter."""
         if self._local:
@@ -754,6 +765,7 @@ def _create_iter(
                         if connection
                         else None
                     ),
+                    force_microsecond_precision=force_microsecond_precision,
                 )
             except Exception:
                 if connection and getattr(connection, "_debug_arrow_chunk", False):
@@ -763,7 +775,11 @@ def _create_iter(
         logger.debug(f"started loading result batch id: {self.id}")
         with TimerContextManager() as load_metric:
             try:
-                loaded_data = self._load(response, iter_unit)
+                loaded_data = self._load(
+                    response,
+                    iter_unit,
+                    force_microsecond_precision=force_microsecond_precision,
+                )
             except Exception:
                 if connection and getattr(connection, "_debug_arrow_chunk", False):
                     logger.debug(f"arrow data can not be parsed: {response}")
@@ -773,10 +789,16 @@ def _create_iter(
         return loaded_data
 
     def _get_arrow_iter(
-        self, connection: SnowflakeConnection | None = None
+        self,
+        connection: SnowflakeConnection | None = None,
+        force_microsecond_precision: bool = False,
     ) -> Iterator[Table]:
         """Returns an iterator for this batch which yields a pyarrow Table"""
-        return self._create_iter(iter_unit=IterUnit.TABLE_UNIT, connection=connection)
+        return self._create_iter(
+            iter_unit=IterUnit.TABLE_UNIT,
+            connection=connection,
+            force_microsecond_precision=force_microsecond_precision,
+        )
 
     def _create_empty_table(self) -> Table:
         """Returns empty Arrow table based on schema"""
@@ -789,27 +811,50 @@ def _create_empty_table(self) -> Table:
         ]
         return pa.schema(fields).empty_table()
 
-    def to_arrow(self, connection: SnowflakeConnection | None = None) -> Table:
+    def to_arrow(
+        self,
+        connection: SnowflakeConnection | None = None,
+        force_microsecond_precision: bool = False,
+    ) -> Table:
         """Returns this batch as a pyarrow Table"""
-        val = next(self._get_arrow_iter(connection=connection), None)
+        val = next(
+            self._get_arrow_iter(
+                connection=connection,
+                force_microsecond_precision=force_microsecond_precision,
+            ),
+            None,
+        )
         if val is not None:
             return val
         return self._create_empty_table()
 
     def to_pandas(
-        self, connection: SnowflakeConnection | None = None, **kwargs
+        self,
+        connection: SnowflakeConnection | None = None,
+        force_microsecond_precision: bool = False,
+        **kwargs,
     ) -> DataFrame:
         """Returns this batch as a pandas DataFrame"""
         self._check_can_use_pandas()
-        table = self.to_arrow(connection=connection)
+        table = self.to_arrow(
+            connection=connection,
+            force_microsecond_precision=force_microsecond_precision,
+        )
         return table.to_pandas(**kwargs)
 
     def _get_pandas_iter(
-        self, connection: SnowflakeConnection | None = None, **kwargs
+        self,
+        connection: SnowflakeConnection | None = None,
+        force_microsecond_precision: bool = False,
+        **kwargs,
     ) -> Iterator[DataFrame]:
         """An iterator for this batch which yields a pandas DataFrame"""
         iterator_data = []
-        dataframe = self.to_pandas(connection=connection, **kwargs)
+        dataframe = self.to_pandas(
+            connection=connection,
+            force_microsecond_precision=force_microsecond_precision,
+            **kwargs,
+        )
         if not dataframe.empty:
             iterator_data.append(dataframe)
         return iter(iterator_data)
@@ -824,12 +869,22 @@ def create_iter(
     ):
         """The interface used by ResultSet to create an iterator for this ResultBatch."""
         iter_unit: IterUnit = kwargs.pop("iter_unit", IterUnit.ROW_UNIT)
+        force_microsecond_precision: bool = kwargs.pop(
+            "force_microsecond_precision", False
+        )
         if iter_unit == IterUnit.TABLE_UNIT:
             structure = kwargs.pop("structure", "pandas")
             if structure == "pandas":
-                return self._get_pandas_iter(connection=connection, **kwargs)
+                return self._get_pandas_iter(
+                    connection=connection,
+                    force_microsecond_precision=force_microsecond_precision,
+                    **kwargs,
+                )
             else:
-                return self._get_arrow_iter(connection=connection)
+                return self._get_arrow_iter(
+                    connection=connection,
+                    force_microsecond_precision=force_microsecond_precision,
+                )
         else:
             return self._create_iter(iter_unit=iter_unit, connection=connection)