diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index c620ea2..0af52aa 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -9,6 +9,8 @@ ** xref:how-to/concurrency.adoc[Concurrency, processes and crashes] ** xref:how-to/logs.adoc[Where the logs go] ** xref:how-to/bulk-loads.adoc[Bulk loads and commit size] +** xref:how-to/import-export.adoc[Importing and exporting data] +** xref:how-to/dataframes-and-columns.adoc[DataFrames, Arrow and columns] ** xref:how-to/vectors.adoc[Vectors] * Development ** xref:development/ci.adoc[Continuous integration] diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index ace65a4..fd38a7a 100644 --- a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc +++ b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc @@ -1,7 +1,7 @@ = Bulk loads and commit size :description: The default commit size of insert_many, graph_batch and create_vertices on the arcadedb (native) and arcadedb-jvm packages, what it means for a failed load, what to do after an interrupted graph batch, how to change it, how insert_many treats Python's cyclic garbage collector on arcadedb-jvm, which property shapes a graph batch loads in bulk on the JVM build, how to pass time-series columns to append_samples (the address path is native only), and a known limitation: a graph batch does not reuse the space of deleted edges. -`insert_many`, `graph_batch` and `GraphBatch.create_vertices` load data in transactions. The commit size is the number of rows in one transaction. It decides how much work the engine holds in memory before it commits, and how many rows stay in the database when a load fails. +`insert_many`, `graph_batch` and `GraphBatch.create_vertices` load data in transactions. The commit size is the number of rows in one transaction. It decides how much work the engine holds in memory before it commits, and how many rows stay in the database when a load fails. Data that comes as a file, a DataFrame or an Arrow table has its own pages: xref:how-to/import-export.adoc[Importing and exporting data] and xref:how-to/dataframes-and-columns.adoc[DataFrames, Arrow and columns]. == Defaults diff --git a/docs/modules/ROOT/pages/how-to/databases-and-queries.adoc b/docs/modules/ROOT/pages/how-to/databases-and-queries.adoc index 3039cfe..be580e7 100644 --- a/docs/modules/ROOT/pages/how-to/databases-and-queries.adoc +++ b/docs/modules/ROOT/pages/how-to/databases-and-queries.adoc @@ -335,4 +335,6 @@ The key is backquoted because it contains dots. The change applies to that datab == Next steps * xref:quickstart.adoc[Quickstart] +* xref:how-to/import-export.adoc[Importing and exporting data] +* xref:how-to/dataframes-and-columns.adoc[DataFrames, Arrow and columns] * xref:index.adoc[ArcadeDB Python] diff --git a/docs/modules/ROOT/pages/how-to/dataframes-and-columns.adoc b/docs/modules/ROOT/pages/how-to/dataframes-and-columns.adoc new file mode 100644 index 0000000..4e70c87 --- /dev/null +++ b/docs/modules/ROOT/pages/how-to/dataframes-and-columns.adoc @@ -0,0 +1,351 @@ += DataFrames, Arrow and columns +:description: Load a pandas or Polars frame, an Arrow table or whole columns with insert_dataframe, insert_arrow and insert_columns, what each Arrow type becomes, how nulls and NaN are stored, what a declared property accepts, how many rows are converted at a time, what a failed load leaves, and what to_columns, to_arrow, to_dataframe and to_polars return, including the difference between the two builds. + +// Sources: src/arcadedb/core.py (insert_columns, insert_dataframe, insert_arrow), dataframe.py, results.py and _native/columns.py (to_columns, to_arrow, to_dataframe, to_polars). The statements below are pinned by tests/test_dataframe_columns_behavior.py, and for the Arrow types, the declared types, the column checks, the readers and the dates also by tests/test_dataframe_ingest.py, tests/test_insert_columns.py, tests/test_resultset_arrow.py, tests/test_columnar_readers.py and tests/test_resultset.py. + +This page covers column-shaped data in and out of a database. Load it with `insert_dataframe` (a pandas or Polars frame), `insert_arrow` (an Arrow table) or `insert_columns` (numpy arrays and lists). Read a result as columns with `to_columns`, `to_arrow`, `to_dataframe` or `to_polars`. For a list of Python dicts use `insert_many`, and for a file use the importer: xref:how-to/bulk-loads.adoc[Bulk loads and commit size] and xref:how-to/import-export.adoc[Importing and exporting data]. The calls here are bindings calls and have no statement in SQL, so the main manual has nothing to say about them. The types that a stored value can have are the engine's: {arcadedb-docs-url}/reference/datatypes[Data types]. + +The libraries are optional extras. A pandas frame needs pandas and pyarrow, a Polars frame needs Polars, and the readers need numpy, pyarrow, pandas or Polars as their names say: + +[source,shell] +---- +uv add 'arcadedb[arrow,pandas]' +uv add 'arcadedb[polars]' +---- + +`arcadedb-jvm` has the same extras: `uv add 'arcadedb-jvm[arrow,pandas]'`. + +The blocks on this page run one after the other against one empty database, and a test in the repository runs them that way on both packages (`tests/test_docs_data_exchange_pages.py`). A block that prints something is followed by its output. Every block assumes that `db` is an open database. + +[source,python] +---- +import datetime as dt + +import numpy as np +import pandas as pd +import polars as pl + +from arcadedb import ArcadeDBError + + +def rows(sql): + return db.query("sql", sql).to_list() +---- + +== Choose a call + +[cols="3,3,3"] +|=== +| Your data is | Load it with | Notes + +| A pandas or Polars `DataFrame` +| `db.insert_dataframe(type_name, frame)` +| Checks every column against the declared properties before it writes a row. + +| A `pyarrow` `Table`, `RecordBatch` or `RecordBatchReader` +| `db.insert_arrow(type_name, table)` +| The same rules as a frame. Any object that speaks the Arrow PyCapsule interface also loads. + +| A dict of numpy arrays or lists, or a pandas frame +| `db.insert_columns(type_name, columns)` +| No check against declared properties: the engine converts the values. + +| A list of dicts +| `db.insert_many(type_name, rows)` +| xref:how-to/bulk-loads.adoc[Bulk loads and commit size]. +|=== + +`insert_dataframe` and `insert_arrow` load a document type or a vertex type and refuse an edge type, which is loaded with `graph_batch().new_edges()`. All three return the number of records written. A frame with no rows writes nothing and returns `0`. + +== Load a frame + +A column becomes a property of the same name. The pandas index is not loaded: call `frame.reset_index()` to make it a column. Column names must be strings and must be different from each other, and a Polars `LazyFrame` must be collected first. The type must exist. It can declare its properties or not. + +A missing value is a property that is not there. A pandas `NaN` in a float column is pandas' missing value, so it loads as missing. A Polars `NaN` is a number that Polars keeps apart from null, so it loads as a `NaN`, and only a Polars null loads as missing. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Trip") +frame = pd.DataFrame( + { + "id": np.array([1, 2, 3], dtype=np.int64), + "fare": [12.5, float("nan"), 7.25], + "city": ["Oslo", "Rome", None], + "at": pd.to_datetime( + ["2026-01-02 03:04:05", "2026-01-03 00:00:00", "2026-01-04 12:00:00"] + ), + } +) +print(db.insert_dataframe("Trip", frame)) +for row in rows("SELECT FROM Trip ORDER BY id"): + print(row["id"], row.get("fare"), row.get("city"), row["at"]) + +db.command("sql", "CREATE DOCUMENT TYPE Reading") +db.insert_dataframe( + "Reading", pl.DataFrame({"k": [0, 1, 2], "x": [1.5, float("nan"), None]}) +) +for row in rows("SELECT FROM Reading ORDER BY k"): + print(row["k"], row.get("x")) +---- + +[source,text] +---- +3 +1 12.5 Oslo 2026-01-02 03:04:05 +2 None Rome 2026-01-03 00:00:00 +3 7.25 None 2026-01-04 12:00:00 +0 1.5 +1 nan +2 None +---- + +== What an Arrow type becomes + +A frame is read as an Arrow table first, so the table below holds for a frame and for `insert_arrow`. It is the mapping for a property that the type does not declare. "Read back as" is the Python type that `Result.get` returns. A date-time is a UTC wall clock: a timestamp without a time zone is stored as the clock shows, and a timestamp with a time zone is converted to UTC first. Reading it back gives a naive `datetime`. The same rule for Python `datetime` values is on xref:how-to/values-and-records.adoc#datetimes[Values and records]. + +[cols="3,3"] +|=== +| Arrow type | Read back as + +| `int8` to `int64`, `uint8` to `uint32` +| `int` + +| `float16`, `float32`, `float64` +| `float` + +| `bool` +| `bool` + +| `string`, `large_string` +| `str` + +| `timestamp`, with or without a time zone, any unit +| `datetime` + +| `date32`, `date64` +| `date` + +| `decimal128`, `decimal256` +| `Decimal`, with every digit + +| `list`, `large_list`, `fixed_size_list` +| `list` + +| `struct` +| `dict` + +| `dictionary` (a pandas categorical, a Polars `Categorical`) +| the type of its values + +| a column of only nulls +| no property +|=== + +An Arrow type with no mapping, which is time of day, duration or map, raises `TypeError` before a row is written. A `uint64` column with a value above 2^63^ - 1 raises `ValueError`. + +== Load into a declared property + +When the type declares a property, the column is checked against it before anything is written. A column loads when the engine can convert it without guessing: + +* Integers load into `BYTE`, `SHORT`, `INTEGER`, `LONG`, `FLOAT`, `DOUBLE`, `DECIMAL` and `STRING`. +* Floats load into `FLOAT`, `DOUBLE`, `DECIMAL` and `STRING`, and never into an integer property, because that would cut the fraction. +* Booleans load into `BOOLEAN` and `STRING`, strings into `STRING` and `LINK`, timestamps into any `DATETIME` precision and `STRING`, and dates into `DATE`, any `DATETIME` precision and `STRING`. + +Any other pair raises `TypeError` that names the column, the Arrow type and the declared type, and nothing is written. A declared property of a parent type counts. A value that the declared type cannot hold, such as an integer beyond the range of `INTEGER`, passes this check and fails when it is written: the load raises `ArcadeDBError` and the batch is rolled back. A column that the type does not declare is stored as its own type. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Item") +db.command("sql", "CREATE PROPERTY Item.qty INTEGER") +db.command("sql", "CREATE PROPERTY Item.label STRING") +try: + db.insert_dataframe("Item", pd.DataFrame({"qty": [1.5, 2.5]})) +except TypeError as error: + print("refused:", "is declared INTEGER" in str(error)) +print(rows("SELECT count(*) AS n FROM Item")[0]["n"]) + +fixed = pd.DataFrame({"qty": [1.5, 2.5]}).astype("int64") +print(db.insert_dataframe("Item", fixed)) +print(db.insert_dataframe("Item", pd.DataFrame({"qty": [3], "label": [7]}))) +print([r["label"] for r in rows("SELECT label FROM Item WHERE label IS NOT NULL")]) +---- + +[source,text] +---- +refused: True +0 +2 +1 +['7'] +---- + +== Batches, memory and a failed load + +The frame becomes one Arrow table first. After that only one batch of `batch_size` rows (10,000 by default) is converted to columns at a time, written, and committed. A table that already comes in chunks keeps its chunks, and a chunk larger than `batch_size` is split. So the frame and its Arrow table are in memory together while the rows are written, one batch at a time. For a source larger than memory, read it in pieces and call the load once for each piece. + +A failure rolls back the batch in flight and leaves the batches before it committed. The message says how many rows that is. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Dup") +db.command("sql", "CREATE PROPERTY Dup.id LONG") +db.command("sql", "CREATE INDEX ON Dup (id) UNIQUE") +repeated = pd.DataFrame({"id": list(range(10)) + [3]}) +try: + db.insert_dataframe("Dup", repeated, batch_size=10) +except ArcadeDBError as error: + print(str(error).endswith("were committed")) +print(rows("SELECT count(*) AS n FROM Dup")[0]["n"]) +---- + +[source,text] +---- +True +10 +---- + +A load inside a transaction of yours joins it, as `insert_columns` and `import_documents` do: the batches are written and nothing is committed, and the message of a failure says that the earlier rows were written in your transaction, not committed. Your `rollback()` removes them. A vertex type is the exception: it loads through a graph batch, which refuses to run inside your transaction (xref:how-to/transactions.adoc#graph-batch[GraphBatch inside a transaction]). + +`parallel=True` hands the batches of a document type to the asynchronous writers. A vertex type raises `ValueError` with it. + +== Load whole columns + +`db.insert_columns(type_name, columns, commit_every=10_000, parallel=False)` takes a dict of property name to column. A column is a numpy array or any sequence of Python values. All columns have the same length. `None` in a sequence is a missing value, and a `NaN` in a numpy float column is stored as a `NaN`, not as missing. A 2-D `float32` array is a column of vectors: each row of it becomes the `ARRAY_OF_FLOATS` property of one record. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Col") +db.command("sql", "CREATE PROPERTY Col.qty INTEGER") +written = db.insert_columns( + "Col", + { + "id": np.arange(3), + "qty": np.array([10, 20, 30], dtype=np.int32), + "label": ["a", None, "c"], + "score": np.array([0.5, float("nan"), 2.5]), + }, +) +print(written) +for row in rows("SELECT FROM Col ORDER BY id"): + print(row["id"], row["qty"], row.get("label"), row["score"]) +---- + +[source,text] +---- +3 +0 10 a 0.5 +1 20 None nan +2 30 c 2.5 +---- + +Bad input is refused before anything is written: columns of different lengths, no column, a name that is not a string, a column that is one `str` or `bytes` value (which would otherwise become one row for each character), a column that is not one-dimensional (a 2-D `float32` array excepted), and a `uint64` value above 2^63^ - 1. + +A numpy array of dates, `datetime64`, raises `TypeError` that points to `insert_many`, on both packages. A list of Python `datetime` objects loads, and so does a `datetime64` column in a frame, through `insert_dataframe`, which reads it as an Arrow timestamp. + +[source,python] +---- +try: + db.insert_columns( + "Col", + { + "id": np.arange(2), + "when": np.array(["2026-01-01", "2026-01-02"], dtype="datetime64[D]"), + }, + ) +except TypeError as error: + print("refused:", "insert_many" in str(error)) +stamps = [dt.datetime(2026, 1, 2, 3, 4, 5), dt.datetime(2026, 1, 3)] +print(db.insert_columns("Col", {"id": np.array([10, 11]), "when": stamps})) +---- + +[source,text] +---- +refused: True +2 +---- + +`insert_columns` does not check a column against a declared property the way `insert_dataframe` does. The values go to the engine, which converts them: a float column into an `INTEGER` property is stored without its fraction, so `1.5` becomes `1`. Use `insert_dataframe` when you want that refused. + +`commit_every` is the transaction size. Its default is 10,000 rows on both packages. The native default of 1000 on xref:how-to/bulk-loads.adoc[Bulk loads and commit size] is the default of `insert_many` when you do not pass a value, and `insert_columns` and `insert_dataframe` always pass their own value to it (10,000 for `commit_every` and for `batch_size`), so a native `insert_columns` load commits in groups of 10,000 unless you say otherwise. A failure rolls back the transaction in flight and keeps the ones before it: 30 rows with `commit_every=10` and a duplicate key in the third group leave 20 rows. A transaction of yours is not committed, as above. + +[[reading]] +== Reading columns + +The four readers take the whole result, so they hold every row in memory. For a result that is larger than memory read the rows one by one (xref:how-to/databases-and-queries.adoc[Databases and queries]) or write it with `export_to_csv`. A result set is read once (xref:how-to/databases-and-queries.adoc#a-result-set-is-read-once[A result set is read once]), so call one reader for each query. + +The next block stores one record with every value and one record with none, then reads them with each reader. The records are the same for all of them, and the readers differ in what they do with the gap: + +* `to_columns()` returns a dict of numpy arrays, and a list for text. It follows pandas: an integer column with a gap becomes `float64` with `NaN`, a boolean column with a gap becomes a list with `None`, and a text column uses `None`. +* `to_arrow()` returns a `pyarrow.Table` that keeps the type: an integer column with a gap stays `int64`, with a null. That is the reader to use when the value must survive exactly, such as an id above 2^53^. +* `to_polars()` goes through Arrow, so it keeps the type too. +* `to_dataframe()` returns a pandas frame with the pandas types, `NaN` for a gap in a number and `NaT` in a date-time. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Rec") +with db.transaction(): + db.command("sql", "INSERT INTO Rec SET k = 'a', n = 7, ok = true, s = 'x'") + db.command("sql", "INSERT INTO Rec SET k = 'b'") +query = "SELECT k, n, ok, s FROM Rec ORDER BY k" + +columns = db.query("sql", query).to_columns() +print(columns["n"].dtype, columns["n"].tolist()) +print(columns["ok"], columns["s"]) + +table = db.query("sql", query).to_arrow() +print(table.schema.field("n").type, table.column("n").to_pylist()) + +polars_frame = db.query("sql", query).to_polars() +print(polars_frame["n"].dtype, polars_frame["n"].to_list()) + +pandas_frame = db.query("sql", query).to_dataframe() +print(pandas_frame["n"].dtype) +---- + +[source,text] +---- +float64 [7.0, nan] +[True, None] ['x', None] +int64 [7, None] +Int64 [7, None] +float64 +---- + +A column without a gap is one numpy array of one type, such as `int64` for integers and `float64` for floats. The columns of the result are the names of all rows, because a document has no fixed set of properties. `columns=["n", "other"]` reads exactly those names, and a row that lacks one reads as missing. An empty result gives `{}` from `to_columns()`. The result does not depend on `batch_size`. + +Dates differ by reader. `to_columns()` and `to_arrow()` carry milliseconds: `datetime64[ms]` and `timestamp[ms]`, and a `DATE` is its midnight. `to_dataframe()` keeps the microseconds that `Result.get` returns. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Stamp") +db.insert_many("Stamp", [{"t": dt.datetime(2026, 10, 3, 12, 34, 56, 123456)}]) +stamp_query = "SELECT t FROM Stamp" +print( + db.query("sql", stamp_query).to_columns()["t"].dtype, + db.query("sql", stamp_query).to_arrow().schema.field("t").type, +) +print(db.query("sql", stamp_query).to_dataframe()["t"].iloc[0].to_pydatetime()) +---- + +[source,text] +---- +datetime64[ms] timestamp[ms] +2026-10-03 12:34:56.123456 +---- + +A missing library does not fail the same way everywhere. `to_columns()` returns `None` when numpy is missing and `to_arrow()` returns `None` when pyarrow is missing, so that a caller can fall back to another reader. `to_polars()` and `to_dataframe()` raise `ImportError`. Check for `None` before you use the result of the first two. + +== Differences between the two builds + +The calls above, the types and the results are the same on `arcadedb` and `arcadedb-jvm`. Two things differ. The first is how the readers get the rows: + +* On `arcadedb-jvm`, `to_columns()` and `to_arrow()` read the result from the engine in batches of `batch_size` rows, as packed columns, so `batch_size` sets how many rows cross at a time. +* On `arcadedb` (native), they read every row into Python dicts first and build the columns from them. `batch_size` is accepted and not used, and the rows exist as Python objects while the columns are built. + +The second is the message of a missing pandas: the JVM package adds the `uv add` hint to the `ImportError` of `to_dataframe()`, and the native package raises the plain `ImportError`. + +== Related pages + +* xref:how-to/bulk-loads.adoc[Bulk loads and commit size], the commit size and the failed-load rules of `insert_many` and `graph_batch`. +* xref:how-to/import-export.adoc[Importing and exporting data], files in and out. +* xref:how-to/transactions.adoc[Transactions], what a transaction of yours does to a load. diff --git a/docs/modules/ROOT/pages/how-to/import-export.adoc b/docs/modules/ROOT/pages/how-to/import-export.adoc new file mode 100644 index 0000000..a1645d5 --- /dev/null +++ b/docs/modules/ROOT/pages/how-to/import-export.adoc @@ -0,0 +1,363 @@ += Importing and exporting data +:description: Which call loads a file, a list of rows or a frame, what import_documents and the SQL IMPORT DATABASE do and leave behind when a row fails, how to export a database to a JSONL archive, with the SQL EXPORT DATABASE, or a query to CSV, and what the two builds share. + +// Sources: src/arcadedb/core.py (import_documents, export_database, export_to_csv), importer.py, exporter.py, and the engine importer and exporter (CSVImporterFormat, ImportDatabaseStatement, ExportDatabaseStatement). The statements below are pinned by tests/test_import_export_behavior.py, and for the importer, the exporter and the CSV export also by tests/test_importer_api.py, tests/test_import_database.py and tests/test_exporter.py. + +This page covers the calls of the Python bindings that move data between files and a database: `import_documents`, the SQL `IMPORT DATABASE`, `export_database`, the SQL `EXPORT DATABASE` and `export_to_csv`. The importer and the exporter are the engine's own, so the file formats, the importer settings and the SQL syntax belong to the main manual: {arcadedb-docs-url}/how-to/migration/importer[Importer], {arcadedb-docs-url}/reference/sql/sql-database-admin#sql-import-database[`IMPORT DATABASE`] and {arcadedb-docs-url}/reference/sql/sql-database-admin#sql-export-database[`EXPORT DATABASE`]. + +The short version: `db.import_documents(path, document_type=...)` loads a file of documents, and when one row fails it leaves none of the rows of that file. `db.export_database(path)` writes a JSONL archive that `IMPORT DATABASE` reads back. `db.export_to_csv(query, path)` writes the rows of a query to a CSV file. + +== Which call for which data + +[cols="3,3,3"] +|=== +| Your data is | Use | Where it is explained + +| A list of Python dicts +| `db.insert_many(type_name, rows)` +| xref:how-to/bulk-loads.adoc[Bulk loads and commit size] + +| Columns: numpy arrays, lists, a pandas or Polars frame, an Arrow table +| `db.insert_columns`, `db.insert_dataframe`, `db.insert_arrow` +| xref:how-to/dataframes-and-columns.adoc[DataFrames, Arrow and columns] + +| A file on disk: CSV, or another format of the engine importer +| `db.import_documents(path, ...)` for a file of documents, or `IMPORT DATABASE` for any of them +| This page + +| A whole database that you want to copy or keep +| `db.export_database(path)`, then `IMPORT DATABASE` into another database +| This page + +| The rows of a query, for a spreadsheet or another tool +| `db.export_to_csv(query, path)` +| This page + +| The rows of a query, for pandas, Polars or numpy +| `to_dataframe()`, `to_polars()`, `to_arrow()`, `to_columns()` of the result +| xref:how-to/dataframes-and-columns.adoc#reading[Reading columns] +|=== + +For a file, the call gives the engine the file name and the settings, and the engine reads the file. The rows and frames above are Python values that you build first. The commit sizes of `insert_many` and `graph_batch` are on xref:how-to/bulk-loads.adoc[Bulk loads and commit size]; a file import has its own rules, in <>. + +The blocks on this page run one after the other against one empty database, and a test in the repository runs them that way on both packages, `arcadedb` and `arcadedb-jvm` (`tests/test_docs_data_exchange_pages.py`). A block that prints something is followed by its output. Every block assumes that `db` is an open database and that `workdir` is a `pathlib.Path` of an empty folder you may write in. + +[source,python] +---- +import arcadedb +from arcadedb import ArcadeDBError + +people = workdir / "people.csv" +people.write_text( + "name,age,city\nAlice,30,New York\nBob,25,London\nCharlie,35,Paris\n", + encoding="utf-8", +) + + +def count(type_name): + query = f"SELECT count(*) AS n FROM {type_name}" + return db.query("sql", query).one().get("n") +---- + +== Importing a file of documents + +`db.import_documents(source, document_type="Document", ...)` loads a file into a document type. `source` is a path or an importer URL. The call returns an `ImportResult` with `result` (`"OK"`), `operation`, the `source_url` that the engine was given, and a `statistics` dictionary that the importer fills; `result.get(name, default)` reads one statistic. + +The type does not have to exist: the importer creates it from the file when it is missing. Create it first when you want declared properties, a unique index or another bucket count, as on xref:how-to/databases-and-queries.adoc[Databases and queries]. + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Person") +result = db.import_documents(people, document_type="Person") +print(result.result, result.operation) +print(count("Person")) +---- + +[source,text] +---- +OK import documents +3 +---- + +The keyword arguments are the settings of the SQL statement under other names. An argument that you leave out is not sent, so the engine default applies. `extra_settings`, a dict, passes any other setting of the importer, such as `typeIdProperty`, and wins over the arguments when both name the same one. + +[cols="1,1"] +|=== +| Argument | Setting of the statement + +| `source` +| `documents` + +| `document_type` +| `documentType` + +| `file_type` +| `documentsFileType` + +| `delimiter` +| `documentsDelimiter` + +| `header` +| `documentsHeader` + +| `skip_entries` +| `documentsSkipEntries` + +| `properties_include` +| `documentPropertiesInclude` + +| `commit_every` +| `commitEvery` + +| `parallel` +| `parallel` + +| `wal` +| `wal` + +| `verbose_level` +| `verboseLevel` + +| `probe_only` +| `probeOnly` + +| `force_database_create` +| `forceDatabaseCreate` + +| `trim_text` +| `trimText` + +| `on_row_error` +| `onRowError` +|=== + +`probe_only=True` reads the file but writes nothing. The result is `"PROBE_ONLY"`. + +[source,python] +---- +probe = db.import_documents(people, document_type="Person", probe_only=True) +print(probe.result, count("Person")) +---- + +[source,text] +---- +PROBE_ONLY 3 +---- + +[[failed-import]] +== When a row fails + +The importer loads the rows of one file of documents in one transaction that it owns, and commits it after the last row. By default (`on_row_error="abort"`) the first row that fails stops the import with an `ArcadeDBError` that starts with "Document import failed", and the transaction is rolled back. No row of that file is left, however many rows were read before the bad one. + +A duplicate key shows the rule. The next blocks load a file whose second row repeats the unique key of the first: + +[source,python] +---- +db.command("sql", "CREATE DOCUMENT TYPE Product") +db.command("sql", "CREATE PROPERTY Product.sku STRING") +db.command("sql", "CREATE INDEX ON Product (sku) UNIQUE") +products = workdir / "products.csv" +products.write_text( + "sku,name\nA-1,first\nA-1,duplicate\nB-2,third\n", encoding="utf-8" +) + + +def skus(): + rows = db.query("sql", "SELECT sku FROM Product").to_list() + return sorted(row["sku"] for row in rows) + + +try: + db.import_documents(products, document_type="Product") +except ArcadeDBError as error: + print("failed:", str(error).startswith("Document import failed")) +print(skus(), db.is_transaction_active()) +---- + +[source,text] +---- +failed: True +[] False +---- + +`on_row_error="skip"` changes the rule. Each row commits on its own, a row that fails is dropped, and the import goes on. The rows that loaded stay, and the first of two rows with one key wins. Any other value than `"abort"` or `"skip"` raises `ValueError` before the import starts, because the engine would read it as `"abort"` without a word. + +[source,python] +---- +db.import_documents(products, document_type="Product", on_row_error="skip") +print(skus()) +---- + +[source,text] +---- +['A-1', 'B-2'] +---- + +Skip mode commits for every row, so it needs the transaction for itself: with a transaction of yours open, it raises `ArcadeDBError` and loads nothing. + +The default mode does not touch a transaction that you opened. The rows join it, the import commits nothing, and your `commit()` or `rollback()` decides whether they stay. This is the same rule as for `insert_dataframe` and `insert_columns`, and different from a graph batch, which refuses to run in a transaction: xref:how-to/transactions.adoc#graph-batch[GraphBatch inside a transaction]. + +[source,python] +---- +db.begin() +db.import_documents(people, document_type="Person") +print(db.is_transaction_active(), count("Person")) +db.rollback() +print(count("Person")) +---- + +[source,text] +---- +True 6 +3 +---- + +NOTE: This section is about one file of documents. A file of vertices and edges follows other rules. No test of this repository covers a failed graph import, so read the main manual before you rely on what one leaves. + +== The SQL statement + +`IMPORT DATABASE` runs the same importer from SQL. `IMPORT DATABASE ` reads one file, and `IMPORT DATABASE WITH name = value, ...` takes the settings by name. A file of documents loads into the type `Document` unless `documentType` says another. The statement returns one row with `result` and `operation` (`"import database"`). + +The URL is a `file://` URL: `path.resolve().as_uri()` makes one. `as_uri()` writes a space as `%20`, and the engine opens the text after `file://` as a plain path without decoding it. For a folder with a space or another special character in its name, use `import_documents`, which builds the URL without the escaping. + +[source,python] +---- +url = people.resolve().as_uri() +row = db.command("sql", f"IMPORT DATABASE {url}").one() +print(row.get("result"), row.get("operation")) +print(count("`Document`")) + +row = db.command( + "sql", + f"IMPORT DATABASE WITH documents = '{url}', documentType = 'Visitor', commitEvery = 2", +).one() +print(row.get("result"), count("Visitor")) +---- + +[source,text] +---- +OK import database +3 +OK 3 +---- + +A missing file and a failed row raise `ArcadeDBError`, and a failed file of documents leaves no row, as above. One failure comes back as a row instead: a setting value that the importer cannot read, such as a text for `commitEvery`, gives `result` `"FAIL"` and the message in `reason`. The Python call raises for the same mistake, so check `result` when you use the statement. + +[source,python] +---- +row = db.command( + "sql", + f"IMPORT DATABASE WITH documents = '{url}', documentType = 'Visitor', commitEvery = 'many'", +).one() +print(row.get("result"), bool(row.get("reason"))) +---- + +[source,text] +---- +FAIL True +---- + +== Exporting a database + +`db.export_database(file_path, format="jsonl", overwrite=False, include_types=None, exclude_types=None, verbose=1)` writes a JSONL archive and returns a dictionary of statistics: `totalRecords`, and a count for each kind of record that it wrote, `vertices`, `edges` and `documents`. A kind that was not written may have no entry, so read a count with `stats.get(name, 0)`. Name the file `something.jsonl.tgz`. + +* An absolute `file_path` is used as it is. +* A relative one goes into the folder `exports` of the working directory. A name that already starts with `exports/` is not prefixed again. +* `include_types` and `exclude_types` take type names. When you give `include_types`, an edge type is exported only if you list it. + +An existing file is not replaced: the call raises `ArcadeDBError` that names `overwrite=True`. GraphML and GraphSON need the engine module `arcadedb-gremlin`, which neither package bundles, so `format="graphml"` and `format="graphson"` raise `ArcadeDBError`, and the message names `jsonl`. + +[source,python] +---- +db.command("sql", "CREATE VERTEX TYPE City") +db.command("sql", "CREATE EDGE TYPE Road") +with db.transaction(): + db.command("sql", "CREATE VERTEX City SET name = 'Oslo'") + db.command("sql", "CREATE VERTEX City SET name = 'Rome'") + db.command( + "sql", + "CREATE EDGE Road FROM (SELECT FROM City WHERE name = 'Oslo') " + "TO (SELECT FROM City WHERE name = 'Rome')", + ) + +backup = workdir / "backup.jsonl.tgz" +stats = db.export_database(str(backup), include_types=["City", "Road"]) +print(int(stats["vertices"]), int(stats["edges"]), backup.exists()) + +try: + db.export_database(str(backup)) +except ArcadeDBError as error: + print("exists:", "overwrite=True" in str(error)) + +try: + db.export_database(str(workdir / "backup.graphml.tgz"), format="graphml") +except ArcadeDBError as error: + print("graphml:", "jsonl" in str(error)) +---- + +[source,text] +---- +2 1 True +exists: True +graphml: True +---- + +A JSONL archive goes back in with `IMPORT DATABASE`. The new database gets the types, the records and the edges of what was exported. + +[source,python] +---- +with arcadedb.create_database(str(workdir / "copy_db")) as copy: + url = backup.resolve().as_uri() + print(copy.command("sql", f"IMPORT DATABASE {url}").one().get("result")) + print(copy.query("sql", "SELECT count(*) AS n FROM City").one().get("n")) + friends = copy.query( + "sql", "SELECT out('Road').name AS dest FROM City WHERE name = 'Oslo'" + ).to_list() + print(list(friends[0]["dest"])) +---- + +[source,text] +---- +OK +2 +['Rome'] +---- + +== The SQL EXPORT DATABASE + +`EXPORT DATABASE file://` does the same from SQL, with the settings in the main manual. It takes a file name and no folder: the file is always written to `exports/` below the working directory of the process, and a name that contains `..` raises `ArcadeDBError`. So does a name that contains the folder separator of the operating system: `/` on Linux and macOS, `\` on Windows. On Windows a forward slash is not refused, so do not rely on the statement to reject a folder there. The row it returns has `operation` set to `"export database"`. + +[source,sql] +---- +EXPORT DATABASE file://nightly.jsonl.tgz +---- + +An open transaction does not survive this statement. The engine rolls back every transaction of your thread before it exports, so rows that you wrote in it and did not commit are gone, and `db.is_transaction_active()` is `False` afterwards. Commit before you export, with this statement and with `db.export_database(...)`. The Python call does not share the folder rule of the statement: it accepts an absolute path. + +== Exporting a query to CSV + +`db.export_to_csv(query, file_path, language="sql", fieldnames=None)` runs a query and writes the rows to a CSV file. The header is the column names of the rows. `fieldnames` chooses the order of the columns; it cannot rename them, and it must name every column that the rows have, or the call raises `ArcadeDBError`. To rename a column, give it an alias in the query. A query without rows writes an empty file, or only the header when you pass `fieldnames`. + +[source,python] +---- +csv_path = workdir / "people_out.csv" +db.export_to_csv( + "SELECT name AS who, city AS town FROM Person ORDER BY name", + str(csv_path), + fieldnames=["who", "town"], +) +print(csv_path.read_text(encoding="utf-8").splitlines()[:2]) +---- + +[source,text] +---- +['who,town', 'Alice,New York'] +---- + +`arcadedb.exporter.export_to_csv(rows, file_path, fieldnames=None)` takes a list of dicts the same way, for data that did not come from a query. It creates the folders of `file_path` when they are missing. + +== Related pages + +* xref:how-to/bulk-loads.adoc[Bulk loads and commit size], the commit size of `insert_many` and `graph_batch`. +* xref:how-to/dataframes-and-columns.adoc[DataFrames, Arrow and columns], frames and columns in and out. +* xref:how-to/transactions.adoc[Transactions], what a transaction of yours does to the calls above. diff --git a/docs/modules/ROOT/pages/quickstart.adoc b/docs/modules/ROOT/pages/quickstart.adoc index 9e697f1..7615794 100644 --- a/docs/modules/ROOT/pages/quickstart.adoc +++ b/docs/modules/ROOT/pages/quickstart.adoc @@ -184,3 +184,5 @@ In the `{jvm-pkg-name}` wheels from 26.11.1 the server also serves the Studio we * xref:how-to/databases-and-queries.adoc[Databases and queries]: open, drop and check databases, parameters, pagination, SQLScript and openCypher. * xref:how-to/values-and-records.adoc[Values and records]: which Python types go in and what comes back, datetimes, record objects and the `Schema` API. +* xref:how-to/import-export.adoc[Importing and exporting data]: load a file, export a database or a query. +* xref:how-to/dataframes-and-columns.adoc[DataFrames, Arrow and columns]: load and read pandas, Polars, Arrow and numpy data. diff --git a/tests/test_dataframe_columns_behavior.py b/tests/test_dataframe_columns_behavior.py new file mode 100644 index 0000000..49b7896 --- /dev/null +++ b/tests/test_dataframe_columns_behavior.py @@ -0,0 +1,611 @@ +"""What loading and reading columns does, as the how-to page ``docs/modules/ROOT/pages/how-to/dataframes-and-columns.adoc`` says it +(ArcadeData/arcadedb-python#267). + +The page covers ``insert_dataframe``, ``insert_arrow``, ``insert_columns`` and the readers ``to_columns``, ``to_arrow``, ``to_dataframe`` and +``to_polars``. Each statement of it is pinned here first: how a frame becomes records (the Arrow type to the stored type, nulls, NaN), what a +declared property accepts and refuses, how many rows are converted at a time, what a failed load leaves behind, what ``insert_columns`` takes +and refuses, and what the readers return for nulls, dates, ``batch_size`` and a missing library. ``test_dataframe_ingest.py``, +``test_insert_columns.py``, ``test_resultset_arrow.py`` and ``test_columnar_readers.py`` pin the same calls in more depth; the tests here are +the ones the page quotes, so that a change to a sentence of the page has one place to fail. They use no Java object, so they run on the JVM +build and the native build alike (the two branches that differ test the backend they run on). +""" + +import datetime as dt +import decimal +import sys + +import pytest + +import arcadedb +from arcadedb import ArcadeDBError + +np = pytest.importorskip("numpy") +pd = pytest.importorskip("pandas") +pa = pytest.importorskip("pyarrow") +pl = pytest.importorskip("polars") + + +def _rows(db, sql): + return db.query("sql", sql).to_list() + + +def _count(db, type_name) -> int: + query = f"SELECT count(*) AS n FROM {type_name}" # nosec B608 - test-controlled + return int(_rows(db, query)[0]["n"]) + + +# -------------------------------------------------------------------------------------------------- loading a frame + + +def test_a_pandas_frame_loads_and_nan_is_a_missing_value(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Trip") + frame = pd.DataFrame( + { + "id": np.array([1, 2, 3], dtype=np.int64), + "fare": [12.5, float("nan"), 7.25], + "city": ["Oslo", "Rome", None], + "paid": [True, False, True], + "at": pd.to_datetime( + ["2026-01-02 03:04:05", "2026-01-03 00:00:00", "2026-01-04 12:00:00"] + ), + } + ) + assert temp_db.insert_dataframe("Trip", frame) == 3 + first = _rows(temp_db, "SELECT FROM Trip WHERE id = 1")[0] + assert first["fare"] == 12.5 and isinstance(first["fare"], float) + assert first["city"] == "Oslo" and first["paid"] is True + assert first["at"] == dt.datetime(2026, 1, 2, 3, 4, 5) + second = _rows(temp_db, "SELECT FROM Trip WHERE id = 2")[0] + assert second.get("fare") is None and second["paid"] is False + third = _rows(temp_db, "SELECT FROM Trip WHERE id = 3")[0] + assert third.get("city") is None + + +def test_a_polars_nan_stays_a_nan_and_a_polars_null_is_missing(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Pl") + frame = pl.DataFrame({"k": [0, 1, 2], "x": [1.5, float("nan"), None]}) + assert temp_db.insert_dataframe("Pl", frame) == 3 + nan_row = _rows(temp_db, "SELECT FROM Pl WHERE k = 1")[0] + assert nan_row["x"] != nan_row["x"] + assert _rows(temp_db, "SELECT FROM Pl WHERE k = 2")[0].get("x") is None + + +def test_the_pandas_index_is_not_loaded_until_it_is_a_column(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Indexed") + frame = pd.DataFrame({"v": [1, 2]}, index=pd.Index(["x", "y"], name="name")) + temp_db.insert_dataframe("Indexed", frame) + assert _count(temp_db, "Indexed") == 2 + assert all(r.get("name") is None for r in _rows(temp_db, "SELECT FROM Indexed")) + temp_db.insert_dataframe("Indexed", frame.reset_index()) + assert _count(temp_db, "Indexed WHERE name = 'y'") == 1 + + +@pytest.mark.parametrize( + "column, expected", + [ + (pa.array([1, 2], pa.int32()), [1, 2]), + (pa.array([0.5, 1.5], pa.float32()), [0.5, 1.5]), + (pa.array([True, False], pa.bool_()), [True, False]), + (pa.array(["a", "b"], pa.string()), ["a", "b"]), + ( + pa.array([dt.datetime(2026, 1, 2, 3, 4, 5)] * 2, pa.timestamp("us")), + [dt.datetime(2026, 1, 2, 3, 4, 5)] * 2, + ), + ( + pa.array( + [ + dt.datetime( + 2026, 1, 2, 3, 4, 5, tzinfo=dt.timezone(dt.timedelta(hours=2)) + ) + ] + * 2, + pa.timestamp("us", tz="UTC"), + ), + [dt.datetime(2026, 1, 2, 1, 4, 5)] * 2, + ), + (pa.array([dt.date(2026, 1, 2)] * 2, pa.date32()), [dt.date(2026, 1, 2)] * 2), + ( + pa.array([decimal.Decimal("1.25")] * 2, pa.decimal128(5, 2)), + [decimal.Decimal("1.25")] * 2, + ), + (pa.array([[1, 2], [3]], pa.list_(pa.int64())), [[1, 2], [3]]), + ( + pa.array([{"a": 1}, {"a": 2}], pa.struct([("a", pa.int64())])), + [{"a": 1}, {"a": 2}], + ), + ], +) +def test_the_arrow_types_of_the_mapping_table_read_back(temp_db, column, expected): + temp_db.command("sql", "CREATE DOCUMENT TYPE Mapped") + table = pa.table({"i": pa.array([0, 1], pa.int64()), "v": column}) + assert temp_db.insert_arrow("Mapped", table) == 2 + got = [r.get("v") for r in _rows(temp_db, "SELECT FROM Mapped ORDER BY i")] + assert got == expected + + +def test_the_declared_type_table_of_the_page(): + from arcadedb.dataframe import _ALLOWED + + integers = { + "BYTE", + "SHORT", + "INTEGER", + "LONG", + "FLOAT", + "DOUBLE", + "DECIMAL", + "STRING", + } + datetimes = {"DATETIME", "DATETIME_SECOND", "DATETIME_MICROS", "DATETIME_NANOS"} + assert _ALLOWED["integer"] == integers + assert _ALLOWED["floating"] == {"FLOAT", "DOUBLE", "DECIMAL", "STRING"} + assert "INTEGER" not in _ALLOWED["floating"] # a float would lose its fraction + assert _ALLOWED["bool"] == {"BOOLEAN", "STRING"} + assert _ALLOWED["string"] == {"STRING", "LINK"} + assert _ALLOWED["timestamp"] == datetimes | {"STRING"} + assert _ALLOWED["date"] == {"DATE"} | datetimes | {"STRING"} + + +def test_a_uint64_value_past_the_signed_range_is_refused(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Big") + with pytest.raises(ValueError): + temp_db.insert_arrow("Big", pa.table({"v": pa.array([2**63], pa.uint64())})) + assert _count(temp_db, "Big") == 0 + + +def test_a_type_with_no_mapping_is_refused_before_a_row_is_written(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE NoMap") + with pytest.raises(TypeError, match="no ArcadeDB mapping"): + temp_db.insert_arrow("NoMap", pa.table({"t": pa.array([1], pa.time64("us"))})) + assert _count(temp_db, "NoMap") == 0 + + +def test_arrow_table_record_batch_and_reader_all_load(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Shapes") + table = pa.table({"v": [1, 2, 3]}) + assert temp_db.insert_arrow("Shapes", table) == 3 + assert temp_db.insert_arrow("Shapes", table.to_batches()[0]) == 3 + reader = pa.RecordBatchReader.from_batches(table.schema, table.to_batches()) + assert temp_db.insert_arrow("Shapes", reader) == 3 + assert _count(temp_db, "Shapes") == 9 + + +# -------------------------------------------------------------------------------------------------- declared types + + +def test_a_declared_property_converts_what_it_can_and_refuses_the_rest(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Typed") + temp_db.command("sql", "CREATE PROPERTY Typed.qty INTEGER") + temp_db.command("sql", "CREATE PROPERTY Typed.label STRING") + ok = pd.DataFrame( + {"qty": np.array([1, 2], dtype=np.int64), "label": np.array([7, 8])} + ) + assert temp_db.insert_dataframe("Typed", ok) == 2 + row = _rows(temp_db, "SELECT FROM Typed WHERE qty = 1")[0] + assert row["qty"] == 1 and row["label"] == "7" # an integer column into STRING + + with pytest.raises(TypeError, match=r"Typed\.qty. is declared INTEGER"): + temp_db.insert_dataframe("Typed", pd.DataFrame({"qty": [1.5, 2.5]})) + assert _count(temp_db, "Typed") == 2 # nothing of the refused frame was written + + fixed = pd.DataFrame({"qty": [1.5, 2.5]}).astype("int64") + assert temp_db.insert_dataframe("Typed", fixed) == 2 + + +def test_a_value_the_declared_type_cannot_hold_fails_the_load_and_rolls_it_back( + temp_db, +): + temp_db.command("sql", "CREATE DOCUMENT TYPE Small") + temp_db.command("sql", "CREATE PROPERTY Small.v INTEGER") + with pytest.raises(ArcadeDBError): + temp_db.insert_dataframe("Small", pd.DataFrame({"v": [1, 2**40]})) + assert _count(temp_db, "Small") == 0 + + +def test_an_undeclared_column_is_stored_as_its_own_type(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Loose") + temp_db.command("sql", "CREATE PROPERTY Loose.a INTEGER") + temp_db.insert_dataframe("Loose", pd.DataFrame({"a": [1], "b": [2.5]})) + row = _rows(temp_db, "SELECT FROM Loose")[0] + assert row["a"] == 1 and row["b"] == 2.5 and isinstance(row["b"], float) + + +def test_vertex_types_get_vertices_and_edge_types_are_refused(temp_db): + temp_db.command("sql", "CREATE VERTEX TYPE City") + temp_db.command("sql", "CREATE EDGE TYPE Road") + assert temp_db.insert_dataframe("City", pd.DataFrame({"name": ["a", "b"]})) == 2 + assert _count(temp_db, "City") == 2 + with pytest.raises(ArcadeDBError, match="edge type"): + temp_db.insert_dataframe("Road", pd.DataFrame({"a": [1]})) + with pytest.raises(ArcadeDBError, match="does not exist"): + temp_db.insert_dataframe("Nope", pd.DataFrame({"a": [1]})) + with pytest.raises(ValueError, match="vertex type"): + temp_db.insert_dataframe("City", pd.DataFrame({"a": [1]}), parallel=True) + + +# -------------------------------------------------------------------------------------------------- batches and failures + + +def test_rows_are_converted_one_batch_at_a_time(temp_db, monkeypatch): + from arcadedb import dataframe + + temp_db.command("sql", "CREATE DOCUMENT TYPE Batched") + sizes = [] + events = [] + real = dataframe._column_values + real_table = dataframe.to_arrow_table + + def table_spy(data, **kwargs): + table = real_table(data, **kwargs) + events.append(f"table:{table.num_rows}") + return table + + monkeypatch.setattr(dataframe, "to_arrow_table", table_spy) + + def spy(name, chunk, *args, **kwargs): + if name == "v": + sizes.append(len(chunk)) + events.append(f"batch:{len(chunk)}") + return real(name, chunk, *args, **kwargs) + + monkeypatch.setattr(dataframe, "_column_values", spy) + assert ( + temp_db.insert_dataframe( + "Batched", pd.DataFrame({"v": range(10)}), batch_size=4 + ) + == 10 + ) + assert sizes == [4, 4, 2] + # the whole frame is one Arrow table before the first batch is converted + assert events == ["table:10", "batch:4", "batch:4", "batch:2"] + + # a table that already comes in chunks keeps them: a chunk is never merged with the next one + sizes.clear() + chunked = pa.concat_tables([pa.table({"v": [0, 1, 2]}), pa.table({"v": [3, 4]})]) + assert temp_db.insert_arrow("Batched", chunked, batch_size=100) == 5 + assert sizes == [3, 2] + + +def _unique_id_type(db, name): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id LONG") + db.command("sql", f"CREATE INDEX ON {name} (id) UNIQUE") + + +def test_a_failed_batch_is_rolled_back_and_the_earlier_ones_stay(temp_db): + _unique_id_type(temp_db, "Dup") + frame = pd.DataFrame({"id": list(range(10)) + [3]}) # the 11th row repeats id 3 + with pytest.raises(ArcadeDBError) as error: + temp_db.insert_dataframe("Dup", frame, batch_size=10) + assert "10 row(s) of the batches before this one were committed" in str(error.value) + assert _count(temp_db, "Dup") == 10 + + +def test_a_load_inside_your_transaction_joins_it(temp_db): + _unique_id_type(temp_db, "Mine") + temp_db.begin() + try: + assert ( + temp_db.insert_dataframe( + "Mine", pd.DataFrame({"id": [1, 2, 3]}), batch_size=2 + ) + == 3 + ) + assert temp_db.is_transaction_active() is True + assert _count(temp_db, "Mine") == 3 + finally: + temp_db.rollback() + assert _count(temp_db, "Mine") == 0 + + +def test_parallel_loads_documents_through_the_async_writers(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Par") + assert ( + temp_db.insert_dataframe( + "Par", pd.DataFrame({"id": range(500)}), batch_size=200, parallel=True + ) + == 500 + ) + assert _count(temp_db, "Par") == 500 + + +# -------------------------------------------------------------------------------------------------- insert_columns + + +def _columns_type(db, name="Col"): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id LONG") + db.command("sql", f"CREATE PROPERTY {name}.qty INTEGER") + db.command("sql", f"CREATE PROPERTY {name}.label STRING") + + +def test_insert_columns_takes_arrays_lists_and_none(temp_db): + _columns_type(temp_db) + written = temp_db.insert_columns( + "Col", + { + "id": np.arange(3, dtype=np.int64), + "qty": np.array([10, 20, 30], dtype=np.int32), + "label": ["a", None, "c"], + "score": np.array([0.5, float("nan"), 2.5]), + }, + ) + assert written == 3 + rows = {r["id"]: r for r in _rows(temp_db, "SELECT FROM Col")} + assert rows[0]["label"] == "a" and rows[0]["qty"] == 10 + assert rows[1].get("label") is None # None is a missing value + nan = rows[1]["score"] + assert nan != nan # a NaN in a numpy float column stays a NaN, it is not a null + + +def test_insert_columns_refuses_datetime64_and_takes_datetime_objects(temp_db): + _columns_type(temp_db) + with pytest.raises(TypeError, match="insert_many"): + temp_db.insert_columns( + "Col", + { + "id": np.arange(2), + "when": np.array(["2026-01-01", "2026-01-02"], dtype="datetime64[D]"), + }, + ) + assert _count(temp_db, "Col") == 0 + + stamps = [dt.datetime(2026, 1, 2, 3, 4, 5), dt.datetime(2026, 1, 3, 0, 0, 0)] + assert temp_db.insert_columns("Col", {"id": np.arange(2), "when": stamps}) == 2 + got = [r["when"] for r in _rows(temp_db, "SELECT FROM Col ORDER BY id")] + assert got == stamps + + # the same dates in a frame go through insert_dataframe, which reads them as Arrow timestamps + temp_db.command("sql", "CREATE DOCUMENT TYPE Dated") + frame = pd.DataFrame( + {"id": [0, 1], "when": pd.to_datetime(["2026-01-02", "2026-01-03"])} + ) + assert temp_db.insert_dataframe("Dated", frame) == 2 + assert [r["when"] for r in _rows(temp_db, "SELECT FROM Dated ORDER BY id")] == [ + dt.datetime(2026, 1, 2), + dt.datetime(2026, 1, 3), + ] + + +def test_insert_columns_refuses_bad_input_before_it_writes(temp_db): + _columns_type(temp_db) + with pytest.raises(ValueError, match="differ in length"): + temp_db.insert_columns("Col", {"id": np.arange(3), "qty": np.arange(2)}) + with pytest.raises(ValueError, match="at least one column"): + temp_db.insert_columns("Col", {}) + with pytest.raises(TypeError, match="not a sequence of values"): + temp_db.insert_columns("Col", {"label": "abc"}) + with pytest.raises(ValueError, match="64-bit signed"): + temp_db.insert_columns("Col", {"id": np.array([1, 2**63 + 5], dtype=np.uint64)}) + with pytest.raises(ValueError, match="one-dimensional"): + temp_db.insert_columns("Col", {"id": np.zeros((2, 2), dtype=np.int64)}) + assert _count(temp_db, "Col") == 0 + + +def test_insert_columns_does_not_check_a_declared_type_the_way_arrow_does(temp_db): + """insert_dataframe refuses a float column for an INTEGER property (see test_a_declared_property_converts_what_it_can_and_refuses_the_rest); + insert_columns hands the values to the engine, which converts them (a float to an INTEGER drops the fraction). + """ + _columns_type(temp_db) + temp_db.insert_columns("Col", {"id": np.arange(2), "qty": np.array([1.5, 2.5])}) + assert [r["qty"] for r in _rows(temp_db, "SELECT FROM Col ORDER BY id")] == [1, 2] + + +def test_insert_columns_commits_every_batch_it_finished(temp_db): + _unique_id_type(temp_db, "Chunked") + ids = np.arange(30, dtype=np.int64) + ids[25] = ids[3] # a duplicate key in the third batch of ten + with pytest.raises(ArcadeDBError): + temp_db.insert_columns("Chunked", {"id": ids}, commit_every=10) + assert ( + _count(temp_db, "Chunked") == 20 + ) # two batches committed, the third rolled back + + assert temp_db.insert_columns("Chunked", {"id": np.arange(100, 105)}) == 5 + + +def test_insert_columns_inside_your_transaction_commits_nothing(temp_db): + _columns_type(temp_db) + temp_db.begin() + try: + assert ( + temp_db.insert_columns("Col", {"id": np.arange(20)}, commit_every=5) == 20 + ) + assert temp_db.is_transaction_active() is True + finally: + temp_db.rollback() + assert _count(temp_db, "Col") == 0 + + +def test_a_float32_matrix_is_a_column_of_vectors(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Vec") + temp_db.command("sql", "CREATE PROPERTY Vec.v ARRAY_OF_FLOATS") + matrix = np.array([[0.5, 1.5], [2.5, 3.5]], dtype=np.float32) + assert temp_db.insert_columns("Vec", {"id": np.arange(2), "v": matrix}) == 2 + rows = _rows(temp_db, "SELECT FROM Vec ORDER BY id") + assert [list(r["v"]) for r in rows] == [[0.5, 1.5], [2.5, 3.5]] + + +def test_insert_columns_takes_a_pandas_frame(temp_db): + _columns_type(temp_db) + frame = pd.DataFrame({"id": [0, 1], "label": ["a", "b"]}) + assert temp_db.insert_columns("Col", frame) == 2 + assert [r["label"] for r in _rows(temp_db, "SELECT FROM Col ORDER BY id")] == [ + "a", + "b", + ] + + +# -------------------------------------------------------------------------------------------------- reading columns + + +@pytest.fixture +def readings(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Rec") + stamp = dt.datetime(2026, 10, 3, 12, 34, 56, 123456) + with temp_db.transaction(): + temp_db.command( + "sql", + "INSERT INTO Rec SET k = 'a', n = 7, ok = true, s = 'x', x = 1.5, t = :t", + {"t": stamp}, + ) + temp_db.command("sql", "INSERT INTO Rec SET k = 'b'") + return temp_db + + +QUERY = "SELECT k, n, ok, s, x, t FROM Rec ORDER BY k" + + +def test_to_columns_follows_the_pandas_conventions_for_missing_values(readings): + columns = readings.query("sql", QUERY).to_columns() + assert set(columns) == {"k", "n", "ok", "s", "x", "t"} + assert columns["k"] == ["a", "b"] + assert ( + columns["n"].dtype == np.float64 + ) # an integer column with a gap is a float column + assert columns["n"][0] == 7.0 and np.isnan(columns["n"][1]) + assert columns["ok"] == [True, None] # a boolean column with a gap is a list + assert columns["s"] == ["x", None] + assert columns["x"].dtype == np.float64 and np.isnan(columns["x"][1]) + assert columns["t"].dtype == np.dtype("datetime64[ms]") and np.isnat( + columns["t"][1] + ) + assert columns["t"][0] == np.datetime64("2026-10-03T12:34:56.123", "ms") + + +def test_to_columns_without_a_gap_is_one_typed_array(readings): + columns = readings.query("sql", "SELECT k, n FROM Rec WHERE k = 'a'").to_columns() + assert columns["n"].dtype == np.int64 and columns["n"].tolist() == [7] + + +def test_to_arrow_keeps_the_type_of_a_column_with_a_gap(readings): + table = readings.query("sql", QUERY).to_arrow() + assert table.schema.field("n").type == pa.int64() + assert table.column("n").to_pylist() == [7, None] + assert table.schema.field("ok").type == pa.bool_() + assert table.column("ok").to_pylist() == [True, None] + assert table.column("s").to_pylist() == ["x", None] + assert table.schema.field("t").type == pa.timestamp("ms") + assert table.column("t").to_pylist() == [ + dt.datetime(2026, 10, 3, 12, 34, 56, 123000), + None, + ] + + +def test_to_dataframe_and_to_polars(readings): + frame = readings.query("sql", QUERY).to_dataframe() + assert set(frame.columns) == {"k", "n", "ok", "s", "x", "t"} + assert frame["t"].iloc[0].to_pydatetime() == dt.datetime( + 2026, 10, 3, 12, 34, 56, 123456 + ) + assert pd.isna(frame["t"].iloc[1]) and pd.isna(frame["n"].iloc[1]) + assert frame["n"].dtype == np.float64 # a gap in an integer column makes it float64 + + polars_frame = readings.query("sql", QUERY).to_polars() + assert set(polars_frame.columns) == {"k", "n", "ok", "s", "x", "t"} + assert polars_frame["n"].to_list() == [7, None] # Polars keeps the integer type + assert polars_frame["n"].dtype == pl.Int64 + + +def test_columns_pins_the_names_and_a_missing_one_reads_as_missing(readings): + table = readings.query("sql", QUERY).to_arrow(columns=["n", "nope"]) + assert table.column_names == ["n", "nope"] + assert table.column("nope").to_pylist() == [None, None] + columns = readings.query("sql", QUERY).to_columns(columns=["k"]) + assert list(columns) == ["k"] + + +def test_the_columns_do_not_depend_on_batch_size(readings): + reference = readings.query("sql", QUERY).to_arrow() + for size in (1, 2, 25_000): + assert readings.query("sql", QUERY).to_arrow(batch_size=size).equals(reference) + columns = readings.query("sql", QUERY).to_columns(batch_size=size) + assert columns["k"] == ["a", "b"] and columns["n"][0] == 7.0 + + +def test_how_the_rows_reach_the_columns_differs_between_the_builds( + readings, monkeypatch +): + """The JVM build reads the result in batches of ``batch_size`` rows from the engine's column buffer. The native build reads + every row into Python dicts first (``_drain_dicts``) and builds the columns from them, so ``batch_size`` is not used. + """ + from arcadedb import results + + if arcadedb.__backend__ == "native": + resultset = type(readings.query("sql", QUERY)) + drained = [] + real = resultset._drain_dicts + + def spy(self, *args, **kwargs): + rows = real(self, *args, **kwargs) + drained.append(len(rows)) + return rows + + monkeypatch.setattr(resultset, "_drain_dicts", spy) + readings.query("sql", QUERY).to_columns(batch_size=1) + assert drained == [2] # one read of all the rows, whatever batch_size says + else: + batches = [] + real_bridge = results._bridge_class + + class Spy: + def __init__(self, inner): + self._inner = inner + + def nextColumnBatch( + self, java_rows, batch_size, *rest + ): # noqa: N802 - the Java name + batches.append(int(batch_size)) + return self._inner.nextColumnBatch(java_rows, batch_size, *rest) + + def bridge(name): + inner = real_bridge(name) + return Spy(inner) if name == "ColumnBatcher" else inner + + monkeypatch.setattr(results, "_bridge_class", bridge) + readings.query("sql", QUERY).to_columns(batch_size=1) + assert batches[:2] == [1, 1] + + +def test_an_empty_result_has_no_columns(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Nothing") + assert temp_db.query("sql", "SELECT FROM Nothing").to_columns() == {} + + +def test_a_missing_library_returns_none_or_raises_import_error(readings, monkeypatch): + # the result sets are made first: only the call under test runs without the library + for_arrow = readings.query("sql", QUERY) + for_polars = readings.query("sql", QUERY) + for_columns = readings.query("sql", QUERY) + for_frame = readings.query("sql", QUERY) + + monkeypatch.setitem(sys.modules, "pyarrow", None) + assert for_arrow.to_arrow() is None + with pytest.raises(ImportError): + for_polars.to_polars() + monkeypatch.undo() + + monkeypatch.setitem(sys.modules, "numpy", None) + assert for_columns.to_columns() is None + monkeypatch.undo() + + monkeypatch.setitem(sys.modules, "pandas", None) + with pytest.raises(ImportError, match="pandas") as error: + for_frame.to_dataframe() + # the JVM build adds the install hint to the message, the native build raises the plain ImportError + assert ("uv add" in str(error.value)) == (arcadedb.__backend__ != "native") + + +def test_a_vertex_frame_inside_your_transaction_is_refused_and_leaves_it_alone( + temp_db, +): + """The graph batch behind a vertex load refuses to run inside a transaction of yours (test_transactions_behavior.py).""" + temp_db.command("sql", "CREATE VERTEX TYPE Town") + temp_db.begin() + try: + with pytest.raises(ArcadeDBError): + temp_db.insert_dataframe("Town", pd.DataFrame({"name": ["a", "b"]})) + assert temp_db.is_transaction_active() is True + finally: + temp_db.rollback() + assert _count(temp_db, "Town") == 0 diff --git a/tests/test_docs_data_exchange_pages.py b/tests/test_docs_data_exchange_pages.py new file mode 100644 index 0000000..0c9ba53 --- /dev/null +++ b/tests/test_docs_data_exchange_pages.py @@ -0,0 +1,87 @@ +"""Every Python block of the two Antora pages ``how-to/import-export.adoc`` and ``how-to/dataframes-and-columns.adoc`` runs, and prints what the +page says (ArcadeData/arcadedb-python#267). + +Same pattern as ``test_docs_transactions_page.py``: the blocks of one page are chained on purpose, they run in the order of the page, in one +namespace, against one fresh temporary database that the namespace holds as ``db`` (and an empty folder as ``workdir``). A block that is followed +by a ``[source,text]`` block must print exactly that text. What each statement of the pages claims is pinned on its own in +``test_import_export_behavior.py`` and ``test_dataframe_columns_behavior.py``. The tests use no Java object, so they run on the JVM build and the +native build alike. +""" + +import contextlib +import io +import re +from pathlib import Path + +import pytest + +PAGES = ( + Path(__file__).resolve().parents[1] + / "docs" + / "modules" + / "ROOT" + / "pages" + / "how-to" +) + +# a Python listing, then (optionally, directly after it) the text listing with its output +BLOCK_RE = re.compile( + r"\[source,python[^\]\n]*\]\n----\n(?P.*?)\n----\n" + r"(?:\n\[source,text\]\n----\n(?P.*?)\n----\n)?", + re.DOTALL, +) + +# page, number of Python blocks, number of them that print +EXPECTED = { + "import-export.adoc": (11, 10), + "dataframes-and-columns.adoc": (8, 7), +} + + +def _blocks(page): + text = (PAGES / page).read_text(encoding="utf-8") + found = [(m.group("code"), m.group("out")) for m in BLOCK_RE.finditer(text)] + # no Python listing may escape the extraction (a different spelling of the block header would run nowhere) + assert text.count("[source,python") == len(found) + return found + + +@pytest.mark.parametrize("page", sorted(EXPECTED)) +def test_the_page_has_its_blocks(page): + blocks = _blocks(page) + assert len(blocks) == EXPECTED[page][0] + assert sum(1 for _, out in blocks if out is not None) == EXPECTED[page][1] + + +def _run(page, namespace): + for number, (code, expected) in enumerate(_blocks(page), start=1): + captured = io.StringIO() + compiled = compile(code, f"{page} python block {number}", "exec") + with contextlib.redirect_stdout(captured): + exec(compiled, namespace) # nosec B102 - the pages of this repository + if expected is not None: + assert ( + captured.getvalue().strip() == expected.strip() + ), f"block {number} of {page} printed other text than the page shows:\n{code}" + else: + assert ( + captured.getvalue() == "" + ), f"block {number} of {page} prints, the page shows no output" + + +def test_every_block_of_the_import_and_export_page_runs_in_order(temp_db, tmp_path): + workdir = tmp_path / "work" + workdir.mkdir() + _run( + "import-export.adoc", + {"db": temp_db, "workdir": workdir, "__name__": "import_export_page"}, + ) + + +def test_every_block_of_the_dataframe_page_runs_in_order(temp_db): + for library in ("numpy", "pandas", "polars", "pyarrow"): + pytest.importorskip(library) + _run( + "dataframes-and-columns.adoc", + {"db": temp_db, "__name__": "dataframes_page"}, + ) diff --git a/tests/test_import_export_behavior.py b/tests/test_import_export_behavior.py new file mode 100644 index 0000000..e597aa0 --- /dev/null +++ b/tests/test_import_export_behavior.py @@ -0,0 +1,515 @@ +"""What importing and exporting does, as the how-to page ``docs/modules/ROOT/pages/how-to/import-export.adoc`` says it (ArcadeData/arcadedb-python#267). + +Each statement of that page is pinned here first: the result of ``import_documents``, ``probe_only``, what is left in the database when an +import fails (the default ``abort`` and ``on_row_error="skip"``), an import inside the caller's transaction, the SQL ``IMPORT DATABASE`` of the +same importer (and the one failure it reports as a row instead of raising), ``export_database`` (where the file lands, the filters, +``overwrite``, the formats), the SQL ``EXPORT DATABASE`` and ``export_to_csv``. The reading of the engine source behind the transaction rules is +named next to each test. The tests use SQL and the public calls only, so they run on the JVM build and the native build alike. +""" + +import csv +import os +import shutil +import uuid +from pathlib import Path + +import pytest + +import arcadedb +from arcadedb import ArcadeDBError + + +def _uri(path) -> str: + return Path(path).resolve().as_uri() + + +def _count(db, type_name) -> int: + query = f"SELECT count(*) AS n FROM {type_name}" # nosec B608 - test-controlled + return int(db.query("sql", query).to_list()[0]["n"]) + + +def _people_csv(directory) -> Path: + path = Path(directory) / "people.csv" + path.write_text( + "name,age,city\nAlice,30,New York\nBob,25,London\nCharlie,35,Paris\n", + encoding="utf-8", + ) + return path + + +def _duplicate_key_csv(directory) -> Path: + """Three rows, the middle one repeats the unique key of the first.""" + path = Path(directory) / "products.csv" + path.write_text("sku,name\nA-1,first\nA-1,duplicate\nB-2,third\n", encoding="utf-8") + return path + + +@pytest.fixture +def unique_product(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Product") + temp_db.command("sql", "CREATE PROPERTY Product.sku STRING") + temp_db.command("sql", "CREATE INDEX ON Product (sku) UNIQUE") + return temp_db + + +# -------------------------------------------------------------------------------------------------- import_documents + + +def test_import_documents_returns_an_import_result(temp_db, tmp_path): + csv_path = _people_csv(tmp_path) + result = temp_db.import_documents(csv_path, "Person") + assert result.result == "OK" + assert result.operation == "import documents" + assert result.source_url.startswith("file:") + assert isinstance(result.statistics, dict) + assert result.get("no such statistic", 7) == 7 + assert {"result", "operation", "source_url"} <= set(result.to_dict()) + + assert _count(temp_db, "Person") == 3 + row = temp_db.query("sql", "SELECT FROM Person WHERE name = 'Alice'").one() + assert row.get("city") == "New York" + + +def test_import_documents_creates_a_missing_type(temp_db, tmp_path): + csv_path = _people_csv(tmp_path) + temp_db.import_documents(csv_path, document_type="Person") + assert _count(temp_db, "Person") == 3 + + +def test_probe_only_reads_the_file_and_writes_nothing(temp_db, tmp_path): + temp_db.command("sql", "CREATE DOCUMENT TYPE Probe") + result = temp_db.import_documents( + _people_csv(tmp_path), document_type="Probe", probe_only=True + ) + assert result.result == "PROBE_ONLY" + assert _count(temp_db, "Probe") == 0 + + +def test_the_keyword_arguments_are_the_importer_settings_of_the_sql_statement(): + from arcadedb.importer import _build_import_settings + + settings = _build_import_settings( + "file:///x.csv", + document_type="Person", + file_type="csv", + delimiter=";", + header="a;b", + skip_entries=1, + properties_include="a,b", + commit_every=100, + parallel=2, + wal=False, + verbose_level=0, + probe_only=True, + force_database_create=False, + trim_text=False, + on_row_error="skip", + extra_settings={"typeIdProperty": "id", "ignored": None}, + ) + assert settings == { + "documents": "file:///x.csv", + "documentType": "Person", + "documentsFileType": "csv", + "documentsDelimiter": ";", + "documentsHeader": "a;b", + "documentsSkipEntries": 1, + "documentPropertiesInclude": "a,b", + "commitEvery": 100, + "parallel": 2, + "wal": False, + "verboseLevel": 0, + "probeOnly": True, + "forceDatabaseCreate": False, + "trimText": False, + "onRowError": "skip", + "typeIdProperty": "id", + } + # an argument left at None is not sent at all, so the engine default applies + assert _build_import_settings( + "file:///x.csv", + document_type="D", + file_type=None, + delimiter=None, + header=None, + skip_entries=None, + properties_include=None, + commit_every=None, + parallel=None, + wal=None, + verbose_level=None, + probe_only=None, + force_database_create=None, + trim_text=None, + on_row_error=None, + extra_settings=None, + ) == {"documents": "file:///x.csv", "documentType": "D"} + + +def test_extra_settings_win_over_the_arguments(): + from arcadedb.importer import _build_import_settings + + kwargs = dict( + document_type="D", + file_type=None, + delimiter=None, + header=None, + skip_entries=None, + properties_include=None, + parallel=None, + wal=None, + verbose_level=None, + probe_only=None, + force_database_create=None, + trim_text=None, + on_row_error=None, + ) + settings = _build_import_settings( + "file:///x.csv", commit_every=100, extra_settings={"commitEvery": 5}, **kwargs + ) + assert settings["commitEvery"] == 5 + + +def test_a_setting_the_importer_cannot_read_raises(temp_db, tmp_path): + with pytest.raises(ArcadeDBError, match="Document import failed"): + temp_db.import_documents( + _people_csv(tmp_path), + document_type="Person", + extra_settings={"commitEvery": "not a number"}, + ) + with pytest.raises(ValueError, match="Invalid on_row_error mode"): + temp_db.import_documents( + _people_csv(tmp_path), document_type="Person", on_row_error="ignore" + ) + + +# -------------------------------------------------------------------------------------------------- a failed import + + +def test_a_failed_import_leaves_none_of_its_rows(unique_product, tmp_path): + """The documents of one file load in one transaction that the import owns (``CSVImporterFormat.loadDocuments``: one + ``database.begin()``, one ``commit()`` after the last row, a rollback of the owned transaction on any failure). + """ + csv_path = _duplicate_key_csv(tmp_path) + with pytest.raises(ArcadeDBError, match="Document import failed"): + unique_product.import_documents(csv_path, document_type="Product") + assert _count(unique_product, "Product") == 0 + assert unique_product.is_transaction_active() is False + + with pytest.raises(ArcadeDBError, match="Document import failed"): + unique_product.import_documents( + csv_path, document_type="Product", on_row_error="abort" + ) + assert _count(unique_product, "Product") == 0 + + +def test_skip_keeps_the_good_rows_and_drops_the_bad_one(unique_product, tmp_path): + unique_product.import_documents( + _duplicate_key_csv(tmp_path), document_type="Product", on_row_error="skip" + ) + skus = sorted( + r.get("sku") for r in unique_product.query("sql", "SELECT sku FROM Product") + ) + assert skus == ["A-1", "B-2"] + first = unique_product.query("sql", "SELECT FROM Product WHERE sku = 'A-1'").one() + assert first.get("name") == "first" # the first row won, the duplicate was dropped + + +def test_skip_needs_the_transaction_for_itself(unique_product, tmp_path): + unique_product.begin() + try: + with pytest.raises(ArcadeDBError): + unique_product.import_documents( + _duplicate_key_csv(tmp_path), + document_type="Product", + on_row_error="skip", + ) + finally: + unique_product.rollback() + assert _count(unique_product, "Product") == 0 + + +def test_an_import_inside_your_transaction_commits_nothing(temp_db, tmp_path): + """``CSVImporterFormat``: a transaction that was active on entry is not the import's, so it neither commits nor rolls it back.""" + temp_db.command("sql", "CREATE DOCUMENT TYPE Person") + temp_db.begin() + try: + temp_db.import_documents(_people_csv(tmp_path), document_type="Person") + assert temp_db.is_transaction_active() is True + assert _count(temp_db, "Person") == 3 + finally: + temp_db.rollback() + assert _count(temp_db, "Person") == 0 + + +# -------------------------------------------------------------------------------------------------- IMPORT DATABASE + + +def test_sql_import_database_is_the_same_importer(temp_db, tmp_path): + url = _uri(_people_csv(tmp_path)) + row = temp_db.command("sql", f"IMPORT DATABASE {url}").one() + assert row.get("result") == "OK" + assert row.get("operation") == "import database" + assert _count(temp_db, "`Document`") == 3 # the default type of a file of documents + + row = temp_db.command( + "sql", + f"IMPORT DATABASE WITH documents = '{url}', documentType = 'Person', " + "documentsFileType = 'csv', commitEvery = 2", + ).one() + assert row.get("result") == "OK" + assert _count(temp_db, "Person") == 3 + + +def test_sql_import_of_a_missing_file_and_of_a_bad_row_raises(unique_product, tmp_path): + with pytest.raises(ArcadeDBError): + unique_product.command( + "sql", f"IMPORT DATABASE {_uri(Path(tmp_path) / 'missing.csv')}" + ) + url = _uri(_duplicate_key_csv(tmp_path)) + with pytest.raises(ArcadeDBError): + unique_product.command( + "sql", f"IMPORT DATABASE WITH documents = '{url}', documentType = 'Product'" + ) + assert _count(unique_product, "Product") == 0 + + +def test_sql_import_reports_an_unusable_setting_as_a_row(temp_db, tmp_path): + """``ImportDatabaseStatement`` turns an ``IllegalArgumentException`` of the importer (a ``NumberFormatException`` of a setting + value is one) into a ``FAIL`` row with the reason, where the Python call raises (see test_a_setting_the_importer_cannot_read_raises). + """ + url = _uri(_people_csv(tmp_path)) + row = temp_db.command( + "sql", + f"IMPORT DATABASE WITH documents = '{url}', documentType = 'Person', " + "commitEvery = 'not a number'", + ).one() + assert row.get("result") == "FAIL" + assert row.get("reason") + + +# -------------------------------------------------------------------------------------------------- export_database + + +@pytest.fixture +def graph_db(temp_db): + temp_db.command("sql", "CREATE VERTEX TYPE Person") + temp_db.command("sql", "CREATE EDGE TYPE Knows") + temp_db.command("sql", "CREATE DOCUMENT TYPE Note") + with temp_db.transaction(): + temp_db.command("sql", "CREATE VERTEX Person SET name = 'a'") + temp_db.command("sql", "CREATE VERTEX Person SET name = 'b'") + temp_db.command( + "sql", + "CREATE EDGE Knows FROM (SELECT FROM Person WHERE name = 'a') " + "TO (SELECT FROM Person WHERE name = 'b')", + ) + temp_db.command("sql", "INSERT INTO Note SET text = 'n1'") + temp_db.command("sql", "INSERT INTO Note SET text = 'n2'") + temp_db.command("sql", "INSERT INTO Note SET text = 'n3'") + return temp_db + + +def test_export_database_writes_the_absolute_path_it_is_given(graph_db, tmp_path): + target = tmp_path / "backup.jsonl.tgz" + stats = graph_db.export_database(str(target)) + assert target.exists() and target.stat().st_size > 0 + assert int(stats["vertices"]) == 2 + assert int(stats["edges"]) == 1 + assert int(stats["documents"]) == 3 + assert int(stats["totalRecords"]) == 6 + + +def test_a_relative_name_lands_in_exports_below_the_working_directory(graph_db): + name = f"adb-test-{uuid.uuid4().hex}.jsonl.tgz" + target = Path("exports") / name + try: + graph_db.export_database(name) + assert target.exists() + # the same text with the directory in front is the same file, not exports/exports + graph_db.export_database(f"exports/{name}", overwrite=True) + assert not (Path("exports") / "exports").exists() + finally: + if target.exists(): + target.unlink() + + +def test_include_and_exclude_types(graph_db, tmp_path): + only = graph_db.export_database( + str(tmp_path / "only.jsonl.tgz"), include_types=["Person"] + ) + assert int(only["vertices"]) == 2 + assert int(only.get("edges", 0)) == 0 and int(only.get("documents", 0)) == 0 + + with_edges = graph_db.export_database( + str(tmp_path / "with_edges.jsonl.tgz"), include_types=["Person", "Knows"] + ) + assert int(with_edges["vertices"]) == 2 and int(with_edges["edges"]) == 1 + + without = graph_db.export_database( + str(tmp_path / "without.jsonl.tgz"), exclude_types=["Note"] + ) + assert int(without["vertices"]) == 2 and int(without["edges"]) == 1 + assert int(without.get("documents", 0)) == 0 + + +def test_an_existing_file_is_refused_unless_overwrite_is_true(graph_db, tmp_path): + target = str(tmp_path / "twice.jsonl.tgz") + graph_db.export_database(target) + with pytest.raises(ArcadeDBError, match="overwrite=True"): + graph_db.export_database(target) + assert int(graph_db.export_database(target, overwrite=True)["totalRecords"]) == 6 + + +@pytest.mark.parametrize("fmt", ["graphml", "graphson"]) +def test_graphml_and_graphson_are_not_bundled(graph_db, tmp_path, fmt): + with pytest.raises(ArcadeDBError, match="jsonl"): + graph_db.export_database(str(tmp_path / f"x.{fmt}.tgz"), format=fmt) + + +def test_an_unknown_format_is_refused(graph_db, tmp_path): + with pytest.raises(ArcadeDBError, match="Invalid export format"): + graph_db.export_database(str(tmp_path / "x.tgz"), format="parquet") + + +def test_the_export_imports_into_a_new_database_with_the_sql_statement( + graph_db, tmp_path +): + target = tmp_path / "copy.jsonl.tgz" + graph_db.export_database(str(target)) + with arcadedb.create_database(str(tmp_path / "copy_db")) as copy: + row = copy.command("sql", f"IMPORT DATABASE {_uri(target)}").one() + assert row.get("result") == "OK" + assert _count(copy, "Person") == 2 + assert _count(copy, "Knows") == 1 + assert _count(copy, "Note") == 3 + names = copy.query( + "sql", "SELECT out('Knows').name AS friends FROM Person WHERE name = 'a'" + ).to_list() + assert list(names[0]["friends"]) == ["b"] + + +# -------------------------------------------------------------------------------------------------- EXPORT DATABASE + + +def test_sql_export_database_writes_into_exports(graph_db): + name = f"adb-test-{uuid.uuid4().hex}.jsonl.tgz" + os.makedirs("exports", exist_ok=True) + target = Path("exports") / name + try: + row = graph_db.command("sql", f"EXPORT DATABASE file://{name}").one() + assert row.get("operation") == "export database" + assert target.exists() and target.stat().st_size > 0 + finally: + if target.exists(): + target.unlink() + + +def test_sql_export_database_takes_a_name_and_no_directory(graph_db): + """``ExportDatabaseStatement`` refuses a name that contains ``..`` or ``File.separator``: the second is ``/`` on Linux and macOS and + ``\\`` on Windows, where a forward slash is not refused (the first CI run of this test showed it). + """ + with pytest.raises(ArcadeDBError): + graph_db.command("sql", "EXPORT DATABASE file://..up.jsonl.tgz") + if os.name != "nt": + with pytest.raises(ArcadeDBError): + graph_db.command("sql", "EXPORT DATABASE file://sub/dir.jsonl.tgz") + else: + folder = f"adb-test-{uuid.uuid4().hex}" + os.makedirs("exports", exist_ok=True) + try: + graph_db.command("sql", f"EXPORT DATABASE file://{folder}/dir.jsonl.tgz") + finally: + shutil.rmtree(Path("exports") / folder, ignore_errors=True) + + +def test_sql_export_database_rolls_back_your_open_transaction(graph_db): + """``ExportDatabaseStatement`` calls ``rollbackAllNested()`` when a transaction is active.""" + name = f"adb-test-{uuid.uuid4().hex}.jsonl.tgz" + os.makedirs("exports", exist_ok=True) + target = Path("exports") / name + try: + graph_db.begin() + graph_db.command("sql", "INSERT INTO Note SET text = 'pending'") + graph_db.command("sql", f"EXPORT DATABASE file://{name}") + assert graph_db.is_transaction_active() is False + finally: + graph_db.rollback() + if target.exists(): + target.unlink() + assert _count(graph_db, "Note") == 3 + + +# -------------------------------------------------------------------------------------------------- export_to_csv + + +def _read_csv(path): + with open(path, newline="", encoding="utf-8") as handle: + return list(csv.reader(handle)) + + +@pytest.fixture +def people_db(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Person") + with temp_db.transaction(): + temp_db.command("sql", "INSERT INTO Person SET name = 'Alice', age = 30") + temp_db.command("sql", "INSERT INTO Person SET name = 'Bob', age = 25") + return temp_db + + +def test_export_to_csv_writes_a_header_and_the_rows(people_db, tmp_path): + target = tmp_path / "people.csv" + people_db.export_to_csv("SELECT name, age FROM Person ORDER BY name", str(target)) + rows = _read_csv(target) + assert set(rows[0]) == {"name", "age"} + by_name = { + dict(zip(rows[0], row))["name"]: dict(zip(rows[0], row)) for row in rows[1:] + } + assert by_name["Alice"]["age"] == "30" and by_name["Bob"]["age"] == "25" + + +def test_fieldnames_choose_the_order_and_cannot_rename(people_db, tmp_path): + target = tmp_path / "ordered.csv" + people_db.export_to_csv( + "SELECT name, age FROM Person ORDER BY name", + str(target), + fieldnames=["age", "name"], + ) + assert _read_csv(target)[:2] == [["age", "name"], ["30", "Alice"]] + + with pytest.raises(ArcadeDBError): + people_db.export_to_csv( + "SELECT name, age FROM Person ORDER BY name", + str(tmp_path / "short.csv"), + fieldnames=["name"], + ) + + renamed = tmp_path / "renamed.csv" + people_db.export_to_csv( + "SELECT name AS who, age AS years FROM Person ORDER BY name", + str(renamed), + fieldnames=["who", "years"], + ) + assert _read_csv(renamed)[:2] == [["who", "years"], ["Alice", "30"]] + + +def test_export_to_csv_takes_a_list_of_dicts_and_makes_the_folder(tmp_path): + from arcadedb.exporter import export_to_csv + + target = tmp_path / "new" / "folder" / "rows.csv" + export_to_csv([{"id": 1, "name": "x"}, {"id": 2, "name": "y"}], str(target)) + assert _read_csv(target) == [["id", "name"], ["1", "x"], ["2", "y"]] + + +def test_export_to_csv_with_no_rows_writes_the_header_only_when_it_is_given( + people_db, tmp_path +): + target = tmp_path / "none.csv" + people_db.export_to_csv( + "SELECT name, age FROM Person WHERE age > 100", + str(target), + fieldnames=["name", "age"], + ) + assert _read_csv(target) == [["name", "age"]] + bare = tmp_path / "bare.csv" + people_db.export_to_csv("SELECT name, age FROM Person WHERE age > 100", str(bare)) + assert bare.read_text(encoding="utf-8") == ""