JVM columnar readers return the values the row readers do

src/arcadedb/results.py

@@ -165,7 +165,7 @@ def _missing_column_part(count: int, template):
         if template.dtype.kind in "iuf":
             return np.full(count, np.nan)
         if template.dtype.kind == "M":
-            return np.full(count, np.datetime64("NaT", "ms"))
+            return np.full(count, np.datetime64("NaT"), dtype=template.dtype)
         if template.dtype.kind == "O":
             return np.full(count, None, dtype=object)
     return [None] * count
@@ -613,6 +613,11 @@ def to_dataframe(self, convert_types: bool = True):
 
         Requires pandas to be installed.
 
+        Values are the ones ``to_list()`` returns: a DATETIME keeps its
+        microseconds (a ``datetime64[us]`` column, where ``to_columns()`` and
+        ``to_arrow()`` carry milliseconds), a DATE is a ``datetime.date`` in an
+        object column, and a ``Decimal`` stays one.
+
         Args:
             convert_types: Convert Java types to Python (default: True)
 
@@ -638,7 +643,10 @@ def to_dataframe(self, convert_types: bool = True):
             # fast path: columnar binary transport straight into typed
             # columns (numpy dtypes, real datetime64) — measured ~2x the JSON
             # row path and ~6x the per-row path on 100k-row scans
-            columns = self.to_columns()
+            # Datetimes cross with their microseconds (datetime64[us]) and dates as
+            # date objects, the values to_list() returns; to_columns() and
+            # to_arrow() keep the documented milliseconds.
+            columns = self._read_columns(25_000, None, micros=True)
             if columns is not None:
                 # pandas rejects 2-D ndarrays as column values; vector
                 # columns (f4v/f8v) become object columns of row slices
@@ -713,9 +721,22 @@ def to_columns(
         those, as a projection would, and skip it. A row lacking one of them
         reads null, and properties not listed are left out.
 
+        List and map values (and a column of mixed types) are decoded as
+        ``to_list()`` decodes them: a ``date``, ``datetime`` or ``Decimal``
+        inside a list or map stays one, and a map keeps its non-str keys. The
+        ``datetime64[ms]`` rule above is for a column whose values are all
+        datetimes.
+
         Returns None when numpy or the bridge jar is unavailable (callers
         fall back to row-based paths).
         """
+        return self._read_columns(batch_size, columns, micros=False)
+
+    def _read_columns(self, batch_size, columns, micros):
+        """``to_columns()``; with ``micros`` a column of datetimes is a
+        ``datetime64[us]`` array and a column holding DATE values is an object
+        column of ``date`` objects, which is what ``to_dataframe()`` needs to
+        return the values ``to_list()`` does."""
         try:
             import numpy as np
         except ImportError:
@@ -733,7 +754,7 @@ def to_columns(
         names: List[str] = []
         batches: List[Tuple[int, Dict[str, Any]]] = []
 
-        def decode_batch(buf):
+        def decode_batch(buf, side):
             hlen = int.from_bytes(buf[:4], "little")
             header = json.loads(bytes(buf[4 : 4 + hlen]))
             count = header["count"]
@@ -768,6 +789,11 @@ def decode_batch(buf):
                     if has_nulls:
                         arr[mask] = np.datetime64("NaT", "ms")
                     values = arr
+                elif ctype == "dtu":
+                    arr = np.frombuffer(data, dtype="<i8").astype("datetime64[us]")
+                    if has_nulls:
+                        arr[mask] = np.datetime64("NaT", "us")
+                    values = arr
                 elif ctype == "b1":
                     arr = np.frombuffer(data, dtype=np.uint8).astype(bool)
                     if has_nulls:
@@ -779,11 +805,16 @@ def decode_batch(buf):
                 elif ctype in ("f4v", "f8v"):
                     # fixed-dimension vector column -> 2-D array (count, dim);
                     # Java already writes NaN rows into null slots
-                    dim = col["dim"]
+                    # an empty vector has no "dim" in the header
+                    dim = col.get("dim", 0)
                     dt = "<f4" if ctype == "f4v" else "<f8"
-                    values = np.frombuffer(data, dtype=dt).reshape(count, dim)
+                    if dim == 0:
+                        values = np.empty((count, 0), dtype=dt)
+                    else:
+                        values = np.frombuffer(data, dtype=dt).reshape(count, dim)
                 elif ctype == "json":
-                    values = json.loads(bytes(data))
+                    # tagged JSON (TypedRows): the values to_list() returns
+                    values = json.loads(bytes(data), object_hook=_typed_hook(side))
                 elif ctype == "dec":
                     # an object array of Decimal (None for null): a double
                     # would lose digits, and the dtype would follow the data
@@ -809,13 +840,13 @@ def decode_batch(buf):
         total = 0
         while True:
             try:
-                java_buf = column_batcher.nextColumnBatch(
-                    self._java_result_set, int(batch_size), spec
+                java_batch = column_batcher.nextColumnBatch(
+                    self._java_result_set, int(batch_size), spec, micros
                 )
             except JException as exc:
                 raise _read_error(exc) from exc
-            buf = memoryview(bytes(java_buf))
-            count = decode_batch(buf)
+            buf = memoryview(bytes(java_batch[0]))
+            count = decode_batch(buf, java_batch[1])
             if count == 0:
                 self._finish()
                 break
@@ -898,7 +929,7 @@ def validity(null_bits, count, has_nulls):
                 return None
             return pa.py_buffer(bytes(np.invert(null_bits).tobytes()))
 
-        def decode_batch(buf):
+        def decode_batch(buf, side):
             hlen = int.from_bytes(buf[:4], "little")
             header = json.loads(bytes(buf[4 : 4 + hlen]))
             count = header["count"]
@@ -934,15 +965,29 @@ def decode_batch(buf):
                     vals = np.frombuffer(data, dtype=np.uint8).astype(bool)
                     arr = pa.array(vals, mask=mask)
                 elif ctype in ("f4v", "f8v"):
-                    dim = col["dim"]
+                    # an empty vector has no "dim" in the header
+                    dim = col.get("dim", 0)
                     dt = "<f4" if ctype == "f4v" else "<f8"
-                    flat = np.frombuffer(data, dtype=dt)
-                    child = pa.array(flat)
-                    arr = pa.FixedSizeListArray.from_arrays(child, dim)
-                    if mask is not None:
-                        arr = pa.array(arr.to_pylist(), type=arr.type, mask=mask)
+                    if dim == 0:
+                        # no fixed-size list of size 0: empty lists, as the row readers give
+                        arr = pa.array(
+                            [
+                                None if mask is not None and mask[i] else []
+                                for i in range(count)
+                            ],
+                            type=pa.list_(
+                                pa.float32() if ctype == "f4v" else pa.float64()
+                            ),
+                        )
+                    else:
+                        flat = np.frombuffer(data, dtype=dt)
+                        child = pa.array(flat)
+                        arr = pa.FixedSizeListArray.from_arrays(child, dim)
+                        if mask is not None:
+                            arr = pa.array(arr.to_pylist(), type=arr.type, mask=mask)
                 elif ctype == "json":
-                    values = json.loads(bytes(data))
+                    # tagged JSON (TypedRows): the values to_list() returns
+                    values = json.loads(bytes(data), object_hook=_typed_hook(side))
                     try:
                         arr = pa.array(values)
                     except pa.ArrowException:
@@ -984,13 +1029,13 @@ def decode_batch(buf):
         total = 0
         while True:
             try:
-                java_buf = column_batcher.nextColumnBatch(
-                    self._java_result_set, int(batch_size), spec
+                java_batch = column_batcher.nextColumnBatch(
+                    self._java_result_set, int(batch_size), spec, False
                 )
             except JException as exc:
                 raise _read_error(exc) from exc
-            buf = memoryview(bytes(java_buf))
-            count = decode_batch(buf)
+            buf = memoryview(bytes(java_batch[0]))
+            count = decode_batch(buf, java_batch[1])
             if count == 0:
                 self._finish()
                 break

src/arcadedb/type_conversion.py

@@ -7,6 +7,7 @@
 
 import json
 import math
+import sys
 from datetime import date, datetime, time, timedelta, timezone
 from decimal import Decimal
 from typing import Any, NamedTuple
@@ -651,6 +652,7 @@ def convert_python_to_java(value: Any) -> Any:
 # surrogate becomes "?". A value the JSON text cannot carry unchanged keeps the
 # call off the bulk path: the per-value path stores it exactly or raises.
 
+_FLOAT_NORMAL_MIN = sys.float_info.min  # below it (and above 0) a double is subnormal
 _INT64_MIN = -(2**63)
 _INT64_MAX = 2**63 - 1
 
@@ -731,6 +733,35 @@ def _json_bulk_check(value: Any) -> None:
         )
 
 
+def _has_subnormal(value: Any) -> bool:
+    """Whether a float anywhere in `value` (a row value, a list or a map) is a subnormal double."""
+    if isinstance(value, dict):
+        return any(_has_subnormal(item) for item in value.values())
+    if isinstance(value, (list, tuple)):
+        return any(_has_subnormal(item) for item in value)
+    if isinstance(value, float) or (
+        _np is not None and isinstance(value, _np.floating)
+    ):
+        return 0.0 < abs(float(value)) < _FLOAT_NORMAL_MIN
+    return False
+
+
+def _dumps_checked(rows: Any, **kwargs: Any) -> str:
+    """json.dumps for the bulk paths, refusing a subnormal double.
+
+    The engine's JSON parser reads the text of a subnormal (``5e-324``) as a
+    BigDecimal, so it was stored as a DECIMAL and read back as ``Decimal``
+    where every other write path stores a double. ValueError sends the row down
+    the per-row path, which stores the double. Every subnormal is written with
+    an exponent of -308 or below, so the text is searched first (C speed) and
+    only a hit pays for the walk over the rows.
+    """
+    text = json.dumps(rows, allow_nan=False, **kwargs)
+    if "e-3" in text and any(_has_subnormal(row) for row in rows):
+        raise ValueError("a subnormal double cannot go through the JSON bulk path")
+    return text
+
+
 def json_bulk_dumps(rows: Any) -> str:
     """json.dumps for the bulk paths: raises ValueError (or TypeError) when
     any value in `rows` would change on the way to the engine.
@@ -763,7 +794,7 @@ def json_bulk_dumps(rows: Any) -> str:
                 _json_bulk_check(
                     value
                 )  # nested lists and dicts, and subclasses of the types above
-    return json.dumps(rows, allow_nan=False)
+    return _dumps_checked(rows)
 
 
 # ---------------------------------------------------------------------------
@@ -896,7 +927,7 @@ def json_bulk_dumps_tagged(rows: Any) -> str:
                 _json_bulk_check(
                     value
                 )  # nested lists and dicts, and subclasses of the types above
-    return json.dumps(rows, allow_nan=False, default=bulk_tag_default)
+    return _dumps_checked(rows, default=bulk_tag_default)
 
 
 def bulk_row_ok(row: Any, skip=()) -> bool:

src/java/com/arcadedb/python/ColumnBatcher.java

@@ -9,6 +9,13 @@
  * fit those encodings are serialized as one JSON array ("json" type). Python decodes with numpy.frombuffer
  * (ResultSet.to_columns / the fast to_dataframe path). Measured ~1.2x of
  * Java-native iteration on 100k-row scans (vs 3.4x for the JSON row path).
+ *
+ * A "json" column is written with the tagged encoding of TypedRows, not the engine's JSON writer: a date, datetime,
+ * DECIMAL or non-String map key inside a list or a map reaches Python as the type ResultSet.to_list() gives it, and the
+ * values JSON cannot carry (NaN, RIDs, ...) are the engine objects of the side array that is returned with the buffer.
+ * With {@code micros} a datetime column is written as epoch MICROseconds ("dtu", what ResultSet.to_dataframe reads) and
+ * any column holding a DATE or a time-zone-aware temporal is a "json" column, so Python gets date and datetime objects as
+ * the row readers return them.
  */
 package com.arcadedb.python;
 
@@ -64,7 +71,33 @@ private static long toEpochMillis(final Object v) {
     throw new IllegalArgumentException("not a temporal: " + v.getClass());
   }
 
-  public static byte[] nextColumnBatch(final ResultSet rs, final int max, final String joinedColumns) {
+  /** Epoch microseconds of a naive datetime: what {@link #toEpochMillis} is in milliseconds, without the cut. */
+  private static long toEpochMicros(final Object v) {
+    if (v instanceof LocalDateTime ldt)
+      return Math.addExact(Math.multiplyExact(ldt.toEpochSecond(ZoneOffset.UTC), 1_000_000L), ldt.getNano() / 1000);
+    if (v instanceof Date d)
+      return Math.multiplyExact(d.getTime(), 1000L);
+    if (v instanceof Calendar c)
+      return Math.multiplyExact(c.getTimeInMillis(), 1000L);
+    throw new IllegalArgumentException("not a naive datetime: " + v.getClass());
+  }
+
+  /** Python's datetime covers years 1 to 9999; a datetime outside it stays an engine object (TypedRows does the same). */
+  private static boolean yearFits(final LocalDateTime ldt) {
+    return ldt.getYear() >= 1 && ldt.getYear() <= 9999;
+  }
+
+  /** The columns of up to {@code max} rows, temporals in milliseconds: {@code {byte[] buffer, Object[] side}}. */
+  public static Object[] nextColumnBatch(final ResultSet rs, final int max, final String joinedColumns) {
+    return nextColumnBatch(rs, max, joinedColumns, false);
+  }
+
+  /**
+   * The columns of up to {@code max} rows as {@code {byte[] buffer, Object[] side}}: {@code side} holds the engine objects
+   * that the tagged JSON of a "json" column refers to by index. With {@code micros} see the class comment.
+   */
+  public static Object[] nextColumnBatch(final ResultSet rs, final int max, final String joinedColumns,
+      final boolean micros) {
     // empty column spec: the column set is the union of every row's property names, in order of first appearance.
     // A document is schemaless, so the first row says nothing about the others: taking its names alone dropped every
     // property it lacked (humemai/arcadedb-embedded-python#113). Each batch reports its own set in its header.
@@ -110,6 +143,7 @@ public static byte[] nextColumnBatch(final ResultSet rs, final int max, final St
     // emit invalid JSON and break the decode of the WHOLE batch on the Python side. RowBatcher already does it
     // this way; the cost is n objects per batch, not per row.
     final JSONArray headerColumns = new JSONArray();
+    final List<Object> side = new ArrayList<>();
     final ByteArrayOutputStream body = new ByteArrayOutputStream(count * n * 8);
 
     for (int c = 0; c < n; c++) {
@@ -130,8 +164,16 @@ else if (v instanceof Boolean)
           t = "b1";
         else if (v instanceof Date || v instanceof Calendar || v instanceof LocalDateTime
             || v instanceof LocalDate || v instanceof Instant
-            || v instanceof ZonedDateTime || v instanceof OffsetDateTime)
-          t = "dt";
+            || v instanceof ZonedDateTime || v instanceof OffsetDateTime) {
+          if (!micros)
+            t = "dt";
+          else if (v instanceof LocalDateTime ldt ? yearFits(ldt) : v instanceof Date || v instanceof Calendar)
+            // a naive datetime is epoch microseconds; a DATE, a time-zone-aware temporal or a year Python cannot hold
+            // is read through the tagged JSON, as the row readers return it
+            t = "dtu";
+          else
+            t = "json";
+        }
         else if (v instanceof String || v instanceof Character)
           t = "str";
         else if (v instanceof java.math.BigDecimal)
@@ -175,7 +217,7 @@ else if (vdim != len)
       }
 
       byte[] colBuf;
-      if (type.equals("i8") || type.equals("dt")) {
+      if (type.equals("i8") || type.equals("dt") || type.equals("dtu")) {
         final ByteBuffer bb = ByteBuffer.allocate(count * 8).order(ByteOrder.LITTLE_ENDIAN);
         for (int r = 0; r < count; r++) {
           final Object v = rows.get(r)[c];
@@ -184,6 +226,8 @@ else if (vdim != len)
             bb.putLong(0);
           } else if (type.equals("dt"))
             bb.putLong(toEpochMillis(v));
+          else if (type.equals("dtu"))
+            bb.putLong(toEpochMicros(v));
           else
             bb.putLong(((Number) v).longValue());
         }
@@ -237,15 +281,20 @@ else if (vdim != len)
         }
         colBuf = bb.array();
       } else if (type.equals("json")) {
-        // one JSON array for the whole column (JSONObject/JSONArray handle
-        // primitive arrays natively since engine 26.7.2, #4967)
-        final com.arcadedb.serializer.json.JSONArray arr = new com.arcadedb.serializer.json.JSONArray();
+        // one JSON array for the whole column, in the tagged encoding of TypedRows: what the engine's JSON writer turns
+        // into epoch milliseconds, a double or a text key (a date or Decimal in a list, an int map key) stays what the
+        // row readers return, and what JSON cannot carry goes to the side array
+        final StringBuilder arr = new StringBuilder();
+        arr.append('[');
         for (int r = 0; r < count; r++) {
           final Object v = rows.get(r)[c];
+          if (r > 0)
+            arr.append(',');
           if (v == null)
             nulls[r >> 3] |= (1 << (r & 7));
-          arr.put(v);
+          TypedRows.append(arr, v, side);
         }
+        arr.append(']');
         colBuf = arr.toString().getBytes(StandardCharsets.UTF_8);
       } else { // strings and decimals: int32 offsets (count+1) then utf8 bytes
         final ByteArrayOutputStream chars = new ByteArrayOutputStream(count * 16);
@@ -295,7 +344,7 @@ else if (vdim != len)
     out.putInt(headerBytes.length);
     out.put(headerBytes);
     out.put(bodyBytes);
-    return out.array();
+    return new Object[] { out.array(), side.toArray() };
   }
 
   /**

src/java/com/arcadedb/python/TypedRows.java

@@ -117,7 +117,8 @@ private static void appendRow(final StringBuilder sb, final Result row, final Li
     sb.append('}');
   }
 
-  private static void append(final StringBuilder sb, final Object value, final List<Object> side) {
+  /** One value as tagged JSON (also the "json" columns of {@link ColumnBatcher}); what JSON cannot carry goes to {@code side}. */
+  static void append(final StringBuilder sb, final Object value, final List<Object> side) {
     if (value == null) {
       sb.append("null");
     } else if (value instanceof String s) {