diff --git a/CHANGELOG.md b/CHANGELOG.md index 8681bb6f..ed6a6ff6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Changed + +- **Memgraph** builds one composite index, in declared field order, for a multi-field + vertex index, identity or upsert match key (Memgraph 3.2 or later). The Docker default is 3.14.0. + +### Fixed + +- **Memgraph** retries a failed automatic index creation on a later batch instead of + treating it as created. + ## [1.24.0] ### Changed diff --git a/docker/README.md b/docker/README.md index a4aed117..ccdb9864 100644 --- a/docker/README.md +++ b/docker/README.md @@ -219,6 +219,8 @@ docker compose --env-file .env --profile graflo.kafka up -d Memgraph is a high-performance, in-memory graph database that supports OpenCypher and uses the Bolt protocol. +The default image is `memgraph/memgraph-mage:3.14.0`; GraFlo needs Memgraph 3.2 or later. + **Connection Details:** - Bolt port: `7687` (default, configurable via `MEMGRAPH_PORT` in `.env`) - Uses Bolt protocol (bolt://) for connections diff --git a/docker/memgraph/.env b/docker/memgraph/.env index 84ac18dd..719418c1 100644 --- a/docker/memgraph/.env +++ b/docker/memgraph/.env @@ -4,7 +4,7 @@ # Container settings SPEC=graflo CONTAINER_NAME="${SPEC}.memgraph" -IMAGE_VERSION=memgraph/memgraph-mage:3.13.1 +IMAGE_VERSION=memgraph/memgraph-mage:3.14.0 # Connection settings MEMGRAPH_HOST=localhost diff --git a/docs/concepts/schema/backend_indexes.md b/docs/concepts/schema/backend_indexes.md index a3fdb05f..c3fcc2ec 100644 --- a/docs/concepts/schema/backend_indexes.md +++ b/docs/concepts/schema/backend_indexes.md @@ -84,7 +84,7 @@ logs a warning, because every filtered read on that tag then fails with | Backend | Identity index | Declared vertex indexes | Declared edge indexes | |---|---|---|---| | Neo4j | A uniqueness constraint over the identity fields, created with the other indexes. If the server refuses a composite constraint (Community edition may), a plain index with a warning | `CREATE INDEX`, or a uniqueness constraint when `unique: true` | A relationship index, or a uniqueness constraint when `unique: true` | -| Memgraph | One single-property index per identity field and a uniqueness constraint over all of them, created with the other indexes; the match-field indexes are ensured again once per connection when vertices are written | One single-property index per field, plus a uniqueness constraint when `unique: true` | One `CREATE EDGE INDEX ON :()` per field | +| Memgraph (3.2 or later) | One composite index over the identity fields and a uniqueness constraint over all of them, created with the other indexes; the match-field index is ensured again when vertices are written, until it succeeds | One index over the declared fields, in their order, plus a uniqueness constraint when `unique: true` | One `CREATE EDGE INDEX ON :()` per field | | FalkorDB | One index per identity field, created with the other indexes | One index per field | One relationship index per field | | NebulaGraph | A tag index over the identity fields, always created, because `LOOKUP` and filtered `MATCH` need it | A tag index | An edge index | | ArangoDB | A unique persistent index over the identity fields, created with the collection; none when the identity is `_key`, which ArangoDB indexes itself | As declared, with `unique` and `type` applied | As declared, on each edge collection | diff --git a/graflo/db/memgraph/conn.py b/graflo/db/memgraph/conn.py index cc89f696..673af8ef 100644 --- a/graflo/db/memgraph/conn.py +++ b/graflo/db/memgraph/conn.py @@ -188,6 +188,28 @@ def edge_property_index_query(relation: str, field: str) -> str: return f"CREATE EDGE INDEX ON :{cypher_name(relation)}({cypher_name(field)})" +def _create_vertex_index(conn: Any, label: str, fields: Sequence[str]) -> None: + """Create one label-property index, preserving composite field order. + + Memgraph 3.2 and later support native composite indexes. An existing index + is accepted; other failures propagate so callers can retry on a later batch. + + Args: + conn: An open, autocommit Memgraph driver connection. + label: Node label to index. + fields: Property names in the declared index order. + """ + properties = ", ".join(cypher_name(field) for field in fields) + cursor = conn.cursor() + try: + cursor.execute(f"CREATE INDEX ON :{cypher_name(label)}({properties})") + except Exception as error: + if "already exists" not in str(error).lower(): + raise + finally: + cursor.close() + + def _create_unique_constraint(conn: Any, label: str, fields: list[str]) -> None: """Constrain *fields* of *label* unique together; an existing one is kept.""" props = ", ".join(f"n.{cypher_name(f)}" for f in fields) @@ -295,8 +317,8 @@ def __init__(self, config: MemgraphConfig): self.conn = mgclient.connect(**connect_kwargs) self.conn.autocommit = True self._database_name = config.database or "memgraph" - # (label, key) pairs whose MERGE index this connection already ensured. - self._merge_indexed: set[tuple[str, str]] = set() + # (label, ordered fields) whose MERGE index this connection ensured. + self._merge_indexed: set[tuple[str, tuple[str, ...]]] = set() def execute(self, query: str, **kwargs) -> QueryResult: """Execute a raw OpenCypher query against the database. @@ -503,8 +525,8 @@ def define_vertex_indexes( """Create indexes for vertex labels based on configuration. Iterates through all vertices defined in the configuration and creates - indexes for each specified field. Memgraph supports indexes on node - properties for faster lookups. Identity index is prepended when schema + indexes for each declared field group. Memgraph supports composite node + property indexes for faster lookups. Identity index is prepended when schema is provided (Memgraph has no implicit identity index). Parameters @@ -516,7 +538,7 @@ def define_vertex_indexes( Notes ----- - Index creation is idempotent (existing indexes are skipped) - - Uses Memgraph syntax: ``CREATE INDEX ON :Label(property)`` + - Uses Memgraph syntax: ``CREATE INDEX ON :Label(prop1, prop2)`` - Errors are logged but don't stop processing of other indexes """ if self.conn is None: @@ -546,25 +568,17 @@ def define_vertex_indexes( index_list = [identity_idx, *index_list] storage_label = db_vertex.vertex_dbname(label) if db_vertex else label for index_obj in index_list: - for field in index_obj.fields: - try: - query = ( - f"CREATE INDEX ON :{cypher_name(storage_label)}" - f"({cypher_name(field)})" - ) - cursor = self.conn.cursor() - cursor.execute(query) - cursor.close() - logger.debug(f"Created index on {storage_label}.{field}") - except Exception as e: - if "already exists" in str(e).lower(): - logger.debug( - f"Index on {storage_label}.{field} already exists" - ) - else: - logger.warning( - f"Failed to create index on {storage_label}.{field}: {e}" - ) + if not index_obj.fields: + continue + try: + _create_vertex_index(self.conn, storage_label, index_obj.fields) + except Exception as error: + logger.warning( + "Failed to create index on %s(%s): %s", + storage_label, + ", ".join(index_obj.fields), + error, + ) if index_obj.unique and index_obj.fields: # A constraint gives MERGE no index, so it comes on top of them. _create_unique_constraint( @@ -839,23 +853,17 @@ def upsert_docs_batch( if not sanitized_docs: return - # Auto-create index on match_keys for MERGE performance, once per + # Auto-create an index on match_keys for MERGE performance, once per # connection: the statement is DDL, and Memgraph takes a unique-access # lock for it even when the index exists. - missing = [k for k in match_keys if (class_name, k) not in self._merge_indexed] - if missing: - cursor = self.conn.cursor() - for key in missing: - try: - cursor.execute( - f"CREATE INDEX ON :{cypher_name(class_name)}({cypher_name(key)})" - ) - logger.debug(f"Created index on {class_name}.{key}") - except Exception as e: - if "already exists" not in str(e).lower(): - logger.debug(f"Index on {class_name}.{key}: {e}") - self._merge_indexed.add((class_name, key)) - cursor.close() + merge_index = (class_name, tuple(match_keys_list)) + if match_keys_list and merge_index not in self._merge_indexed: + try: + _create_vertex_index(self.conn, class_name, match_keys_list) + except Exception as error: + logger.debug("Index on %s%s: %s", class_name, match_keys_list, error) + else: + self._merge_indexed.add(merge_index) # Build the MERGE clause with match keys index_str = ", ".join( diff --git a/test/backends/test_memgraph_ddl.py b/test/backends/test_memgraph_ddl.py index 717f70ce..6cd1656f 100644 --- a/test/backends/test_memgraph_ddl.py +++ b/test/backends/test_memgraph_ddl.py @@ -4,7 +4,9 @@ from typing import Any -from graflo.db.memgraph.conn import edge_property_index_query +import pytest + +from graflo.db.memgraph.conn import _create_vertex_index, edge_property_index_query def test_an_edge_index_is_an_edge_type_index() -> None: @@ -17,21 +19,57 @@ def test_an_edge_index_is_an_edge_type_index() -> None: class _Cursor: def __init__(self, statements: list[str]) -> None: self._statements = statements + self.close_calls = 0 def execute(self, query: str, params: object = None) -> None: + if self.close_calls: + raise RuntimeError("Cannot execute a closed cursor") self._statements.append(" ".join(query.split())) def close(self) -> None: - return None + self.close_calls += 1 class _Conn: def __init__(self) -> None: self.statements: list[str] = [] + self.cursors: list[_Cursor] = [] self.autocommit = False def cursor(self) -> _Cursor: - return _Cursor(self.statements) + cursor = _Cursor(self.statements) + self.cursors.append(cursor) + return cursor + + +@pytest.mark.parametrize( + "error_message", [None, "Index already exists", "Database busy"] +) +def test_vertex_index_cursor_closes_after_every_outcome( + monkeypatch: pytest.MonkeyPatch, error_message: str | None +) -> None: + """Close the index cursor once, including ignored and propagated errors.""" + fake = _Conn() + execute = _Cursor.execute + error = RuntimeError(error_message) if error_message is not None else None + + def execute_index(self: _Cursor, query: str, params: object = None) -> None: + """Record the attempted DDL before simulating a driver error.""" + execute(self, query, params) + if error is not None: + raise error + + monkeypatch.setattr(_Cursor, "execute", execute_index) + if error_message == "Database busy": + with pytest.raises(RuntimeError, match="Database busy") as caught: + _create_vertex_index(fake, "person", ["tenant", "id"]) + assert caught.value is error + else: + _create_vertex_index(fake, "person", ["tenant", "id"]) + + assert fake.statements == ["CREATE INDEX ON :person(tenant, id)"] + assert len(fake.cursors) == 1 + assert fake.cursors[0].close_calls == 1 def test_an_upsert_creates_its_merge_index_once_per_connection(monkeypatch) -> None: @@ -87,12 +125,93 @@ def test_a_vertex_identity_is_indexed_and_constrained_unique(monkeypatch) -> Non db.define_vertex_indexes(schema.core_schema.vertex_config, schema) assert fake.statements == [ - "CREATE INDEX ON :person(name)", - "CREATE INDEX ON :person(born)", + "CREATE INDEX ON :person(name, born)", "CREATE CONSTRAINT ON (n:person) ASSERT n.name, n.born IS UNIQUE", ] +@pytest.mark.parametrize("tuple_keys", [False, True]) +def test_composite_merge_indexes_distinguish_field_order( + monkeypatch: pytest.MonkeyPatch, tuple_keys: bool +) -> None: + """Reversing the same fields creates a distinct index, each ensured once.""" + db, fake = _memgraph(monkeypatch) + + for fields in [["tenant", "id"], ["id", "tenant"]] * 2: + db.upsert_docs_batch( + [{"tenant": "acme", "id": "1"}], + "person", + match_keys=tuple(fields) if tuple_keys else fields, + ) + assert all(cursor.close_calls == 1 for cursor in fake.cursors) + + assert [s for s in fake.statements if s.startswith("CREATE INDEX")] == [ + "CREATE INDEX ON :person(tenant, id)", + "CREATE INDEX ON :person(id, tenant)", + ] + assert sum(s.startswith("UNWIND") for s in fake.statements) == 4 + + +@pytest.mark.parametrize("already_exists", [False, True]) +def test_only_successful_merge_indexes_are_cached( + monkeypatch: pytest.MonkeyPatch, already_exists: bool +) -> None: + """Retry failed DDL without interrupting ingestion or retrying existing indexes.""" + db, fake = _memgraph(monkeypatch) + execute = _Cursor.execute + attempts: list[str] = [] + + def fail_first_index(self: _Cursor, query: str, params: object = None) -> None: + """Simulate a catalogue failure on the first index request.""" + if query.startswith("CREATE INDEX"): + attempts.append(query) + if len(attempts) == 1: + raise RuntimeError("Index already exists" if already_exists else "busy") + execute(self, query, params) + + monkeypatch.setattr(_Cursor, "execute", fail_first_index) + for _ in range(3): + db.upsert_docs_batch([{"id": "1"}], "person", match_keys=["id"]) + assert all(cursor.close_calls == 1 for cursor in fake.cursors) + + assert len(attempts) == (1 if already_exists else 2) + assert sum(s.startswith("UNWIND") for s in fake.statements) == 3 + + +def test_merge_indexes_distinguish_field_groups( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An existing simple index does not stand in for a composite one.""" + db, fake = _memgraph(monkeypatch) + + for fields in [["id"], ["tenant", "id"], ["id"], ["tenant", "id"]]: + db.upsert_docs_batch( + [{"tenant": "acme", "id": "1"}], "person", match_keys=fields + ) + + assert [s for s in fake.statements if s.startswith("CREATE INDEX")] == [ + "CREATE INDEX ON :person(id)", + "CREATE INDEX ON :person(tenant, id)", + ] + + +def test_composite_indexes_quote_each_identifier( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A comma or backtick in a property name remains part of that name.""" + db, fake = _memgraph(monkeypatch) + + db.upsert_docs_batch( + [{"tenant, id": "acme", "external`id": "1"}], + "work item", + match_keys=["tenant, id", "external`id"], + ) + + assert fake.statements[0] == ( + "CREATE INDEX ON :`work item`(`tenant, id`, `external``id`)" + ) + + class _CatalogCursor(_Cursor): """Answers ``SHOW CONSTRAINT INFO`` with a fixed catalogue.""" @@ -111,7 +230,9 @@ def fetchall(self) -> list[tuple]: class _CatalogConn(_Conn): def cursor(self) -> _Cursor: - return _CatalogCursor(self.statements) + cursor = _CatalogCursor(self.statements) + self.cursors.append(cursor) + return cursor def _memgraph_with_catalog(monkeypatch) -> tuple[Any, _Conn]: diff --git a/test/db/memgraphs/test_db_index.py b/test/db/memgraphs/test_db_index.py index 7007d81a..61f38054 100644 --- a/test/db/memgraphs/test_db_index.py +++ b/test/db/memgraphs/test_db_index.py @@ -3,6 +3,10 @@ Schema-driven vertex index creation is shared from cypher_cases. """ +import pytest + +from graflo.architecture.schema import Schema +from graflo.connections.onto import MemgraphConfig from graflo.db.manager import ConnectionManager from test.db.cypher_cases.lifecycle import ( test_create_vertex_index, @@ -13,6 +17,68 @@ ] +def test_composite_vertex_indexes( + conn_conf: MemgraphConfig, test_graph_name: str, clean_db: None +) -> None: + """Schema and upserts preserve composite indexes and enforce the identity.""" + label = test_graph_name + schema = Schema.model_validate( + { + "metadata": {"name": "composite_indexes"}, + "core_schema": { + "vertex_config": { + "vertices": [ + { + "name": label, + "properties": ["tenant", "id", "name", "born"], + "identity": ["tenant", "id"], + } + ] + }, + "edge_config": {"edges": []}, + }, + "db_profile": { + "db_flavor": "memgraph", + "vertex_indexes": {label: [{"fields": ["name", "born"]}]}, + }, + } + ) + with ConnectionManager(connection_config=conn_conf) as db: + # Repeating schema definition must also accept existing composite indexes. + for _ in range(2): + db.define_vertex_indexes(schema.core_schema.vertex_config, schema) + db.upsert_docs_batch( + [ + {"tenant": "acme", "id": "1", "name": "Alice", "born": 1990}, + {"tenant": "other", "id": "1", "name": "Bob", "born": 1991}, + ], + label, + match_keys=["tenant", "id"], + ) + db.upsert_docs_batch( + [{"tenant": "acme", "id": "1", "name": "Alicia"}], + label, + match_keys=["tenant", "id"], + ) + + catalog = db.execute("SHOW INDEX INFO") + indexes = [dict(zip(catalog.columns, row)) for row in catalog] + assert { + tuple(row["property"]) + for row in indexes + if row["label"] == label and row["index type"] == "label+property" + } == {("tenant", "id"), ("name", "born")} + assert sorted(db.fetch_docs(label), key=lambda doc: doc["tenant"]) == [ + {"tenant": "acme", "id": "1", "name": "Alicia", "born": 1990}, + {"tenant": "other", "id": "1", "name": "Bob", "born": 1991}, + ] + with pytest.raises(Exception, match="[Uu]nique"): + db.execute( + f"CREATE (:{label} {{tenant: $tenant, id: $id}})", tenant="acme", id="1" + ) + assert len(db.fetch_docs(label)) == 2 + + def test_create_edge_index(conn_conf, test_graph_name, schema_obj): """Test creating edge indexes from schema.""" schema_o = schema_obj("review")