diff --git a/src/arcadedb/results.py b/src/arcadedb/results.py index 41ec46a4..313d83e5 100644 --- a/src/arcadedb/results.py +++ b/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=" 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 = " 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: diff --git a/src/java/com/arcadedb/python/ColumnBatcher.java b/src/java/com/arcadedb/python/ColumnBatcher.java index 6211ac0d..cf971a20 100644 --- a/src/java/com/arcadedb/python/ColumnBatcher.java +++ b/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 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() }; } /** diff --git a/src/java/com/arcadedb/python/TypedRows.java b/src/java/com/arcadedb/python/TypedRows.java index 5fdef579..9c2e0f82 100644 --- a/src/java/com/arcadedb/python/TypedRows.java +++ b/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 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 side) { if (value == null) { sb.append("null"); } else if (value instanceof String s) { diff --git a/tests/test_columnar_readers.py b/tests/test_columnar_readers.py index ef6877e8..b4c74cfc 100644 --- a/tests/test_columnar_readers.py +++ b/tests/test_columnar_readers.py @@ -10,8 +10,10 @@ """ import math +from datetime import date, datetime from decimal import Decimal +import numpy as np import pytest import arcadedb @@ -302,3 +304,176 @@ def test_pinned_columns_are_read_as_given(events_db, batch_size): assert table.column_names == ["n", "missing"] assert table.column("n").to_pylist() == [1, 2, 3] assert table.column("missing").to_pylist() == [None, None, None] + + +# --------------------------------------------------------------------------- +# The columnar readers return the values the row readers do (native-parity +# triage 2026-11-01: C2, B3, B2, A4). The expected value is always what +# to_list() returns for the stored row, the same on both builds. + +STAMP = datetime(2026, 10, 3, 12, 34, 56, 123456) +DAY = date(2026, 10, 3) + + +@pytest.fixture +def stamp_db(temp_db_path): + """One row with a datetime that has microseconds and a date, and one row with neither.""" + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE Stamp") + db.insert_many( + "Stamp", + [{"n": 0, "t": STAMP, "d": DAY}, {"n": 1, "t": None, "d": None}], + ) + yield db + + +STAMP_QUERY = "SELECT n, t, d FROM Stamp ORDER BY n" + + +def test_to_dataframe_keeps_microseconds_and_dates(stamp_db): + """C2: to_dataframe went through to_columns' datetime64[ms]: the microseconds were + dropped and a date became a Timestamp. It returns the values to_list() does.""" + row = stamp_db.query("sql", STAMP_QUERY).to_list()[0] + assert row["t"] == STAMP and row["d"] == DAY # what is stored + + frame = stamp_db.query("sql", STAMP_QUERY).to_dataframe() + assert frame["t"].dtype.kind == "M" + assert frame["t"].iloc[0].to_pydatetime() == STAMP + assert pd.isna(frame["t"].iloc[1]) + assert type(frame["d"].iloc[0]) is date + assert frame["d"].iloc[0] == DAY + assert pd.isna(frame["d"].iloc[1]) + + +def test_to_columns_and_to_arrow_keep_the_documented_milliseconds(stamp_db): + """to_columns() is documented as datetime64[ms], and to_arrow() as timestamp[ms]: a + DATE is its midnight. C2 changes to_dataframe only.""" + columns = stamp_db.query("sql", STAMP_QUERY).to_columns() + assert columns["t"].dtype == np.dtype("datetime64[ms]") + assert columns["t"][0] == np.datetime64("2026-10-03T12:34:56.123", "ms") + assert np.isnat(columns["t"][1]) + assert columns["d"].dtype == np.dtype("datetime64[ms]") + assert columns["d"][0] == np.datetime64("2026-10-03", "ms") + + table = stamp_db.query("sql", STAMP_QUERY).to_arrow() + assert table.schema.field("t").type == pa.timestamp("ms") + assert table.column("t").to_pylist() == [ + datetime(2026, 10, 3, 12, 34, 56, 123000), + None, + ] + assert table.schema.field("d").type == pa.timestamp("ms") + + +NESTED_CASES = { + "list of datetimes": [STAMP], + "list of dates": [DAY], + "list of decimals": [Decimal("1.10")], + "list of mixed temporals and a decimal": [STAMP, DAY, Decimal("1.10")], + "map with a date": {"k": DAY}, + "map with a datetime": {"k": STAMP}, + "map with a decimal": {"k": Decimal("1.10")}, + "map with int keys": {1: "a", 2: "b"}, +} +# a list that holds one kind of value is an Arrow list of it; the others have no one Arrow type +NESTED_IN_ARROW = [ + "list of datetimes", + "list of dates", + "list of decimals", + "map with a date", + "map with a datetime", + "map with a decimal", +] + + +@pytest.fixture +def nested_db(temp_db_path): + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE Nested") + yield db + + +def _store_nested(db, value): + with db.transaction(): + db.command("sql", "DELETE FROM Nested") + with db.transaction(): + db.command("sql", "INSERT INTO Nested SET n = 0, v = :v", {"v": value}) + return db.query("sql", "SELECT n, v FROM Nested").to_list()[0]["v"] + + +@pytest.mark.parametrize("batch_size", [25_000, 1]) +@pytest.mark.parametrize("case", list(NESTED_CASES)) +def test_nested_values_are_read_as_the_row_reader_reads_them( + nested_db, case, batch_size +): + """B3: a date, datetime or Decimal inside a list or map, and an int map key, came + back from the columnar readers as epoch milliseconds, a float and a str.""" + stored = _store_nested(nested_db, NESTED_CASES[case]) + assert stored == NESTED_CASES[case] # the row reader returns it whole + q = "SELECT n, v FROM Nested" + + column = nested_db.query("sql", q).to_columns(batch_size=batch_size)["v"] + # repr: the type of every element and the scale of a Decimal count, not only == + assert repr(list(column)) == repr([stored]) + + frame = nested_db.query("sql", q).to_dataframe() + assert repr(frame["v"].iloc[0]) == repr(stored) + + +@pytest.mark.parametrize("batch_size", [25_000, 1]) +@pytest.mark.parametrize("case", NESTED_IN_ARROW) +def test_nested_values_in_arrow_keep_their_types(nested_db, case, batch_size): + stored = _store_nested(nested_db, NESTED_CASES[case]) + table = nested_db.query("sql", "SELECT n, v FROM Nested").to_arrow( + batch_size=batch_size + ) + assert table.column("v").to_pylist() == [stored] + + +@pytest.fixture +def vector_db(temp_db_path): + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE Vec") + with db.transaction(): + db.command( + "sql", + "INSERT INTO Vec SET n = ?, v = ?", + 0, + np.array([], dtype=np.float32), + ) + yield db + + +@pytest.mark.parametrize("batch_size", [25_000, 1]) +def test_an_empty_float32_vector_reads_as_empty(vector_db, batch_size): + """B2: an empty float32 vector made to_columns, to_dataframe and to_arrow raise + KeyError: 'dim' on the JVM build; native returns [].""" + q = "SELECT n, v FROM Vec" + assert list(vector_db.query("sql", q).to_list()[0]["v"]) == [] + + columns = vector_db.query("sql", q).to_columns(batch_size=batch_size) + assert [list(x) for x in columns["v"]] == [[]] + frame = vector_db.query("sql", q).to_dataframe() + assert [list(x) for x in frame["v"]] == [[]] + table = vector_db.query("sql", q).to_arrow(batch_size=batch_size) + assert table.column("v").to_pylist() == [[]] + + +@pytest.mark.parametrize( + "value", + [ + pytest.param(5e-324, id="smallest-subnormal"), + pytest.param(2.5e-320, id="subnormal"), + pytest.param(2.2250738585072014e-308, id="smallest-normal"), + pytest.param([1.0, 5e-324], id="in-a-list"), + ], +) +def test_insert_many_keeps_a_tiny_double_a_float(temp_db_path, value): + """A4: 5e-324 sent through insert_many was stored as a DECIMAL and read back as + Decimal('5E-324'); every other write path stores a double.""" + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE Tiny") + db.insert_many("Tiny", [{"n": 0, "x": value}]) + x = db.query("sql", "SELECT x FROM Tiny").to_list()[0]["x"] + assert x == value + for element in x if isinstance(x, list) else [x]: + assert type(element) is float