diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index 87fa82c..b2e793c 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 332fd1b..29ef657 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 4c7beb2..6fa3de5 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 0edeb45..8d892a9 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 0000000..533fb3c --- /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); + } + } +} diff --git a/tests/conftest.py b/tests/conftest.py index fd7402f..a58d5ed 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 0000000..ef7c6dc --- /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 449e6d7..6b6c8db 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")