Encode non-negative nanos for negative ORC timestamps

cpp/include/cudf/io/orc.hpp

@@ -1070,6 +1070,9 @@ class orc_writer_options_builder {
  *
  * @note If an exception is thrown during encoding or compression, no data is written to the sink.
  *
+ * @note Timestamps in the last 999 milliseconds before the UNIX epoch are not representable in ORC;
+ * they are read back one second later, as with the Apache ORC writer (ORC-763, ORC-771).
+ *
  * @param options Settings for controlling reading behavior
  * @param stream CUDA stream used for device memory operations and kernel launches
  */
@@ -1487,6 +1490,9 @@ class chunked_orc_writer_options_builder {
  *    ...
  *  writer.close();
  * @endcode
+ *
+ * @note Timestamps in the last 999 milliseconds before the UNIX epoch are not representable in ORC;
+ * see `write_orc()` for details.
  */
 class orc_chunked_writer {
  public:

cpp/src/io/orc/stripe_enc.cu

@@ -799,15 +799,29 @@ CUDF_KERNEL void __launch_bounds__(block_size)
           case BOOLEAN:
           case BYTE: s->vals.u8[nz_idx] = column.element<uint8_t>(row); break;
           case TIMESTAMP: {
-            int64_t ts          = column.element<int64_t>(row);
-            int32_t ts_scale    = cudf::detail::powers_of_ten[9 - min(s->chunk.scale, 9)];
-            int64_t seconds     = ts / ts_scale;
-            int64_t nanos       = (ts - seconds * ts_scale);
+            auto const ts             = column.element<int64_t>(row);
+            auto const ticks_per_sec  = cudf::detail::powers_of_ten[9 - min(s->chunk.scale, 9)];
+            auto const nanos_per_tick = cudf::detail::powers_of_ten[min(s->chunk.scale, 9)];
+
+            auto seconds = ts / ticks_per_sec;
+            auto nanos   = (ts - seconds * ticks_per_sec) * nanos_per_tick;
+
+            // Adjust to keep the nanosecond remainder non-negative.
+            // See https://github.com/rapidsai/cudf/issues/19350.
+            if (nanos < 0) {
+              seconds -= 1;
+              nanos += 1'000'000'000;
+            }
+            // ORC readers borrow a second when the stored seconds are negative and the stored nanos
+            // are at least 1 ms (ORC-306/ORC-763, mirrored in the reader), so give that second back
+            // here. Timestamps in the last 999 ms before the epoch then store zero seconds and are
+            // read back a second later, as documented on `write_orc` (ORC-771).
+            if (seconds < 0 and nanos > 999'999) { seconds += 1; }
+
             s->vals.i64[nz_idx] = seconds - orc_utc_epoch;
             if (nanos != 0) {
               // Trailing zeroes are encoded in the lower 3-bits
               uint32_t zeroes = 0;
-              nanos *= cudf::detail::powers_of_ten[min(s->chunk.scale, 9)];
               if (!(nanos % 100)) {
                 nanos /= 100;
                 zeroes = 1;

cpp/tests/io/orc_test.cpp

@@ -653,6 +653,71 @@ TEST_F(OrcWriterTest, negTimestampsNano)
   CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view());
 }
 
+template <typename T>
+void test_timestamp_roundtrip(std::vector<typename T::rep> const& values,
+                              std::vector<typename T::rep> const& expected_values)
+{
+  cudf::test::fixed_width_column_wrapper<T, typename T::rep> const input(values.begin(),
+                                                                         values.end());
+  cudf::test::fixed_width_column_wrapper<T, typename T::rep> const expected(expected_values.begin(),
+                                                                            expected_values.end());
+  cudf::table_view const input_table({input});
+
+  std::vector<char> out_buffer;
+  cudf::io::orc_writer_options const out_opts =
+    cudf::io::orc_writer_options::builder(cudf::io::sink_info{&out_buffer}, input_table);
+  cudf::io::write_orc(out_opts);
+
+  cudf::io::orc_reader_options const in_opts =
+    cudf::io::orc_reader_options::builder(
+      cudf::io::source_info{cudf::host_span<std::byte const>{
+        reinterpret_cast<std::byte const*>(out_buffer.data()), out_buffer.size()}})
+      .use_index(false)
+      .timestamp_type(cudf::data_type{cudf::type_to_id<T>()});
+  auto const result = cudf::io::read_orc(in_opts);
+
+  CUDF_TEST_EXPECT_COLUMNS_EQUAL(
+    expected, result.tbl->view().column(0), cudf::test::debug_output_level::ALL_ERRORS);
+}
+
+TEST_F(OrcWriterTest, NegativeFractionalTimestamps)
+{
+  // ORC readers handle remainders of >= 1 ms above the lower second differently, so cover both
+  auto const timestamps_us = std::vector<cudf::timestamp_us::rep>{
+    -54'218'791'351'223'251L,  // 776.749 ms above the lower second, so >= 1 ms
+    -5'999'999L,               // 1 us above the lower second, so < 1 ms
+  };
+  test_timestamp_roundtrip<cudf::timestamp_us>(timestamps_us, timestamps_us);
+
+  auto const timestamps_ns = std::vector<cudf::timestamp_ns::rep>{
+    -131'968'727'238'000'000L,  // 762 ms above the lower second, so >= 1 ms
+    -5'999'999'999L,            // 1 ns above the lower second, so < 1 ms
+  };
+  test_timestamp_roundtrip<cudf::timestamp_ns>(timestamps_ns, timestamps_ns);
+
+  // Millisecond timestamps cannot have a < 1 ms remainder
+  auto const timestamps_ms = std::vector<cudf::timestamp_ms::rep>{
+    -123'456L,  // 544 ms above the lower second
+    -5'999L,    // 1 ms above the lower second, the smallest remainder at this resolution
+  };
+  test_timestamp_roundtrip<cudf::timestamp_ms>(timestamps_ms, timestamps_ms);
+}
+
+// Timestamps in the last 999 ms before the epoch are not representable in ORC; they are read back
+// one second later, as with the Apache ORC writer, whose own tests assert the same values (ORC-763,
+// ORC-771).
+TEST_F(OrcWriterTest, NegativeTimestampsNearEpoch)
+{
+  auto const timestamps_us = std::vector<cudf::timestamp_us::rep>{-1L, -500L, -500'000L, -999'000L};
+  auto const read_back_us =
+    std::vector<cudf::timestamp_us::rep>{999'999L, 999'500L, 500'000L, 1'000L};
+  test_timestamp_roundtrip<cudf::timestamp_us>(timestamps_us, read_back_us);
+
+  test_timestamp_roundtrip<cudf::timestamp_ns>({-1L, -999'000'000L}, {999'999'999L, 1'000'000L});
+
+  test_timestamp_roundtrip<cudf::timestamp_ms>({-1L, -999L}, {999L, 1L});
+}
+
 TEST_F(OrcWriterTest, Slice)
 {
   int32_col col{{1, 2, 3, 4, 5}, cudf::test::iterators::null_at(3)};

python/cudf/cudf/utils/ioutils.py

@@ -579,6 +579,12 @@
     doesn't require much space and is faster. Other indexes will
     be included as columns in the file output.
 
+Notes
+-----
+Timestamps in the last 999 milliseconds before the UNIX epoch are not
+representable in ORC; they are read back one second later, as with the Apache
+ORC writer (ORC-763, ORC-771).
+
 See Also
 --------
 cudf.read_orc