From 4ee91947e9721c682294872d907934378bee1986 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 03:00:47 +0900 Subject: [PATCH 1/2] Tests: the native columnar ingest reads the declared property types once per call, not once per chunk (refs #315) A vector column (a list of lists for an ARRAY_OF_FLOATS property) is laid out as a float32 matrix only when the property is declared, so the packer asks for the declared types, and it asked for every chunk of 1000 rows with a full read of schema:types. These tests count the schema reads per call (at the statement, on the handle's query), for insert_many, an insert_columns that falls back to rows, and an insert_arrow whose batches fall back to rows: one per call, where the old code made one per chunk. A generator that creates the type and a vector property on the first chunk and another property before the second must still see every chunk as columns, and two successive calls see a schema change made between them. Declared is unit-tested on a stub handle: it reads once, reads again after a statement ran on the handle, never reads when given its types, and asks again while the schema cannot be read. They fail on the current code (it reads once per chunk and has no statement stamp on the handle). Nothing is measured here. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_native_columnar_insert.py | 162 +++++++++++++++++++++++++++ tests/test_native_columnar_pack.py | 63 +++++++++++ 2 files changed, 225 insertions(+) diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py index e729696..9b159f0 100644 --- a/tests/test_native_columnar_insert.py +++ b/tests/test_native_columnar_insert.py @@ -59,6 +59,25 @@ def spy(monkeypatch): return Spy(monkeypatch) +@pytest.fixture +def schema_reads(monkeypatch): + """The statements that read the schema (``FROM schema:types``) on a native handle, in the order they ran: the one thing an ingest + call reads the declared property types with. Counted at the statement, so it holds however the lookup is written. + """ + from arcadedb._native import database as native_database + + owner = native_database.NativeDatabase + real, seen = owner.query, [] + + def query(self, language, command, *args): + if "schema:types" in command: + seen.append(command) + return real(self, language, command, *args) + + monkeypatch.setattr(owner, "query", query) + return seen + + def _count_sql(name): return f"SELECT count(*) AS n FROM {name}" # nosec B608 - test-owned type name, no user input @@ -515,3 +534,146 @@ def test_an_unknown_kind_and_a_wrong_table_size_are_errors(temp_db): ) assert not reply["ok"] # 5 entries are not a multiple of 6 assert _count(temp_db) == 0 + + +# ---- the declared property types are read from the schema once per call, not once per chunk (#315) +# +# A vector column (a list of lists for an ARRAY_OF_FLOATS property) is laid out as a float32 matrix only when the property is declared, so +# the packer asks for the declared types. It used to ask for every chunk of 1000 rows with a full read of the schema. + + +def _vectors(n, dim=8, seed=1): + return np.random.default_rng(seed).standard_normal((n, dim), dtype=np.float32) + + +def _bits(stored): + """A stored vector as the float32 bytes it holds.""" + return np.asarray(stored, dtype=np.float32).tobytes() + + +def test_insert_many_reads_the_schema_once_for_all_its_chunks( + temp_db, spy, schema_reads, monkeypatch +): + _types(temp_db, "Ref", "Many") + rows = _rows(_probe(3_500)) + _json_reference(temp_db, "Ref", rows, monkeypatch) + spy.calls.clear() + schema_reads.clear() + assert temp_db.insert_many("Many", rows, commit_every=1_000) == 3_500 + assert spy.count(COLP) == 4 # chunks of 1000, 1000, 1000 and 500, all columnar + assert len(schema_reads) == 1, schema_reads + assert _stored(temp_db, "Many") == _stored(temp_db, "Ref") + + +def test_insert_columns_that_falls_back_to_rows_reads_the_schema_once( + temp_db, spy, schema_reads, monkeypatch +): + _types(temp_db, "Ref", "Fall") + n, x = 2_500, _vectors(2_500) + columns = { + "id": list(range(n)), + "vec": x.tolist(), + "meta": [{"k": i} for i in range(n)], # a map cannot cross: rows, as JSON + } + _json_reference(temp_db, "Ref", _rows(columns), monkeypatch) + spy.calls.clear() + schema_reads.clear() + assert temp_db.insert_columns("Fall", columns, commit_every=1_000) == n + assert spy.count(COLP) == 0 and spy.count(ROWS) == 3 # three chunks, as JSON + # the columnar attempt looked at the vector column; the three chunks of the fallback reuse what it found + assert len(schema_reads) == 1, schema_reads + assert _stored(temp_db, "Fall") == _stored(temp_db, "Ref") + + +def test_insert_arrow_with_a_column_that_cannot_cross_reads_the_schema_once( + temp_db, spy, schema_reads, monkeypatch +): + import pyarrow as pa + + _types(temp_db, "Ref", "Arr") + n, x = 2_500, _vectors(2_500) + table = pa.table( + { + "id": pa.array(range(n), type=pa.int64()), + "vec": pa.array([v.tolist() for v in x], type=pa.list_(pa.float32())), + "meta": pa.array([{"k": i} for i in range(n)]), # a struct: rows, as JSON + } + ) + _json_reference(temp_db, "Ref", table.to_pylist(), monkeypatch) + spy.calls.clear() + schema_reads.clear() + assert temp_db.insert_arrow("Arr", table, batch_size=1_000) == n + assert spy.count(COLP) == 0 and spy.count(ROWS) == 3 # three batches, as JSON + # the declared types are read once for the call (by insert_arrow) and not again by the row fallback of each batch + assert len(schema_reads) == 1, schema_reads + assert _stored(temp_db, "Arr") == _stored(temp_db, "Ref") + + +def test_a_schema_made_and_extended_inside_the_call_is_seen_by_every_chunk( + temp_db, spy, schema_reads +): + n, first, second = 3_000, _vectors(3_000, seed=1), _vectors(3_000, seed=2) + + def rows(): + for i in range(n): + # while the first chunk is read: the type and its first vector property + if i == 0: + temp_db.command("sql", "CREATE DOCUMENT TYPE Late").close() + temp_db.command("sql", "CREATE PROPERTY Late.a ARRAY_OF_FLOATS").close() + # while the second is read: another vector property + if i == 1_000: + temp_db.command("sql", "CREATE PROPERTY Late.b ARRAY_OF_FLOATS").close() + row = {"id": i, "a": first[i].tolist()} + if i >= 1_000: + row["b"] = second[i].tolist() + yield row + + assert temp_db.insert_many("Late", rows(), commit_every=1_000) == n + # every chunk crossed as columns: the second and the third saw the property made after the first was read + assert spy.count(COLP) == 3 and spy.count(ROWS) == 0, spy.calls + # one read for the first chunk, one after the schema changed, none for the third + assert len(schema_reads) == 2, schema_reads + stored = { + r["id"]: r for r in temp_db.query("sql", "SELECT id, a, b FROM Late").to_list() + } + assert sorted(stored) == list(range(n)) + for i in range(n): + assert _bits(stored[i]["a"]) == first[i].tobytes() + if i < 1_000: + assert stored[i].get("b") is None + else: + assert _bits(stored[i]["b"]) == second[i].tobytes() + + +def test_two_calls_each_read_the_schema_and_see_a_change_made_between_them( + temp_db, spy, schema_reads +): + temp_db.command("sql", "CREATE DOCUMENT TYPE Twice") + x = _vectors(1_500) + first = [{"id": i, "v": x[i].tolist()} for i in range(750)] + second = [{"id": i, "w": x[i].tolist()} for i in range(750, 1_500)] + temp_db.insert_many("Twice", first, commit_every=1_000) + assert len(schema_reads) == 1 # v is not declared: the chunk goes as JSON + assert spy.count(COLP) == 0 and spy.count(ROWS) == 1 + temp_db.command("sql", "CREATE PROPERTY Twice.w ARRAY_OF_FLOATS") + temp_db.insert_many("Twice", second, commit_every=1_000) + assert ( + len(schema_reads) == 2 + ) # a new call reads the schema again, and w is declared + assert spy.count(COLP) == 1 and spy.count(ROWS) == 1 + stored = { + r["id"]: r for r in temp_db.query("sql", "SELECT id, v, w FROM Twice").to_list() + } + assert sorted(stored) == list(range(1_500)) + for i in range(750): + assert _bits(stored[i]["v"]) == x[i].tobytes() and stored[i].get("w") is None + for i in range(750, 1_500): + assert _bits(stored[i]["w"]) == x[i].tobytes() and stored[i].get("v") is None + + +def test_a_statement_moves_the_stamp_the_declared_types_are_kept_against(temp_db): + before = temp_db._command_count + temp_db.query("sql", "SELECT FROM schema:types").to_list() + assert temp_db._command_count == before # a read does not + temp_db.command("sql", "CREATE DOCUMENT TYPE Stamp").close() + assert temp_db._command_count > before diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py index e892566..6a42b3e 100644 --- a/tests/test_native_columnar_pack.py +++ b/tests/test_native_columnar_pack.py @@ -399,3 +399,66 @@ def test_arrow_numbers_with_nulls_and_arrow_vectors(): columnar.pack({"v": ragged}, columnar.Declared(known={"v": "ARRAY_OF_FLOATS"})) is None ) + + +# ---- Declared: the schema is read once per object, so once per ingest call (#315) + + +class _Rows: + def __init__(self, rows): + self._rows = rows + + def to_list(self): + return self._rows + + +class _Handle: + """Just enough of a database handle for ``Declared``: a ``query`` that counts the schema reads, and ``_command_count``, which a + test moves the way a statement run on the handle does.""" + + def __init__(self, properties): + self.properties = properties + self.reads = 0 + self.unreadable = False + self._command_count = 0 + + def query(self, language, command): + self.reads += 1 + if self.unreadable: + raise RuntimeError("the schema cannot be read") + entries = [{"name": n, "type": t} for n, t in self.properties.items()] + return _Rows([{"name": "T", "properties": entries, "parentTypes": []}]) + + +def test_declared_reads_the_schema_once_and_again_only_after_a_statement(): + handle = _Handle({"v": "ARRAY_OF_FLOATS"}) + declared = columnar.Declared(handle, "T") + assert handle.reads == 0 # nothing is read until a column asks + assert [declared.get("v") for _ in range(50)] == ["ARRAY_OF_FLOATS"] * 50 + assert declared.get("w") is None + assert handle.reads == 1 + handle.properties["w"] = "FLOAT" + assert declared.get("w") is None and handle.reads == 1 # no statement ran: kept + handle._command_count += 1 # a statement ran on the handle (a DDL, say) + assert declared.get("w") == "FLOAT" and handle.reads == 2 + assert declared.get("v") == "ARRAY_OF_FLOATS" and handle.reads == 2 + + +def test_declared_with_known_types_never_reads_the_schema(): + handle = _Handle({"v": "FLOAT"}) + declared = columnar.Declared(handle, "T", {"v": "ARRAY_OF_FLOATS"}) + handle._command_count += 1 + assert declared.get("v") == "ARRAY_OF_FLOATS" and handle.reads == 0 + + +def test_declared_asks_again_while_the_schema_cannot_be_read(): + handle = _Handle({"v": "FLOAT"}) + handle.unreadable = True + declared = columnar.Declared(handle, "T") + assert declared.get("v") is None and handle.reads == 1 + handle.unreadable = False + assert declared.get("v") == "FLOAT" and handle.reads == 2 + + +def test_declared_without_a_handle_or_types_answers_none(): + assert columnar.Declared().get("v") is None From a48eac4c0ac9d06f92843b80d6382f23998922f2 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 03:03:58 +0900 Subject: [PATCH 2/2] Native columnar ingest: read the declared property types once per call, not once per chunk (refs #315) A vector column (a list of lists for an ARRAY_OF_FLOATS property) is laid out as a float32 matrix only when the property is declared, and the packer found that out with a full SELECT FROM schema:types (every type and property in one reply) for every chunk of 1000 rows of insert_many, and again for every batch of the row fallback of insert_columns, insert_arrow and insert_dataframe. Now one Declared belongs to one call: insert_many and insert_columns make it (columnar.declared_for) and every chunk shares it, and insert_table passes the types it already read to the row fallback of each batch. It is dropped when the call ends, so nothing is kept across calls. A schema change made while a call runs is still seen by the next chunk. The ingest itself never changes the schema (the image takes the type with getType, and only openCypher CREATE and MERGE make types or properties implicitly), so the one way is the caller's own code running while a generator of rows is read, and that is a statement on the handle: NativeDatabase counts the statements run through command() (_command_count), and Declared reads again when the count moved since it read. A query, a transaction call or a bulk crossing does not move it, so an ordinary load reads once. A read that fails is not kept, as before a chunk asked again. _insert_chunk takes the call's Declared before the chunk, because the ingest-integrity mutation tests wrap it as (type_name, *lead, chunk). The public insert_many is unchanged. Not measured yet: the speed claim waits for a timed A/B of a candidate wheel against 26.11.1.post2. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/bulk.py | 15 +++++++----- src/arcadedb/_native/columnar.py | 36 +++++++++++++++++++++++----- src/arcadedb/_native/database.py | 40 ++++++++++++++++++++++++-------- src/arcadedb/dataframe.py | 8 ++++--- 4 files changed, 74 insertions(+), 25 deletions(-) diff --git a/src/arcadedb/_native/bulk.py b/src/arcadedb/_native/bulk.py index c16d225..bc983d6 100644 --- a/src/arcadedb/_native/bulk.py +++ b/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 diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py index d001b3d..e9ad0eb 100644 --- a/src/arcadedb/_native/columnar.py +++ b/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) diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index e0c5c20..6dda2ba 100644 --- a/src/arcadedb/_native/database.py +++ b/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 diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index 4c7beb2..dec121c 100644 --- a/src/arcadedb/dataframe.py +++ b/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, )