From 2d7760a224edc8bf4d1bd4f945a27d30cfc06333 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:47:57 +0900 Subject: [PATCH 1/2] Tests first: timestamp and date columns of insert_arrow and insert_dataframe on the JVM wheel (refs #315) Round trips of every Arrow temporal type through a real load on both wheels (every timestamp unit with and without a zone, to the nanosecond; date32 and date64 before 1970; nulls; batches; vertex and parallel loads; pandas and Polars frames), and JVM-only tests that count what one load does: no java.time call and no convert_python_to_java per cell, and each timestamp or date column reaching the bridge as one TemporalColumn per batch. The counting tests fail on the code before, which builds one LocalDateTime or LocalDate per cell in Python and converts every element again. Co-Authored-By: Claude Sonnet 5.5 --- tests/conftest.py | 1 + tests/test_arrow_temporal_column_jvm.py | 271 ++++++++++++++++++++++++ tests/test_dataframe_ingest.py | 229 ++++++++++++++++++++ 3 files changed, 501 insertions(+) create mode 100644 tests/test_arrow_temporal_column_jvm.py diff --git a/tests/conftest.py b/tests/conftest.py index fd7402f3..a58d5ede 100644 --- a/tests/conftest.py +++ b/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)", diff --git a/tests/test_arrow_temporal_column_jvm.py b/tests/test_arrow_temporal_column_jvm.py new file mode 100644 index 00000000..ef7c6dc6 --- /dev/null +++ b/tests/test_arrow_temporal_column_jvm.py @@ -0,0 +1,271 @@ +"""insert_arrow and insert_dataframe with timestamp and date columns on the JVM wheel: a column crosses as one array of epoch ticks per batch, +not as one JPype call per cell, and what is stored is unchanged. + +``_column_values`` turned every timestamp or date column into a Python list of ``LocalDateTime`` or ``LocalDate``, one +``ofEpochSecond`` or ``ofEpochDay`` call per cell, and ``_column_to_java`` then converted every element again with +``convert_python_to_java``: a million-row timestamp column was two million JPype calls. Now the column is the integer epoch values in ONE +``long[]`` (plus one ``boolean[]`` when a row is null) wrapped in the bridge's ``TemporalColumn``, built with one call per column and batch, and +``DocumentBatcher`` builds each ``LocalDateTime`` or ``LocalDate`` as it fills the document of that row. + +Two tests fail on the code before (they count the per-cell ``java.time`` calls, the per-element conversions and the columns that reach the bridge +as a ``TemporalColumn``); the others pin what the old route stored, so the new one must store the same: the bridge column against the per-cell +reading it replaced (every unit, negative ticks, nulls) and the refusal of a bad unit and of a timestamp past the years of ``java.time`` +before a row is written. The round trips of every Arrow temporal type through a real load, with the stored wall clock to the nanosecond, +are in ``test_dataframe_ingest.py::TestTemporalColumns``, which runs on both wheels. The JVM build only: it counts JPype calls and builds +bridge objects, so conftest.py does not collect this file on the native wheel. +""" + +from collections import Counter + +import jpype +import numpy as np +import pandas as pd +import pyarrow as pa +import pytest + +from arcadedb import core, type_conversion +from arcadedb.jvm import start_jvm + +TEMPORAL_COLUMN = "com.arcadedb.python.TemporalColumn" +UNITS = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} +TEMPORAL_COLUMNS = 7 # the five timestamp columns and the two date columns of _table() + + +@pytest.fixture(scope="module", autouse=True) +def _jvm(): + """The tests that build a bridge object without a database need the JVM too (a no-op when it already runs).""" + start_jvm() + + +def _column_class(): + return jpype.JClass(TEMPORAL_COLUMN) + + +def _table(n, start=0): + """n rows: a timestamp column of every unit, one with a zone, a date32 and a date64, a null in every seventh row.""" + idx = np.arange(start, start + n, dtype=np.int64) + null = idx % 7 == 3 + stamp = idx * 1_000 - 5_000_000 + columns = {"idx": pa.array(idx)} + for unit in UNITS: + columns[f"t_{unit}"] = pa.array(stamp, mask=null).cast(pa.timestamp(unit)) + columns["zoned"] = pa.array(stamp, mask=null).cast( + pa.timestamp("us", tz="Asia/Karachi") + ) + columns["d32"] = pa.array(idx - 300, mask=null).cast(pa.int32()).cast(pa.date32()) + columns["d64"] = pa.array((idx - 300) * 86_400_000, mask=null).cast(pa.date64()) + return pa.table(columns) + + +def _make(db, name="Stamp"): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.idx LONG") + + +def _count(db, name): + sql = f"SELECT count(*) AS c FROM {name}" # nosec B608 - test-owned name + return db.query("sql", sql).first().get("c") + + +@pytest.fixture +def crossings(monkeypatch): + """What one load does on the JPype side: the ``java.time`` calls Python makes, the per-value conversions, and the type and length of every column handed to the bridge.""" + # the type conversion caches the classes it fetched first: fetch them now, so the counting proxy below never lands in that cache + type_conversion._get_java_time_types() + type_conversion._get_java_python_types() + log = {"java_time": [], "convert": [], "columns": []} + real_convert = core.convert_python_to_java + real_column = core._column_to_java + real_jclass = jpype.JClass + + class _Counting: + def __init__(self, name): + self._name = name + self._target = real_jclass(name) + + def __getattr__(self, attr): + member = getattr(self._target, attr) + if not callable(member): + return member + + def call(*args): + log["java_time"].append(f"{self._name}.{attr}") + return member(*args) + + return call + + def convert(value): + log["convert"].append(type(value).__name__) + return real_convert(value) + + def column_to_java(name, values): + array, n = real_column(name, values) + log["columns"].append((type(array).__name__.rsplit(".", 1)[-1], n)) + return array, n + + def jclass(name, *args, **kwargs): + if isinstance(name, str) and name.startswith("java.time."): + return _Counting(name) + return real_jclass(name, *args, **kwargs) + + monkeypatch.setattr(core, "convert_python_to_java", convert) + monkeypatch.setattr(core, "_column_to_java", column_to_java) + monkeypatch.setattr(jpype, "JClass", jclass) + return log + + +def _temporal_columns(log): + """The lengths of the columns that reached the bridge as a TemporalColumn.""" + return Counter(n for kind, n in log["columns"] if kind == "TemporalColumn") + + +# ---- the tests that fail on the code before the fix: they count what the load does, never how long it takes + + +def test_a_temporal_column_makes_no_java_time_call_and_no_conversion_per_cell( + temp_db, crossings +): + """Before: one ``ofEpochSecond`` or ``ofEpochDay`` and one ``convert_python_to_java`` per cell. After: neither, for any of the seven columns.""" + n = 600 + _make(temp_db) + assert temp_db.insert_arrow("Stamp", _table(n), batch_size=n) == n + assert crossings["java_time"] == [] + assert crossings["convert"] == [] + assert _temporal_columns(crossings) == {n: TEMPORAL_COLUMNS} + + +def test_each_temporal_column_crosses_as_one_temporal_column_per_batch( + temp_db, crossings +): + """1,300 rows in batches of 500 (500, 500, 300): each of the seven columns is ONE bridge column of the batch's rows, three times.""" + _make(temp_db) + assert temp_db.insert_arrow("Stamp", _table(1_300), batch_size=500) == 1_300 + assert _temporal_columns(crossings) == { + 500: 2 * TEMPORAL_COLUMNS, + 300: TEMPORAL_COLUMNS, + } + + +def test_a_dataframe_and_a_parallel_load_take_the_same_route(temp_db, crossings): + """``insert_dataframe`` reads the frame as an Arrow table and ``parallel=True`` fills the same columns: no per-cell call on either.""" + _make(temp_db) + frame = _table(400).to_pandas() + assert isinstance(frame, pd.DataFrame) + assert temp_db.insert_dataframe("Stamp", frame, batch_size=200) == 400 + _make(temp_db, "Par") + assert ( + temp_db.insert_arrow("Par", _table(400), batch_size=200, parallel=True) == 400 + ) + assert crossings["java_time"] == [] and crossings["convert"] == [] + assert _temporal_columns(crossings) == {200: 4 * TEMPORAL_COLUMNS} + + +# ---- what the bridge column holds + + +def _reference(ticks, per_second): + """The per-cell reading the old route made: ``LocalDateTime.ofEpochSecond`` of the floor quotient, the remainder in nanoseconds.""" + local_datetime = jpype.JClass("java.time.LocalDateTime") + utc = jpype.JClass("java.time.ZoneOffset").UTC + seconds, rest = np.divmod(np.asarray(ticks, dtype=np.int64), per_second) + nanos = rest * (1_000_000_000 // per_second) + return [ + local_datetime.ofEpochSecond(int(s), int(n), utc) + for s, n in zip(seconds, nanos) + ] + + +@pytest.mark.parametrize("unit", list(UNITS)) +def test_a_date_time_column_reads_as_the_per_cell_reading_did(unit): + per_second = UNITS[unit] + limit = min(100_000_000_000 * per_second, 2**62) # about the year 5000 + ticks = np.random.default_rng(7).integers(-limit, limit, size=2_000) + ticks[:4] = [ + -1, + 0, + 1, + -per_second, + ] # one tick before the epoch is a floor, not a truncation + nulls = np.zeros(len(ticks), dtype=np.bool_) + nulls[5::11] = True + column = _column_class().dateTimes( + jpype.JArray(jpype.JLong)(ticks), + per_second, + jpype.JArray(jpype.JBoolean)(nulls), + ) + assert int(column.rows()) == len(ticks) + expected = _reference(ticks, per_second) + for i, want in enumerate(expected): + got = column.get(i) + if nulls[i]: + assert got is None + else: + assert bool(got.equals(want)), (i, int(ticks[i]), str(got), str(want)) + + +def test_a_nanosecond_tick_keeps_its_nanosecond_and_a_null_flag_is_optional(): + ns = 1_577_934_245_123_456_789 # 2020-01-02 03:04:05.123456789 + column = _column_class().dateTimes( + jpype.JArray(jpype.JLong)([ns, -1]), 1_000_000_000, None + ) + assert str(column.get(0)) == "2020-01-02T03:04:05.123456789" + assert str(column.get(1)) == "1969-12-31T23:59:59.999999999" + + +@pytest.mark.parametrize( + "per_day, days", + [(1, [-1, 0, 1, 18_322, -135_140]), (86_400_000, [-86_400_000, 0, 86_400_000])], + ids=["date32", "date64"], +) +def test_a_date_column_reads_the_day(per_day, days): + column = _column_class().dates(jpype.JArray(jpype.JLong)(days), per_day, None) + local_date = jpype.JClass("java.time.LocalDate") + for i, value in enumerate(days): + assert bool(column.get(i).equals(local_date.ofEpochDay(value // per_day))) + + +def test_a_date64_value_is_milliseconds_divided_with_a_floor(): + """The numpy ``//`` the old route used: a millisecond before midnight is still the day before.""" + column = _column_class().dates( + jpype.JArray(jpype.JLong)([86_400_000 - 1, -1]), 86_400_000, None + ) + assert str(column.get(0)) == "1970-01-01" and str(column.get(1)) == "1969-12-31" + + +def test_the_bridge_refuses_what_it_cannot_read(): + longs = jpype.JArray(jpype.JLong)([1, 2]) + for per_second in ( + 0, + -1, + 7, + 3_000_000_000, + ): # none divides a second into whole nanoseconds + with pytest.raises(jpype.JException): + _column_class().dateTimes(longs, per_second, None) + with pytest.raises(jpype.JException): + _column_class().dates(longs, 0, None) + with pytest.raises(jpype.JException): # one null flag for two rows + _column_class().dateTimes(longs, 1_000, jpype.JArray(jpype.JBoolean)([False])) + + +def test_column_to_java_hands_a_temporal_column_over_as_it_is(): + column = _column_class().dates(jpype.JArray(jpype.JLong)([0, 1, 2]), 1, None) + array, n = core._column_to_java("day", column) + assert n == 3 and array is column + + +def test_a_timestamp_past_the_years_of_java_time_is_refused_before_a_row_is_written( + temp_db, +): + """``timestamp[s]`` holds 2**62 seconds, which no ``LocalDateTime`` does: the call raises what it raised before, with the batch unwritten.""" + _make(temp_db) + table = pa.table( + { + "idx": pa.array([0, 1], pa.int64()), + "t": pa.array([0, 2**62], pa.int64()).cast(pa.timestamp("s")), + } + ) + with pytest.raises(jpype.JException): + temp_db.insert_arrow("Stamp", table) + assert _count(temp_db, "Stamp") == 0 + assert not temp_db.is_transaction_active() diff --git a/tests/test_dataframe_ingest.py b/tests/test_dataframe_ingest.py index 449e6d70..6b6c8db9 100644 --- a/tests/test_dataframe_ingest.py +++ b/tests/test_dataframe_ingest.py @@ -212,6 +212,235 @@ def test_dates_before_the_epoch(self, temp_db): assert _one_column(temp_db, pa.timestamp("ms"), values) == values +# The engine pattern that shows a stored wall clock to the nanosecond: a read through Result.get returns a Python datetime, +# which holds microseconds and cuts the rest. +_WALL_CLOCK = "yyyy-MM-dd HH:mm:ss.SSSSSSSSS" +_TICKS_PER_SECOND = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} +# Epoch ticks of each unit: one tick before the epoch (a floor, never a truncation toward zero), the epoch, one tick after +# it, 2020-01-02 03:04:05.123456789 cut to the unit, 1900-01-01 00:00:00 plus one tick, a null, and the year 2255 (the last +# year a nanosecond column holds, so every unit stays inside the range of a DATETIME_NANOS property). +_STAMPS = { + "s": [-1, 0, 1, 1_577_934_245, -2_208_988_799, None, 9_000_000_000], + "ms": [-1, 0, 1, 1_577_934_245_123, -2_208_988_799_999, None, 9_000_000_000_000], + "us": [ + -1, + 0, + 1, + 1_577_934_245_123_456, + -2_208_988_799_999_999, + None, + 9_000_000_000_000_000, + ], + "ns": [ + -1, + 0, + 1, + 1_577_934_245_123_456_789, + -2_208_988_799_999_999_999, + None, + 9_000_000_000_000_000_000, + ], +} +_DAYS = [ + dt.date(1600, 3, 1), + dt.date(1969, 12, 31), + dt.date(1970, 1, 1), + dt.date(2020, 2, 29), + None, + dt.date(2999, 12, 31), +] + + +def _wall_clock(ticks, unit): + """The UTC wall clock of ``ticks`` of ``unit`` since the epoch as ``_WALL_CLOCK`` prints it (a None stays None).""" + if ticks is None: + return None + seconds, rest = divmod(ticks, _TICKS_PER_SECOND[unit]) + moment = dt.datetime(1970, 1, 1) + dt.timedelta(seconds=seconds) + nanos = rest * (1_000_000_000 // _TICKS_PER_SECOND[unit]) + return f"{moment.isoformat(sep=' ', timespec='seconds')}.{nanos:09d}" + + +def _timestamps(ticks, unit, tz=None): + """An Arrow timestamp column of integer ticks (None is a null), so that no Python datetime cuts a nanosecond.""" + return pa.array(ticks, pa.int64()).cast(pa.timestamp(unit, tz=tz)) + + +def _wall_clocks(db, type_name, *columns): + """Each column as the text of its stored wall clock, in the order of ``idx``: {column: [text or None, ...]}.""" + shown = ", ".join(f"{c}.format('{_WALL_CLOCK}') AS {c}" for c in columns) + sql = f"SELECT idx, {shown} FROM {type_name} ORDER BY idx" # nosec B608 - test-controlled + rows = _rows(db, sql) + return {c: [r.get(c) for r in rows] for c in columns} + + +class TestTemporalColumns: + """A timestamp or date column of a load keeps what the Arrow integers say: the unit, the zone rule, the nulls, the dates before 1970. + + The JVM wheel sends such a column to the bridge as one array of epoch ticks and builds the date-times in Java; the native + wheel lays it out as columnar buffers. The expected values come from the integers and the UTC rule, never from a + database read, so both wheels must store the same wall clock. + """ + + @pytest.mark.parametrize( + "tz", [None, "UTC", "Asia/Karachi"], ids=["naive", "utc", "zoned"] + ) + @pytest.mark.parametrize("unit", ["s", "ms", "us", "ns"]) + def test_a_timestamp_keeps_its_wall_clock_to_the_nanosecond( + self, temp_db, unit, tz + ): + """A zoned timestamp is already UTC epoch time and a naive one is the clock it shows: both read the integer.""" + ticks = _STAMPS[unit] + temp_db.command("sql", "CREATE DOCUMENT TYPE Stamp") + temp_db.command("sql", "CREATE PROPERTY Stamp.val DATETIME_NANOS") + table = pa.table( + { + "idx": pa.array(range(len(ticks)), pa.int64()), + "val": _timestamps(ticks, unit, tz), + } + ) + assert temp_db.insert_arrow("Stamp", table) == len(ticks) + assert _wall_clocks(temp_db, "Stamp", "val") == { + "val": [_wall_clock(t, unit) for t in ticks] + } + + @pytest.mark.parametrize("declared", [None, "DATE"], ids=["undeclared", "DATE"]) + @pytest.mark.parametrize( + "arrow_type", [pa.date32(), pa.date64()], ids=["date32", "date64"] + ) + def test_a_date_before_and_after_the_epoch_with_a_null( + self, temp_db, arrow_type, declared + ): + assert _one_column(temp_db, arrow_type, _DAYS, declared) == _DAYS + + @pytest.mark.parametrize( + "kind, parallel", + [("DOCUMENT", False), ("DOCUMENT", True), ("VERTEX", False)], + ids=["document", "document_parallel", "vertex"], + ) + def test_every_temporal_column_of_a_load_in_batches(self, temp_db, kind, parallel): + """23 rows in batches of 5 (the last of 3): six timestamp columns and two date columns, with their nulls in different rows.""" + n = 23 + idx = np.arange(n, dtype=np.int64) + units = { + "t_s": "s", + "t_ms": "ms", + "t_us": "us", + "t_ns": "ns", + "t_zoned": "us", + "t_null": "us", + } + ticks = { + "t_s": (idx * 86_400 - 1_000_000_000).tolist(), + "t_ms": (idx * 1_000_003 - 1_000_000_000_001).tolist(), + "t_us": (idx * 1_000_000_007 - 1_000_000_000_000_001).tolist(), + "t_ns": (idx * 1_000_000_000_011 - 1_000_000_000_000_000_001).tolist(), + "t_zoned": (idx * 777_777_777 - 5).tolist(), + "t_null": [None] * n, + } + for name, every in ( + ("t_s", 4), + ("t_ms", 5), + ("t_us", 6), + ("t_ns", 7), + ("t_zoned", 3), + ): + ticks[name] = [ + None if i % every == 2 else t for i, t in enumerate(ticks[name]) + ] + days = [ + ( + None + if i % 6 == 1 + else dt.date(2000, 1, 1) + dt.timedelta(days=i * 997 - 30_000) + ) + for i in range(n) + ] + temp_db.command("sql", f"CREATE {kind} TYPE Wide") + for name in units: + temp_db.command("sql", f"CREATE PROPERTY Wide.{name} DATETIME_NANOS") + columns = {"idx": pa.array(idx)} + for name, unit in units.items(): + tz = "Asia/Karachi" if name == "t_zoned" else None + columns[name] = _timestamps(ticks[name], unit, tz) + columns["d32"] = pa.array(days, pa.date32()) + columns["d64"] = pa.array(days, pa.date64()) + assert ( + temp_db.insert_arrow( + "Wide", pa.table(columns), batch_size=5, parallel=parallel + ) + == n + ) + assert _count(temp_db, "Wide") == n + assert _wall_clocks(temp_db, "Wide", *units) == { + name: [_wall_clock(t, unit) for t in ticks[name]] + for name, unit in units.items() + } + stored = _rows(temp_db, "SELECT idx, d32, d64 FROM Wide ORDER BY idx") + assert [r.get("d32") for r in stored] == days + assert [r.get("d64") for r in stored] == days + + def test_pandas_and_polars_columns_keep_their_unit_zone_and_nulls(self, temp_db): + """pandas datetime64[ns] with a NaT, the same instants with a zone, and Polars Datetime of two units and a Date.""" + ticks = _STAMPS["ns"] + nat = np.iinfo(np.int64).min + values = np.array([nat if t is None else t for t in ticks], dtype="int64") + naive = pd.Series(values.view("datetime64[ns]")) + pandas_frame = pd.DataFrame( + { + "idx": np.arange(len(ticks), dtype=np.int64), + "t_naive": naive, + "t_zoned": naive.dt.tz_localize("UTC").dt.tz_convert("Etc/GMT-5"), + } + ) + whole_ms = [ + dt.datetime(1969, 12, 31, 23, 59, 59, 999000), + dt.datetime(2020, 1, 2, 3, 4, 5, 123000), + None, + dt.datetime(1900, 1, 1, 0, 0, 0, 1000), + ] + micros = [ + dt.datetime(1969, 12, 31, 23, 59, 59, 999999), + dt.datetime(2020, 1, 2, 3, 4, 5, 123456), + None, + dt.datetime(1900, 1, 1, 0, 0, 0, 1), + ] + polars_frame = pl.DataFrame( + { + "idx": pl.Series(range(4), dtype=pl.Int64), + "t_ms": pl.Series(whole_ms, dtype=pl.Datetime("ms")), + "t_ns": pl.Series(micros, dtype=pl.Datetime("ns")), + "day": pl.Series([m and m.date() for m in micros], dtype=pl.Date), + } + ) + for name, columns in ( + ("FromPandas", ("t_naive", "t_zoned")), + ("FromPolars", ("t_ms", "t_ns")), + ): + temp_db.command("sql", f"CREATE DOCUMENT TYPE {name}") + for column in columns: + temp_db.command( + "sql", f"CREATE PROPERTY {name}.{column} DATETIME_NANOS" + ) + assert temp_db.insert_dataframe("FromPandas", pandas_frame) == len(ticks) + assert temp_db.insert_dataframe("FromPolars", polars_frame) == 4 + shown = [_wall_clock(t, "ns") for t in ticks] + assert _wall_clocks(temp_db, "FromPandas", "t_naive", "t_zoned") == { + "t_naive": shown, + "t_zoned": shown, # the zone is only how pandas shows the instant: it is stored as the UTC wall clock + } + got = _wall_clocks(temp_db, "FromPolars", "t_ms", "t_ns") + assert got["t_ms"] == [ + m and m.isoformat(sep=" ", timespec="milliseconds") + "000000" + for m in whole_ms + ] + assert got["t_ns"] == [ + m and m.isoformat(sep=" ", timespec="microseconds") + "000" for m in micros + ] + days = _rows(temp_db, "SELECT idx, day FROM FromPolars ORDER BY idx") + assert [r.get("day") for r in days] == [m and m.date() for m in micros] + + class TestNulls: def test_every_kind_stores_nulls_as_missing(self, temp_db): temp_db.command("sql", "CREATE DOCUMENT TYPE Nul") From 2df9d178d2a00efd30dea47347b8f64f7bbec758 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:50:26 +0900 Subject: [PATCH 2/2] 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) _temporal_to_java made one LocalDateTime.ofEpochSecond or LocalDate.ofEpochDay call per cell, and insert_columns converted every element of that list again with convert_python_to_java: a million-row timestamp column was two million JPype calls. The column is now the integer epoch values in one long[] (plus one boolean[] when a row is null) wrapped in the new bridge class TemporalColumn (dateTimes(long[] ticks, long ticksPerSecond, boolean[] nulls) and dates(long[] ticks, long ticksPerDay, boolean[] nulls)), and DocumentBatcher builds each LocalDateTime or LocalDate as it fills the document of that row, as StringColumn does for text. The reading rules are the ones of the code before: UTC wall clock, a zoned timestamp is already UTC epoch time, a naive one is the clock it shows, the unit's precision is kept (a nanosecond tick is not cut to microseconds), a tick before the epoch floors, date64 is milliseconds divided by 86,400,000 with a floor. The factory checks the smallest and the largest tick, so a value past the years of java.time still raises before a row is written. The per-cell path is retired: the wheel carries its own bridge jar, so no fallback for an older one. Not measured: the speed claim waits for a timed A/B against 26.11.1.post2. Co-Authored-By: Claude Sonnet 5.5 --- .../modules/ROOT/pages/how-to/bulk-loads.adoc | 2 + src/arcadedb/core.py | 5 +- src/arcadedb/dataframe.py | 60 ++++----- .../com/arcadedb/python/DocumentBatcher.java | 16 ++- .../com/arcadedb/python/TemporalColumn.java | 118 ++++++++++++++++++ 5 files changed, 165 insertions(+), 36 deletions(-) create mode 100644 src/java/com/arcadedb/python/TemporalColumn.java diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index 87fa82c8..b2e793cd 100644 --- a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc +++ b/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 diff --git a/src/arcadedb/core.py b/src/arcadedb/core.py index 332fd1bc..29ef6574 100644 --- a/src/arcadedb/core.py +++ b/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 ```` 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: diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index 4c7beb22..6fa3de5c 100644 --- a/src/arcadedb/dataframe.py +++ b/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": diff --git a/src/java/com/arcadedb/python/DocumentBatcher.java b/src/java/com/arcadedb/python/DocumentBatcher.java index 0edeb457..8d892a91 100644 --- a/src/java/com/arcadedb/python/DocumentBatcher.java +++ b/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); diff --git a/src/java/com/arcadedb/python/TemporalColumn.java b/src/java/com/arcadedb/python/TemporalColumn.java new file mode 100644 index 00000000..533fb3c2 --- /dev/null +++ b/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); + } + } +}