Native columnar ingest: the declared property types are read once per call, not once per chunk (refs #315)

src/arcadedb/_native/bulk.py

@@ -43,12 +43,12 @@ def insert_columns_fast(
 ):
     """Insert ``columns`` (name -> column) in one crossing and return the row count, or None when a column cannot be sent this way.
 
-    ``declared``: a dict of declared property types when the caller already knows them (the DataFrame path), else they are read from the
-    schema if a column needs them.
+    ``declared``: a dict of declared property types when the caller already knows them (the DataFrame path), the call's
+    :class:`columnar.Declared` when it goes on to another entry after this one, else they are read from the schema if a column needs them.
     """
     if not _usable():
         return None
-    packed = columnar.pack(columns, columnar.Declared(db, type_name, declared))
+    packed = columnar.pack(columns, columnar.declared_for(db, type_name, declared))
     if packed is None:
         return None
     rows, cols = packed
@@ -63,7 +63,7 @@ def insert_columns_parallel(db, type_name, columns, declared=None):
     """
     if not _usable():
         return None
-    packed = columnar.pack(columns, columnar.Declared(db, type_name, declared))
+    packed = columnar.pack(columns, columnar.declared_for(db, type_name, declared))
     if packed is None:
         return None
     rows, cols = packed
@@ -80,16 +80,19 @@ def insert_columns_parallel(db, type_name, columns, declared=None):
     return db._finish_parallel(type_name, [fid], rows)
 
 
-def insert_rows_columnar(db, type_name, chunk):
+def insert_rows_columnar(db, type_name, chunk, declared=None):
     """One ``insert_many`` chunk (a list of dicts with the same keys) through the columnar entry inside the caller's transaction (no
     commit here). Returns True when it was sent, False when the chunk needs the JSON path.
+
+    ``declared``: the :class:`columnar.Declared` of the call, so that its chunks read the schema once between them (a chunk on its own
+    reads it when a column needs it).
     """
     if not _usable():
         return False
     columns = columnar.rows_to_columns(chunk)
     if columns is None:
         return False
-    packed = columnar.pack(columns, columnar.Declared(db, type_name))
+    packed = columnar.pack(columns, columnar.declared_for(db, type_name, declared))
     if packed is None:
         return False
     rows, cols = packed

src/arcadedb/_native/columnar.py

@@ -68,28 +68,52 @@ def __init__(self, name, kind, dim, array, aux=None, nulls=None, keep=()):
 
 
 class Declared:
-    """The declared property types of a type, read once and only when a column needs them (``INTEGER``, ``FLOAT``, ``ARRAY_OF_FLOATS``).
+    """The declared property types of a type, read only when a column needs them (``INTEGER``, ``FLOAT``, ``ARRAY_OF_FLOATS``).
 
-    ``known`` (a dict) is used as it is; otherwise ``db`` is asked for ``schema:types`` the first time :meth:`get` is called.
+    One ingest call makes one object (:func:`declared_for`) and every chunk of the call shares it, so the call reads ``schema:types``
+    (every type and property in one reply) once, not once per chunk. The object is dropped when the call ends: nothing is kept across
+    calls, so a schema change made between two calls is always seen.
+
+    ``known`` (a dict) is the caller's own snapshot and is used as it is. Otherwise ``db`` is asked the first time :meth:`get` is
+    called, and again when a statement has run on ``db`` since (its ``_command_count`` moved): a schema change made while the call
+    runs (the rows come from a generator that runs a ``CREATE PROPERTY``, say) is a statement on the handle, so the next chunk sees it.
+    A read that fails is not kept: the next :meth:`get` asks again.
     """
 
     def __init__(self, db=None, type_name=None, known=None):
         self._db, self._type, self._known = db, type_name, known
+        self._pinned = known is not None
+        self._stamp = None
 
     def get(self, name):
-        if self._known is None:
-            self._known = _declared_types(self._db, self._type)
+        if not self._pinned:
+            stamp = getattr(self._db, "_command_count", None)
+            if self._known is None or stamp != self._stamp:
+                known = _declared_types(self._db, self._type)
+                if known is None:
+                    self._known = None
+                    return None
+                self._known, self._stamp = known, stamp
         return self._known.get(name)
 
 
+def declared_for(db, type_name, declared=None):
+    """The :class:`Declared` of one ingest call: ``declared`` itself when it is one (a call that goes on from one entry to another
+    shares it), a dict as the caller's snapshot, None as read from the schema when a column needs it.
+    """
+    if isinstance(declared, Declared):
+        return declared
+    return Declared(db, type_name, declared)
+
+
 def _declared_types(db, type_name):
-    """``{property: TYPE}`` of ``type_name`` and its parent types (the nearest declaration wins); ``{}`` when it cannot be read."""
+    """``{property: TYPE}`` of ``type_name`` and its parent types (the nearest declaration wins); None when it cannot be read."""
     try:
         rows = {
             r["name"]: r for r in db.query("sql", "SELECT FROM schema:types").to_list()
         }
     except Exception:  # nosec B110 - the insert reports the real error
-        return {}
+        return None
     out, seen, queue = {}, set(), [type_name]
     while queue:
         current = queue.pop(0)

src/arcadedb/_native/database.py

@@ -25,6 +25,7 @@
     loads_tagged,
     row_default,
 )
+from .columnar import declared_for
 
 _LAZY_COMMAND_ROWS = os.environ.get("ARCADEDB_NATIVE_LAZY_CMD", "1") != "0"
 _BINARY_REPLY = os.environ.get("ARCADEDB_NATIVE_BIN_REPLY", "1") != "0"
@@ -708,6 +709,9 @@ def __init__(self, handle: int, owned_by_server: bool = False):
         self._vector_dimensions: dict = {}
         self._vector_id_properties: dict = {}
         self._auto_transaction = False  # the engine default; set_auto_transaction() is the only way it changes
+        # statements run through command() on this handle: what columnar.Declared keeps the declared property types against (a schema
+        # change made on this handle is one of them); a query, a transaction call or a bulk crossing does not move it
+        self._command_count = 0
 
     # -- lifecycle
 
@@ -794,6 +798,8 @@ def _exec(
         self._check_not_closed()
         eager = flags & _EAGER
         flags &= ~_EAGER
+        if flags & 1:
+            self._command_count += 1
         if args:
             refuse_surplus_positionals(
                 "command" if flags & 1 else "query", language, command, args
@@ -1081,10 +1087,20 @@ def insert_many(
         this build (the JVM build's is 10,000). Rows that are already committed stay committed if a later chunk fails, so a smaller
         ``commit_every`` keeps fewer rows of a failed load (only the transaction that was open is rolled back).
         """
+        return self._insert_many(type_name, rows, commit_every, parallel)
+
+    def _insert_many(
+        self, type_name: str, rows, commit_every: int, parallel: bool, declared=None
+    ) -> int:
+        """:meth:`insert_many`, with the call's declared property types when an entry that ran before it already has them
+        (``declared``: a ``columnar.Declared`` or a dict, see ``columnar.declared_for``). The chunks of the call share one, so the call
+        reads the schema once when a column needs the declared types, not once per chunk; it ends with the call.
+        """
         self._check_not_closed()
         if parallel:
             return self._insert_many_parallel(type_name, rows)
         sql = f"INSERT INTO `{type_name}` CONTENT :row"
+        declared = declared_for(self, type_name, declared)
         rows = iter(rows)
         was_active = self.is_transaction_active()
         every = int(commit_every)
@@ -1099,7 +1115,7 @@ def insert_many(
                 chunk = list(itertools.islice(rows, size))
                 if not chunk:
                     break
-                self._insert_chunk(type_name, sql, chunk)
+                self._insert_chunk(type_name, sql, declared, chunk)
                 n += len(chunk)
                 if not was_active and every > 0 and n % every == 0:
                     self.commit()
@@ -1116,20 +1132,22 @@ def insert_many(
                     pass
             raise
 
-    def _insert_chunk(self, type_name: str, sql: str, chunk) -> None:
-        """One chunk of ``insert_many`` inside the open transaction; see :meth:`insert_many` for the paths.
+    def _insert_chunk(self, type_name: str, sql: str, declared, chunk) -> None:
+        """One chunk of ``insert_many`` inside the open transaction; see :meth:`insert_many` for the paths. ``declared`` is the
+        ``columnar.Declared`` of the call, shared by its chunks. The chunk comes last: the ingest-integrity tests wrap this seam
+        as ``(type_name, *lead, chunk)``.
 
         A row with a value the JSON text cannot carry unchanged (NaN or Infinity, an integer beyond 64 bits, a date-time or decimal inside
         a list or map, a lone surrogate) goes through ``INSERT INTO t CONTENT :row`` on its own; the rows around it still go in bulk, and
         every row is inserted in the order given.
         """
         if lib.INSERT_ROWS:
-            if insert_rows_columnar(self, type_name, chunk):
+            if insert_rows_columnar(self, type_name, chunk, declared):
                 return
             try:
                 payload = json_bulk_dumps_tagged(chunk)
             except (TypeError, ValueError):
-                self._insert_chunk_runs(type_name, sql, chunk)
+                self._insert_chunk_runs(type_name, sql, declared, chunk)
                 return
             self._insert_rows_json(type_name, payload)
             return
@@ -1147,7 +1165,7 @@ def _insert_rows_json(self, type_name: str, payload: str) -> None:
         if not reply.get("ok"):
             raise _error(f"Failed to bulk-insert into '{type_name}'", reply)
 
-    def _insert_chunk_runs(self, type_name: str, sql: str, chunk) -> None:
+    def _insert_chunk_runs(self, type_name: str, sql: str, declared, chunk) -> None:
         """A chunk with a row the JSON text cannot carry: runs of carriable rows go in bulk, the other rows one statement each, in order."""
         ok = [bulk_row_ok(row) for row in chunk]
         n = len(chunk)
@@ -1161,7 +1179,7 @@ def _insert_chunk_runs(self, type_name: str, sql: str, chunk) -> None:
             start = end
             if not carriable:
                 self._insert_chunk_per_row(sql, run)
-            elif not insert_rows_columnar(self, type_name, run):
+            elif not insert_rows_columnar(self, type_name, run, declared):
                 self._insert_rows_json(type_name, json_bulk_dumps_tagged(run))
 
     def _insert_chunk_per_row(self, sql: str, rows) -> None:
@@ -1562,16 +1580,18 @@ def insert_columns(
         known = {n: len(v) for n, v in items if hasattr(v, "__len__")}
         if len(set(known.values())) > 1:
             raise ValueError(f"columns differ in length: {known}")
+        # one for the call: the columnar attempt and, when it gives up, every chunk of the row path read the schema once between them
+        declared = declared_for(self, type_name)
         if parallel:
             from .bulk import insert_columns_parallel
 
-            done = insert_columns_parallel(self, type_name, columns)
+            done = insert_columns_parallel(self, type_name, columns, declared)
             if done is not None:
                 return done
         else:
             from .bulk import insert_columns_fast
 
-            done = insert_columns_fast(self, type_name, columns, commit_every)
+            done = insert_columns_fast(self, type_name, columns, commit_every, declared)
             if done is not None:
                 return done
         names = [n for n, _ in items]
@@ -1582,7 +1602,7 @@ def insert_columns(
         if not data[0]:
             return 0
         rows = (dict(zip(names, vals)) for vals in zip(*data))
-        return self.insert_many(type_name, rows, commit_every, parallel)
+        return self._insert_many(type_name, rows, commit_every, parallel, declared)
 
     def insert_dataframe(
         self, type_name, df, *, batch_size: int = 10_000, parallel: bool = False

src/arcadedb/dataframe.py

@@ -619,9 +619,11 @@ def _insert_documents_native(
         else:
             # jvm-payload: not-a-crossing -- no JVM here: insert_many serialises the rows to JSON text
             lists.append(col.tolist() if hasattr(col, "tolist") else col)
-    db.insert_many(
+    # the call's declared types (read once by insert_table) go on to the row path, so it does not read the schema again per chunk
+    db._insert_many(
         type_name,
         [dict(zip(names, row)) for row in zip(*lists)],
-        commit_every=batch_size,
-        parallel=parallel,
+        batch_size,
+        parallel,
+        declared,
     )