From 67b44a18e4db71c8fcde308816fc41def374b44b Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 1 Nov 2026 21:07:34 +0900 Subject: [PATCH] JVM columnar readers return the values the row readers do to_dataframe keeps microseconds and dates (C2): ColumnBatcher takes a micros flag and writes datetime columns as epoch microseconds, DATE and tz-aware columns as tagged JSON; to_columns and to_arrow keep their documented milliseconds. A json column is written with TypedRows' tagged encoding plus a side array of engine objects, so a date, datetime, Decimal or int map key inside a list or map reads as to_list() reads it (B3). An empty float vector has no dim in the header and reads as empty (B2). insert_many refuses a subnormal double for the bulk JSON path, whose parser read 5e-324 as a BigDecimal (A4). Java is not compiled here; CI builds it. Against the main 26.11.1.dev0 wheels the JVM wheel fails the new tests (33 failed) and the native wheel passes the whole file (67 passed). The Java change is compiled and tested by CI. Closes #178 Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/results.py | 89 ++++++--- src/arcadedb/type_conversion.py | 35 +++- .../com/arcadedb/python/ColumnBatcher.java | 67 ++++++- src/java/com/arcadedb/python/TypedRows.java | 3 +- tests/test_columnar_readers.py | 175 ++++++++++++++++++ 5 files changed, 335 insertions(+), 34 deletions(-) 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