feat(memgraph): use composite indexes and validate Memgraph 3.14

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

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

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

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 :<relation>(<field>)` 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 :<relation>(<field>)` 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 |

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(

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]: