JVM insert_arrow, insert_dataframe, insert_columns: a timestamp or date column crosses as one long[] of epoch ticks, not one JPype call per cell (refs #315)

docs/modules/ROOT/pages/how-to/bulk-loads.adoc

@@ -78,6 +78,8 @@ A column the buffers cannot carry makes the batch take the row path, which store
 
 On the `arcadedb-jvm` package, a column of only `str` and `None` crosses as one UTF-8 buffer with its row offsets, and a mask when a row is null. A column that mixes types, is empty, or holds more than 2 GiB of text crosses value by value. A column with a lone surrogate also crosses value by value, and is refused with the error it always raised, before anything is written.
 
+On the same package, a timestamp or date column of `insert_arrow` and `insert_dataframe` crosses as one array of epoch values for each batch, and the bridge builds the date-times in Java. The stored value does not change: a date-time is a UTC wall clock, a nanosecond value reaches the property with its nanoseconds (the precision of the property, such as `DATETIME_NANOS`, decides what is stored), and a null stays a null. A list of Python `datetime` objects passed to `insert_columns` still crosses value by value.
+
 [[parallel-inserts]]
 == Parallel and async inserts
 

src/arcadedb/core.py

@@ -243,7 +243,8 @@ def _column_to_java(name, values):
     into a ``long[]``, ``double[]``, or ``boolean[]``; a string or object array
     and any other sequence of mixed values convert per element into an ``Object[]`` (a ``str``
     reuses one Java String per distinct value, ``None`` is a null); a column of nothing but ``str`` and ``None`` crosses instead as one
-    ``StringColumn`` (``_string_column``). A pandas
+    ``StringColumn`` (``_string_column``). A bridge ``TemporalColumn``, which ``insert_arrow`` and ``insert_dataframe`` build from a
+    timestamp or date column (``dataframe._temporal_column``), crosses as it is. A pandas
     ``Series`` is read through ``to_numpy``, with a nullable dtype's ``<NA>``
     as ``None``. A 2-D ``float32`` array is a column of vectors: it crosses as
     one ``float[]`` and each document gets one row of it. Kinds that do not cross natively raise ``TypeError`` rather
@@ -315,6 +316,8 @@ def _column_to_java(name, values):
         # a string, bytes, or object array: its elements are Python objects (str, None, ...), converted one by one below
         values = list(values.astype(object, copy=False))
     elif not isinstance(values, (list, tuple)):
+        if isinstance(values, _java_class("com.arcadedb.python.TemporalColumn")):
+            return values, int(values.rows())
         values = list(values)
     column = _string_column(values)
     if column is not None:

src/arcadedb/dataframe.py

@@ -245,43 +245,46 @@ def _check_declared(
             )
 
 
-def _temporal_to_java(chunk, pa):
-    """A timestamp or date chunk as a list of ``LocalDateTime``/``LocalDate``, keeping the unit's precision.
+def _temporal_column(chunk, pa):
+    """A timestamp or date chunk as one ``TemporalColumn`` of the bridge, keeping the unit's precision.
+
+    The integer epoch values cross as ONE ``long[]`` (a buffer copy), the null rows as one ``boolean[]`` when there is a null, and the
+    bridge builds each ``LocalDateTime`` or ``LocalDate`` as it fills the document of that row: one call per column and batch, where a
+    ``LocalDateTime.ofEpochSecond`` or ``LocalDate.ofEpochDay`` call per cell (and one conversion per element in ``insert_columns``) came
+    before. ``insert_columns`` takes the column as it is.
 
     DATETIME is a UTC wall clock: a timestamp with a zone is already UTC epoch time, and a timestamp without
     one is taken as the wall clock it shows (the same rule ``Document.set`` applies to a ``datetime``), so both
-    are read from the integer epoch value. Going through ``datetime`` would cut a nanosecond value to microseconds.
+    are read from the integer epoch value, in the unit of the column. Going through ``datetime`` would cut a nanosecond value to
+    microseconds. A date32 is days, a date64 is milliseconds (the bridge divides by 86,400,000, rounding toward negative infinity).
     """
     import numpy as np
 
     types = pa.types
-    mask = chunk.is_null().to_numpy(zero_copy_only=False)
     if types.is_date(chunk.type):
-        local_date = jpype.JClass("java.time.LocalDate")
-        raw = chunk.cast(pa.int64() if types.is_date64(chunk.type) else pa.int32())
-        days = raw.fill_null(0).to_numpy(zero_copy_only=False).astype(np.int64)
-        if types.is_date64(chunk.type):
-            days = days // 86_400_000
-        return [
-            None if m else local_date.ofEpochDay(int(d)) for d, m in zip(days, mask)
-        ]
-    per_second = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000}[
-        chunk.type.unit
-    ]
-    values = chunk.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False)
-    seconds, rest = np.divmod(values.astype(np.int64), per_second)
-    nanos = rest * (1_000_000_000 // per_second)
-    local_datetime = jpype.JClass("java.time.LocalDateTime")
-    utc = jpype.JClass("java.time.ZoneOffset").UTC
-    return [
-        None if m else local_datetime.ofEpochSecond(int(s), int(n), utc)
-        for s, n, m in zip(seconds, nanos, mask)
-    ]
+        date64 = types.is_date64(chunk.type)
+        raw = chunk.cast(pa.int64() if date64 else pa.int32())
+        ticks, per_unit, factory = raw, 86_400_000 if date64 else 1, "dates"
+    else:
+        per_second = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000}
+        ticks = chunk.cast(pa.int64())
+        per_unit, factory = per_second[chunk.type.unit], "dateTimes"
+    values = ticks.fill_null(0).to_numpy(zero_copy_only=False)
+    nulls = None
+    if chunk.null_count:
+        nulls = jpype.JArray(jpype.JBoolean)(
+            chunk.is_null().to_numpy(zero_copy_only=False)
+        )
+    return getattr(jpype.JClass("com.arcadedb.python.TemporalColumn"), factory)(
+        jpype.JArray(jpype.JLong)(np.ascontiguousarray(values, dtype=np.int64)),
+        per_unit,
+        nulls,
+    )
 
 
 def _temporal_to_tagged(chunk, pa):
     """A timestamp or date chunk as tagged values (``T`` date-time, ``D`` date) for the native backend, with the same
-    reading rules as :func:`_temporal_to_java`: ISO text of the integer epoch value, so a nanosecond value keeps its
+    reading rules as :func:`_temporal_column`: ISO text of the integer epoch value, so a nanosecond value keeps its
     nanoseconds (a Python ``datetime`` would cut them to microseconds).
     """
     import numpy as np
@@ -381,7 +384,8 @@ def _column_values(
     float_vectors: bool = False,
 ):
     """One Arrow array as what ``Database.insert_columns`` takes: a numpy array when the column crosses as one
-    buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null).
+    buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null). On the JVM backend a timestamp or
+    date column is one bridge ``TemporalColumn`` instead (``_temporal_column``: one ``long[]`` of epoch ticks, no value per cell).
 
     ``float_vectors`` (the JVM backend, a property declared ``ARRAY_OF_FLOATS``): a float32 vector column of one length and no null is
     one 2-D float32 matrix, which ``insert_columns`` sends as one ``float[]``; ``to_pylist`` made one Python float per element and
@@ -434,9 +438,7 @@ def _column_values(
         if matrix is not None:
             return matrix
     if kind in ("timestamp", "date"):
-        return (
-            _temporal_to_tagged(chunk, pa) if native else _temporal_to_java(chunk, pa)
-        )
+        return _temporal_to_tagged(chunk, pa) if native else _temporal_column(chunk, pa)
     if native and kind in ("decimal", "binary"):
         return _scalars_to_tagged(kind, chunk)
     if kind == "null":

src/java/com/arcadedb/python/DocumentBatcher.java

@@ -81,17 +81,17 @@ public static long insertManyJson(final Database db, final String typeName, fina
 
   /**
    * Columnar insert: the documents are built Java-side from whole columns, so each column crosses the FFI once (a long[] or
-   * double[] copied from a numpy buffer, a boolean[], a StringColumn (one UTF-8 byte[] plus the int[] offsets), or an Object[] of
-   * boxed numbers, Strings of a mixed column, and nulls) instead of one
+   * double[] copied from a numpy buffer, a boolean[], a StringColumn (one UTF-8 byte[] plus the int[] offsets), a TemporalColumn (one
+   * long[] of epoch ticks), or an Object[] of boxed numbers, Strings of a mixed column, and nulls) instead of one
    * JSON text per batch (bindings issue #150: 2.24x over insertManyJson on the first
    * 2,000,000 TPC-H SF1 line items, nine typed properties, commit every 10,000, same sums and count).
    *
    * Same failure contract as insertManyJson (#7882): batch commits and the failure rollback apply only to a transaction
    * opened here; a caller's own transaction is the caller's to commit or roll back.
    *
-   * @param columns one array per name, each of length n: long[], double[], boolean[], Object[], a FloatVectors.Matrix or a
-   *                StringColumn; a null element (or null row of a StringColumn) sets the property to null, as a JSON null does
-   *                on the insertManyJson path
+   * @param columns one array per name, each of length n: long[], double[], boolean[], Object[], a FloatVectors.Matrix, a
+   *                StringColumn or a TemporalColumn; a null element (or null row of a StringColumn or TemporalColumn) sets the
+   *                property to null, as a JSON null does on the insertManyJson path
    */
   public static long insertColumns(final Database db, final String typeName, final String[] names, final Object[] columns,
       final int n, final int commitEvery) {
@@ -229,9 +229,11 @@ else if (col instanceof FloatVectors.Matrix m)
         len = m.rows();
       else if (col instanceof StringColumn s)
         len = s.rows();
+      else if (col instanceof TemporalColumn t)
+        len = t.rows();
       else
         throw new IllegalArgumentException("column '" + names[c] + "' is a " + (col == null ? "null" : col.getClass().getName())
-            + ", not a long[], double[], boolean[], Object[], FloatVectors.Matrix or StringColumn");
+            + ", not a long[], double[], boolean[], Object[], FloatVectors.Matrix, StringColumn or TemporalColumn");
       if (len != n)
         throw new IllegalArgumentException("column '" + names[c] + "' has " + len + " values for " + n + " rows");
     }
@@ -251,6 +253,8 @@ else if (col instanceof FloatVectors.Matrix m)
         v = m.row(i);
       else if (col instanceof StringColumn s)
         v = s.get(i);
+      else if (col instanceof TemporalColumn t)
+        v = t.get(i);
       else
         v = ((Object[]) col)[i];
       doc.set(names[c], v);

src/java/com/arcadedb/python/TemporalColumn.java

@@ -0,0 +1,118 @@
+/*
+ * Python-bindings bridge: a timestamp or date column of Database.insert_columns that crosses as one long[] of epoch ticks.
+ *
+ * Building a LocalDateTime or LocalDate in Python costs one JPype call per cell (and the list of them one conversion per element
+ * again), so a million-row timestamp column was two million calls. Here the whole column is ONE long[] (the integer epoch value of every
+ * row, as Arrow holds it), the tick size, and optionally ONE boolean[] marking the null rows. Row i is read when the document of that
+ * row is built, so the values are short-lived.
+ *
+ * A date-time is {@code ticks / ticksPerSecond} whole seconds (rounded toward negative infinity, so a tick before the epoch is the
+ * last tick of 1969) and the rest as nanoseconds, read as the UTC wall clock: the engine stores a DATETIME as a UTC wall clock, a
+ * zone-aware Arrow timestamp is already UTC epoch time and a zone-less one is taken as the wall clock it shows. The unit's precision is
+ * kept: a nanosecond tick is not cut to microseconds. A date is {@code ticks / ticksPerDay} days since the epoch, rounded the same way
+ * (a date32 is days, a date64 is milliseconds).
+ *
+ * Compiled into arcadedb-python-bridge.jar during the wheel build and consumed by DocumentBatcher.insertColumns() and
+ * insertColumnsParallel().
+ */
+package com.arcadedb.python;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+
+public final class TemporalColumn {
+  private static final long NANOS_PER_SECOND = 1_000_000_000L;
+
+  private final long[]    ticks;
+  private final boolean[] nulls;
+  private final boolean   date;
+  /** Ticks in one second (a date-time) or in one day (a date). */
+  private final long      ticksPerUnit;
+  /** Nanoseconds in one tick, for a date-time. */
+  private final long      nanosPerTick;
+
+  private TemporalColumn(final long[] ticks, final boolean[] nulls, final boolean date, final long ticksPerUnit,
+      final long nanosPerTick) {
+    if (nulls != null && nulls.length != ticks.length)
+      throw new IllegalArgumentException("a temporal column has " + nulls.length + " null flags for " + ticks.length + " rows");
+    this.ticks = ticks;
+    this.nulls = nulls;
+    this.date = date;
+    this.ticksPerUnit = ticksPerUnit;
+    this.nanosPerTick = nanosPerTick;
+    checkRange();
+  }
+
+  /**
+   * A column of date-times.
+   *
+   * @param ticks          the epoch value of every row, in units of {@code 1 / ticksPerSecond} second since 1970-01-01T00:00:00
+   * @param ticksPerSecond 1 (seconds), 1,000, 1,000,000 or 1,000,000,000: any divisor of a second into whole nanoseconds
+   * @param nulls          null for a column without a null, else one flag per row (a flagged row reads as null and its tick is ignored)
+   * @throws IllegalArgumentException for a tick size that is not a whole number of nanoseconds, or a wrong number of null flags.
+   *                                  A row past the years a LocalDateTime holds raises the DateTimeException of LocalDateTime itself,
+   *                                  here, before anything is written
+   */
+  public static TemporalColumn dateTimes(final long[] ticks, final long ticksPerSecond, final boolean[] nulls) {
+    if (ticksPerSecond < 1 || NANOS_PER_SECOND % ticksPerSecond != 0)
+      throw new IllegalArgumentException(
+          "ticks per second must divide 1,000,000,000 into whole nanoseconds, got " + ticksPerSecond);
+    return new TemporalColumn(ticks, nulls, false, ticksPerSecond, NANOS_PER_SECOND / ticksPerSecond);
+  }
+
+  /**
+   * A column of dates.
+   *
+   * @param ticks       the epoch value of every row, in units of {@code 1 / ticksPerDay} day since 1970-01-01
+   * @param ticksPerDay 1 for a date32 (days), 86,400,000 for a date64 (milliseconds)
+   * @param nulls       null for a column without a null, else one flag per row (a flagged row reads as null and its tick is ignored)
+   * @throws IllegalArgumentException for a tick size below 1, or a wrong number of null flags. A row past the days a LocalDate holds
+   *                                  raises the DateTimeException of LocalDate itself, here, before anything is written
+   */
+  public static TemporalColumn dates(final long[] ticks, final long ticksPerDay, final boolean[] nulls) {
+    if (ticksPerDay < 1)
+      throw new IllegalArgumentException("ticks per day must be at least 1, got " + ticksPerDay);
+    return new TemporalColumn(ticks, nulls, true, ticksPerDay, 0L);
+  }
+
+  public int rows() {
+    return ticks.length;
+  }
+
+  /** The LocalDateTime or LocalDate of row {@code i}, or null for a null row. */
+  public Object get(final int i) {
+    if (nulls != null && nulls[i])
+      return null;
+    return valueOf(ticks[i]);
+  }
+
+  private Object valueOf(final long tick) {
+    if (date)
+      return LocalDate.ofEpochDay(Math.floorDiv(tick, ticksPerUnit));
+    return LocalDateTime.ofEpochSecond(Math.floorDiv(tick, ticksPerUnit), (int) (Math.floorMod(tick, ticksPerUnit) * nanosPerTick),
+        ZoneOffset.UTC);
+  }
+
+  /**
+   * The value of a row grows with its tick, so the smallest and the largest tick of the non-null rows are the two that can fall outside
+   * the years of java.time: a value past them raises here, as one call per cell did in Python, and nothing of the batch is written.
+   */
+  private void checkRange() {
+    long min = Long.MAX_VALUE;
+    long max = Long.MIN_VALUE;
+    for (int i = 0; i < ticks.length; i++) {
+      if (nulls != null && nulls[i])
+        continue;
+      final long tick = ticks[i];
+      if (tick < min)
+        min = tick;
+      if (tick > max)
+        max = tick;
+    }
+    if (min <= max) {
+      valueOf(min);
+      valueOf(max);
+    }
+  }
+}

tests/conftest.py

@@ -154,6 +154,7 @@
     "test_graph_batch_vector_crossings_jvm.py": "the JPype crossings of the JVM wheel's create_vertices vector path (the bridge classes, the property-matrix fallback)",
     "test_insert_columns_string_blob.py": "the bridge's StringColumn and the JVM wheel's core._column_to_java (JPype arrays, a spy on the per-element conversion)",
     "test_arrow_vector_column_jvm.py": "the JPype crossings of the JVM wheel's Arrow vector column (core.convert_python_to_java, the bridge's FloatVectors.Matrix) and its Python heap",
+    "test_arrow_temporal_column_jvm.py": "the JPype calls of the JVM wheel's Arrow timestamp and date columns (the bridge's TemporalColumn, core._column_to_java, java.time classes)",
     "test_async_proxy_cache.py": "the JPype proxy classes of the JVM build's async executor",
     "test_read_crossings.py": "crossing counts of the JVM build's read path (JPype proxies of the Java objects and the bridge classes)",
     "test_reader_crossings.py": "crossing counts of the JVM build's record and row readers (JPype proxies of the Java objects and the bridge classes)",