diff --git a/drivine/backend/oracle/oracle_schema_grammar.py b/drivine/backend/oracle/oracle_schema_grammar.py index 775bc25..3ca05a0 100644 --- a/drivine/backend/oracle/oracle_schema_grammar.py +++ b/drivine/backend/oracle/oracle_schema_grammar.py @@ -52,15 +52,50 @@ def recorded_as(kind: CatalogKind) -> list[str]: """ +BUSY_ATTEMPTS = 300 +""" +How many times `_made` runs an index's statement that another session's lock on the node table turned away (ORA-00054), +a tenth of a second apart, before it raises: as drivine4j's `OracleGraph.BUSY_ATTEMPTS`. +""" + +_WAITS_OUT_BUSY = "PRAGMA EXCEPTION_INIT(e_busy, -54);" +"""What declares the busy lock `_made` waits out: a block that holds it has waited before it fails for that lock.""" + + +def waits_out_busy_lock(statement: str) -> bool: + """ + Whether `statement` waited out another session's busy lock (ORA-00054) itself before it failed for it, as a block of + `_made` does for thirty seconds: the schema executor doesn't run it again for that lock, or each run would wait as + long again. + """ + return _WAITS_OUT_BUSY in statement + + def _made(ddl: str) -> str: - """A block that runs `ddl`, an object another session made meanwhile (ORA-00955) taken as made.""" + """ + A block that runs `ddl`, an index over the node table, an object another session made meanwhile (ORA-00955) taken + as made. A statement turned away because another session holds the node table's lock (ORA-00054), as one making the + same index, or writing a node, does, is run again a tenth of a second later, for up to `BUSY_ATTEMPTS` tries, and + then finds what the other made: managers making the store key's constraint together otherwise ran out of attempts + on the busy lock, and went on without the constraint. drivine4j's `OracleGraph.ensuring` waits out a busy lock so. + """ return f"""DECLARE e_exists EXCEPTION; PRAGMA EXCEPTION_INIT(e_exists, -955); + e_busy EXCEPTION; + {_WAITS_OUT_BUSY} BEGIN - EXECUTE IMMEDIATE {sql_text(ddl)}; -EXCEPTION - WHEN e_exists THEN NULL; + FOR attempt IN 1 .. {BUSY_ATTEMPTS} LOOP + BEGIN + EXECUTE IMMEDIATE {sql_text(ddl)}; + RETURN; + EXCEPTION + WHEN e_exists THEN RETURN; + WHEN e_busy THEN + IF attempt = {BUSY_ATTEMPTS} THEN RAISE; END IF; + DBMS_SESSION.SLEEP(0.1); + END; + END LOOP; END;""" diff --git a/drivine/schema/schema_executor.py b/drivine/schema/schema_executor.py index 99a6e8b..39d284d 100644 --- a/drivine/schema/schema_executor.py +++ b/drivine/schema/schema_executor.py @@ -5,24 +5,27 @@ from typing import Any from drivine.backend.cypher_grammar import FALKORDB_GRAMMAR, NEO4J_GRAMMAR +from drivine.backend.oracle.oracle_schema_grammar import waits_out_busy_lock from drivine.backend.postgres.postgres_statement import value_of from drivine.connection.connection_provider import ConnectionProvider from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError -from drivine.manager.contention import any_cause, ddl_contended, when_not_contended +from drivine.manager.contention import any_cause, ddl_contended, ddl_deadlocked, when_not_contended from drivine.query.query_specification import QuerySpecification from drivine.schema.native_operation import NativeOperation _ORACLE_SERIAL = re.compile(r"ORA-08177\b") -def _contended(failure: BaseException) -> bool: +def _contended(failure: BaseException, waited_out_busy: bool) -> bool: """ A transient error over Bolt, FalkorDB turning a write away while another is in progress, an Oracle serialization failure, or a SQL store turning DDL away because of another session's (see ddl_contended), as when managers that - start together each make the indexes of the store's key. + start together each make the indexes of the store's key. A statement that `waited_out_busy`, as an Oracle index's + block does (see waits_out_busy_lock), is taken as turned away only for a deadlock (see ddl_deadlocked). """ - return ddl_contended(failure) or any_cause( + ddl = ddl_deadlocked(failure) if waited_out_busy else ddl_contended(failure) + return ddl or any_cause( failure, lambda it: ( NEO4J_GRAMMAR.contended(it) or FALKORDB_GRAMMAR.contended(it) or bool(_ORACLE_SERIAL.search(str(it))) @@ -42,7 +45,10 @@ async def run(self, statement: str | NativeOperation, bindings: Mapping[str, Any """ Runs a statement, again after a pause when the engine turned it away because another write was in progress. """ - return await when_not_contended(lambda: self._run_once(statement, bindings or {}), _contended) + waited_out_busy = isinstance(statement, str) and waits_out_busy_lock(statement) + return await when_not_contended( + lambda: self._run_once(statement, bindings or {}), lambda it: _contended(it, waited_out_busy) + ) async def _run_once(self, statement: str | NativeOperation, bindings: Mapping[str, Any]) -> list[Any]: if isinstance(statement, NativeOperation): diff --git a/drivine/sync/schema/schema_executor.py b/drivine/sync/schema/schema_executor.py index c060fc7..149ac82 100644 --- a/drivine/sync/schema/schema_executor.py +++ b/drivine/sync/schema/schema_executor.py @@ -7,24 +7,27 @@ from typing import Any from drivine.backend.cypher_grammar import FALKORDB_GRAMMAR, NEO4J_GRAMMAR +from drivine.backend.oracle.oracle_schema_grammar import waits_out_busy_lock from drivine.backend.postgres.postgres_statement import value_of from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError from drivine.query.query_specification import QuerySpecification from drivine.schema.native_operation import NativeOperation from drivine.sync.connection.connection_provider import ConnectionProvider -from drivine.sync.manager.contention import any_cause, ddl_contended, when_not_contended +from drivine.sync.manager.contention import any_cause, ddl_contended, ddl_deadlocked, when_not_contended _ORACLE_SERIAL = re.compile(r"ORA-08177\b") -def _contended(failure: BaseException) -> bool: +def _contended(failure: BaseException, waited_out_busy: bool) -> bool: """ A transient error over Bolt, FalkorDB turning a write away while another is in progress, an Oracle serialization failure, or a SQL store turning DDL away because of another session's (see ddl_contended), as when managers that - start together each make the indexes of the store's key. + start together each make the indexes of the store's key. A statement that `waited_out_busy`, as an Oracle index's + block does (see waits_out_busy_lock), is taken as turned away only for a deadlock (see ddl_deadlocked). """ - return ddl_contended(failure) or any_cause( + ddl = ddl_deadlocked(failure) if waited_out_busy else ddl_contended(failure) + return ddl or any_cause( failure, lambda it: ( NEO4J_GRAMMAR.contended(it) or FALKORDB_GRAMMAR.contended(it) or bool(_ORACLE_SERIAL.search(str(it))) @@ -44,7 +47,10 @@ def run(self, statement: str | NativeOperation, bindings: Mapping[str, Any] | No """ Runs a statement, again after a pause when the engine turned it away because another write was in progress. """ - return when_not_contended(lambda: self._run_once(statement, bindings or {}), _contended) + waited_out_busy = isinstance(statement, str) and waits_out_busy_lock(statement) + return when_not_contended( + lambda: self._run_once(statement, bindings or {}), lambda it: _contended(it, waited_out_busy) + ) def _run_once(self, statement: str | NativeOperation, bindings: Mapping[str, Any]) -> list[Any]: if isinstance(statement, NativeOperation): diff --git a/tests/golden/VERSION b/tests/golden/VERSION index 3c91767..e7bfdc8 100644 --- a/tests/golden/VERSION +++ b/tests/golden/VERSION @@ -1 +1 @@ -32b7497ad14ce26d40058037da16b89542562540 +c34cf55178cf5971c4963e5a0f2445b581d469a7 diff --git a/tests/golden/statements.json b/tests/golden/statements.json index 50f1b11..485b5b2 100644 --- a/tests/golden/statements.json +++ b/tests/golden/statements.json @@ -333,10 +333,10 @@ "schema.listUniques": "SELECT JSON_OBJECT('label' VALUE c.label, 'properties' VALUE JSON(c.properties), 'name' VALUE c.name, 'details' VALUE c.details, 'incomplete' VALUE CASE WHEN c.kind = 'RANGE_INDEX' AND NOT EXISTS (SELECT 1 FROM user_indexes l WHERE l.index_name = UPPER(JSON_VALUE(c.details, '$.labelled'))) AND NOT EXISTS (SELECT 1 FROM drivine_schema u WHERE u.label = c.label AND u.properties = c.properties AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name))) THEN 'true' END RETURNING JSON) AS \"item\" FROM drivine_schema c WHERE c.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND (c.object_name IS NULL OR EXISTS (SELECT 1 FROM user_indexes i WHERE i.index_name = UPPER(c.object_name)) OR EXISTS (SELECT 1 FROM user_tables t WHERE t.table_name = UPPER(c.object_name))) ORDER BY c.label, c.properties", "schema.listVectors": "SELECT JSON_OBJECT('label' VALUE c.label, 'properties' VALUE JSON(c.properties), 'name' VALUE c.name, 'details' VALUE c.details, 'incomplete' VALUE CASE WHEN c.kind = 'RANGE_INDEX' AND NOT EXISTS (SELECT 1 FROM user_indexes l WHERE l.index_name = UPPER(JSON_VALUE(c.details, '$.labelled'))) AND NOT EXISTS (SELECT 1 FROM drivine_schema u WHERE u.label = c.label AND u.properties = c.properties AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name))) THEN 'true' END RETURNING JSON) AS \"item\" FROM drivine_schema c WHERE c.kind IN ('VECTOR_INDEX', 'VECTOR') AND (c.object_name IS NULL OR EXISTS (SELECT 1 FROM user_indexes i WHERE i.index_name = UPPER(c.object_name)) OR EXISTS (SELECT 1 FROM user_tables t WHERE t.table_name = UPPER(c.object_name))) ORDER BY c.label, c.properties", "schema.listFullTexts": "SELECT JSON_OBJECT('label' VALUE c.label, 'properties' VALUE JSON(c.properties), 'name' VALUE c.name, 'details' VALUE c.details, 'incomplete' VALUE CASE WHEN c.kind = 'RANGE_INDEX' AND NOT EXISTS (SELECT 1 FROM user_indexes l WHERE l.index_name = UPPER(JSON_VALUE(c.details, '$.labelled'))) AND NOT EXISTS (SELECT 1 FROM drivine_schema u WHERE u.label = c.label AND u.properties = c.properties AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name))) THEN 'true' END RETURNING JSON) AS \"item\" FROM drivine_schema c WHERE c.kind IN ('FULLTEXT_INDEX', 'FULLTEXT') AND (c.object_name IS NULL OR EXISTS (SELECT 1 FROM user_indexes i WHERE i.index_name = UPPER(c.object_name)) OR EXISTS (SELECT 1 FROM user_tables t WHERE t.table_name = UPPER(c.object_name))) ORDER BY c.label, c.properties", - "schema.createRange": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_lastActivityAtsessionId_8a35af29 ON drivine_node (JSON_VALUE(props, ''$.lastActivityAt''), JSON_VALUE(props, ''$.sessionId''))';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nDECLARE\n v_unique NUMBER;\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n SELECT COUNT(*) INTO v_unique FROM drivine_schema u WHERE u.label = 'Session' AND u.properties = '[\"lastActivityAt\",\"sessionId\"]' AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name));\n IF v_unique = 0 THEN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_lx_SessionlastActivityAtsessionId_396a2573 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"lastActivityAt\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"sessionId\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n END IF;\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Session' AS label, '[\"lastActivityAt\",\"sessionId\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN MATCHED THEN UPDATE SET c.details = '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}'\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Session_lastActivityAt_sessionId_range', 'drivine_ix_lastActivityAtsessionId_8a35af29', '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", + "schema.createRange": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_lastActivityAtsessionId_8a35af29 ON drivine_node (JSON_VALUE(props, ''$.lastActivityAt''), JSON_VALUE(props, ''$.sessionId''))';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nDECLARE\n v_unique NUMBER;\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n SELECT COUNT(*) INTO v_unique FROM drivine_schema u WHERE u.label = 'Session' AND u.properties = '[\"lastActivityAt\",\"sessionId\"]' AND u.kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND EXISTS (SELECT 1 FROM user_indexes q WHERE q.index_name = UPPER(u.object_name));\n IF v_unique = 0 THEN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_lx_SessionlastActivityAtsessionId_396a2573 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"lastActivityAt\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Session\")'') THEN JSON_VALUE(n.props, ''$.\"sessionId\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n END IF;\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Session' AS label, '[\"lastActivityAt\",\"sessionId\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN MATCHED THEN UPDATE SET c.details = '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}'\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Session_lastActivityAt_sessionId_range', 'drivine_ix_lastActivityAtsessionId_8a35af29', '{\"labelled\": \"drivine_lx_SessionlastActivityAtsessionId_396a2573\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", "schema.dropRange": "DECLARE\n v_others NUMBER;\nBEGIN\n FOR it IN (SELECT object_name, JSON_VALUE(details, '$.labelled') AS labelled FROM drivine_schema WHERE kind IN ('RANGE_INDEX', 'RANGE') AND label = 'Session' AND properties = '[\"lastActivityAt\",\"sessionId\"]') LOOP\n SELECT COUNT(*) INTO v_others FROM drivine_schema\n WHERE UPPER(object_name) = UPPER(it.object_name) AND NOT (kind IN ('RANGE_INDEX', 'RANGE') AND label = 'Session' AND properties = '[\"lastActivityAt\",\"sessionId\"]');\n IF it.labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.labelled;\n END IF;\n IF it.object_name IS NOT NULL AND v_others = 0 THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END IF;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind IN ('RANGE_INDEX', 'RANGE') AND label = 'Session' AND properties = '[\"lastActivityAt\",\"sessionId\"]';\n COMMIT;\nEND;", - "schema.createUnique": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_uuid_36f3bb ON drivine_node (JSON_VALUE(props, ''$.uuid''))';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_range', 'drivine_ix_uuid_36f3bb', '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personuuid_44d1cd66 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"uuid\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_unique', 'drivine_uq_Personuuid_44d1cd66', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", - "schema.createUnique.pair": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_emailtenant_6d0f958e ON drivine_node (JSON_VALUE(props, ''$.email''), JSON_VALUE(props, ''$.tenant''))';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_range', 'drivine_ix_emailtenant_6d0f958e', '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\nBEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personemailtenant_83779439 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"email\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"tenant\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\nEXCEPTION\n WHEN e_exists THEN NULL;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_unique', 'drivine_uq_Personemailtenant_83779439', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", + "schema.createUnique": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_uuid_36f3bb ON drivine_node (JSON_VALUE(props, ''$.uuid''))';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_range', 'drivine_ix_uuid_36f3bb', '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"uuid\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personuuid_44d1cd66\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"uuid\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personuuid_44d1cd66 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"uuid\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"uuid\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_uuid_unique', 'drivine_uq_Personuuid_44d1cd66', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", + "schema.createUnique.pair": "DECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE INDEX IF NOT EXISTS drivine_ix_emailtenant_6d0f958e ON drivine_node (JSON_VALUE(props, ''$.email''), JSON_VALUE(props, ''$.tenant''))';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'RANGE_INDEX' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_range', 'drivine_ix_emailtenant_6d0f958e', '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;\nDECLARE\n v_labelled VARCHAR2(128);\nBEGIN\n SELECT MAX(JSON_VALUE(details, '$.labelled')) INTO v_labelled FROM drivine_schema WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n IF v_labelled IS NOT NULL THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || v_labelled;\n END IF;\n FOR it IN (SELECT object_name FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]') LOOP\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind = 'RANGE' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n UPDATE drivine_schema SET details = '{\"unique\": \"drivine_uq_Personemailtenant_83779439\"}' WHERE kind = 'RANGE_INDEX' AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n COMMIT;\nEND;\nDECLARE\n e_exists EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_exists, -955);\n e_busy EXCEPTION;\n PRAGMA EXCEPTION_INIT(e_busy, -54);\nBEGIN\n FOR attempt IN 1 .. 300 LOOP\n BEGIN\n EXECUTE IMMEDIATE 'CREATE UNIQUE INDEX IF NOT EXISTS drivine_uq_Personemailtenant_83779439 ON drivine_node n (CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"email\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END, CASE WHEN JSON_EXISTS(n.labels, ''$?(@ == \"Person\")'') THEN JSON_VALUE(n.props, ''$.\"tenant\"'' RETURNING VARCHAR2(4000) NULL ON ERROR) END)';\n RETURN;\n EXCEPTION\n WHEN e_exists THEN RETURN;\n WHEN e_busy THEN\n IF attempt = 300 THEN RAISE; END IF;\n DBMS_SESSION.SLEEP(0.1);\n END;\n END LOOP;\nEND;\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'UNIQUENESS_CONSTRAINT' AS kind, 'Person' AS label, '[\"email\",\"tenant\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Person_email_tenant_unique', 'drivine_uq_Personemailtenant_83779439', '{}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", "schema.dropUnique": "DECLARE\n v_others NUMBER;\nBEGIN\n FOR it IN (SELECT object_name, JSON_VALUE(details, '$.labelled') AS labelled FROM drivine_schema WHERE kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND label = 'Person' AND properties = '[\"email\",\"tenant\"]') LOOP\n SELECT COUNT(*) INTO v_others FROM drivine_schema\n WHERE UPPER(object_name) = UPPER(it.object_name) AND NOT (kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND label = 'Person' AND properties = '[\"email\",\"tenant\"]');\n IF it.object_name IS NOT NULL AND v_others = 0 THEN\n EXECUTE IMMEDIATE 'DROP INDEX IF EXISTS ' || it.object_name;\n END IF;\n END LOOP;\n DELETE FROM drivine_schema WHERE kind IN ('UNIQUENESS_CONSTRAINT', 'UNIQUE') AND label = 'Person' AND properties = '[\"email\",\"tenant\"]';\n COMMIT;\nEND;", "schema.createVector": "CREATE TABLE IF NOT EXISTS drivine_v_Chunkembedding_7e60bd64 (nid NUMBER PRIMARY KEY REFERENCES drivine_node (nid) ON DELETE CASCADE, embedding VECTOR(1536, FLOAT32) NOT NULL)\nBEGIN EXECUTE IMMEDIATE 'CREATE OR REPLACE TRIGGER drivine_v_Chunkembedding_7e60bd64_t AFTER INSERT OR UPDATE OF labels, props ON drivine_node FOR EACH ROW BEGIN IF JSON_EXISTS(:NEW.labels, ''$?(@ == \"Chunk\")'') AND JSON_EXISTS(:NEW.props, ''$.\"embedding\"'') THEN BEGIN MERGE INTO drivine_v_Chunkembedding_7e60bd64 s USING (SELECT :NEW.nid AS nid, TO_VECTOR(JSON_QUERY(:NEW.props, ''$.\"embedding\"'' RETURNING CLOB)) AS embedding FROM dual) n ON (s.nid = n.nid) WHEN MATCHED THEN UPDATE SET s.embedding = n.embedding WHEN NOT MATCHED THEN INSERT (nid, embedding) VALUES (n.nid, n.embedding); EXCEPTION WHEN OTHERS THEN DELETE FROM drivine_v_Chunkembedding_7e60bd64 WHERE nid = :NEW.nid; END; ELSE DELETE FROM drivine_v_Chunkembedding_7e60bd64 WHERE nid = :NEW.nid; END IF; END;'; END;\nINSERT INTO drivine_v_Chunkembedding_7e60bd64 (nid, embedding) SELECT n.nid, TO_VECTOR(JSON_QUERY(n.props, '$.\"embedding\"' RETURNING CLOB)) FROM drivine_node n WHERE JSON_EXISTS(n.labels, '$?(@ == \"Chunk\")') AND JSON_EXISTS(n.props, '$.\"embedding\"') AND VECTOR_DIMENSION_COUNT(TO_VECTOR(JSON_QUERY(n.props, '$.\"embedding\"' RETURNING CLOB))) = 1536 AND NOT EXISTS (SELECT 1 FROM drivine_v_Chunkembedding_7e60bd64 s WHERE s.nid = n.nid)\nCREATE VECTOR INDEX IF NOT EXISTS drivine_v_Chunkembedding_7e60bd64_i ON drivine_v_Chunkembedding_7e60bd64 (embedding) ORGANIZATION NEIGHBOR PARTITIONS DISTANCE COSINE\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'VECTOR_INDEX' AS kind, 'Chunk' AS label, '[\"embedding\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Chunk_embedding_vector', 'drivine_v_Chunkembedding_7e60bd64', '{\"dimensions\": 1536, \"similarity\": \"COSINE\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", "schema.createVector.plain": "CREATE TABLE IF NOT EXISTS drivine_v_Chunksummary_1a6b3cf3 (nid NUMBER PRIMARY KEY REFERENCES drivine_node (nid) ON DELETE CASCADE, embedding VECTOR(384, FLOAT32) NOT NULL)\nBEGIN EXECUTE IMMEDIATE 'CREATE OR REPLACE TRIGGER drivine_v_Chunksummary_1a6b3cf3_t AFTER INSERT OR UPDATE OF labels, props ON drivine_node FOR EACH ROW BEGIN IF JSON_EXISTS(:NEW.labels, ''$?(@ == \"Chunk\")'') AND JSON_EXISTS(:NEW.props, ''$.\"summary\"'') THEN BEGIN MERGE INTO drivine_v_Chunksummary_1a6b3cf3 s USING (SELECT :NEW.nid AS nid, TO_VECTOR(JSON_QUERY(:NEW.props, ''$.\"summary\"'' RETURNING CLOB)) AS embedding FROM dual) n ON (s.nid = n.nid) WHEN MATCHED THEN UPDATE SET s.embedding = n.embedding WHEN NOT MATCHED THEN INSERT (nid, embedding) VALUES (n.nid, n.embedding); EXCEPTION WHEN OTHERS THEN DELETE FROM drivine_v_Chunksummary_1a6b3cf3 WHERE nid = :NEW.nid; END; ELSE DELETE FROM drivine_v_Chunksummary_1a6b3cf3 WHERE nid = :NEW.nid; END IF; END;'; END;\nINSERT INTO drivine_v_Chunksummary_1a6b3cf3 (nid, embedding) SELECT n.nid, TO_VECTOR(JSON_QUERY(n.props, '$.\"summary\"' RETURNING CLOB)) FROM drivine_node n WHERE JSON_EXISTS(n.labels, '$?(@ == \"Chunk\")') AND JSON_EXISTS(n.props, '$.\"summary\"') AND VECTOR_DIMENSION_COUNT(TO_VECTOR(JSON_QUERY(n.props, '$.\"summary\"' RETURNING CLOB))) = 384 AND NOT EXISTS (SELECT 1 FROM drivine_v_Chunksummary_1a6b3cf3 s WHERE s.nid = n.nid)\nCREATE VECTOR INDEX IF NOT EXISTS drivine_v_Chunksummary_1a6b3cf3_i ON drivine_v_Chunksummary_1a6b3cf3 (embedding) ORGANIZATION NEIGHBOR PARTITIONS DISTANCE EUCLIDEAN_SQUARED\nBEGIN\n MERGE INTO drivine_schema c\n USING (SELECT 'VECTOR_INDEX' AS kind, 'Chunk' AS label, '[\"summary\"]' AS properties FROM dual) k\n ON (c.kind = k.kind AND c.label = k.label AND c.properties = k.properties)\n WHEN NOT MATCHED THEN INSERT (kind, label, properties, name, object_name, details)\n VALUES (k.kind, k.label, k.properties, 'Chunk_summary_vector', 'drivine_v_Chunksummary_1a6b3cf3', '{\"dimensions\": 384, \"similarity\": \"EUCLIDEAN\"}');\nEXCEPTION\n WHEN DUP_VAL_ON_INDEX THEN NULL;\nEND;", diff --git a/tests/unit/test_contention.py b/tests/unit/test_contention.py index 58527f8..b01692e 100644 --- a/tests/unit/test_contention.py +++ b/tests/unit/test_contention.py @@ -10,10 +10,13 @@ from drivine.backend.cypher_backend import CypherBackend from drivine.backend.cypher_grammar import FALKORDB_GRAMMAR, MEMGRAPH_GRAMMAR, NEO4J_GRAMMAR +from drivine.backend.oracle.oracle_graph import OracleGraph +from drivine.backend.oracle.oracle_schema_grammar import oracle_schema_grammar from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError from drivine.manager.contention import CONTENDED_ATTEMPTS, ddl_contended, ddl_deadlocked, when_not_contended from drivine.schema.schema_executor import SchemaExecutor +from drivine.schema.schema_specs import uniqueness_constraint def bolt(code: str) -> Neo4jError: @@ -191,6 +194,52 @@ async def connect(self) -> Connection: assert ran == runs +@pytest.mark.parametrize( + ("message", "runs"), + [ + ("ORA-00054: resource busy and acquire with NOWAIT specified or timeout expired", 1), + ("ORA-04020: deadlock detected while trying to lock object JVM_NODE", 2), + ], +) +async def test_an_oracle_indexs_block_that_waited_out_a_busy_lock_itself_is_not_run_again_for_it_but_is_for_a_deadlock( + message: str, runs: int +): + # An Oracle index's block has waited thirty seconds on the busy lock before it gives up: run again, as other DDL + # is, it would wait as long again each time, three minutes in all. + block = next( + it + for it in oracle_schema_grammar(OracleGraph("jvm")).create_unique(uniqueness_constraint("DrivineStore", "key")) + if isinstance(it, str) and "CREATE UNIQUE INDEX" in it + ) + ran = 0 + + class Connection: + async def query(self, _spec: object) -> list[object]: + nonlocal ran + ran += 1 + if ran == 1: + raise oracledb.DatabaseError(message) + return [] + + async def release(self, _error: BaseException | None = None) -> None: + return None + + class Provider: + name = "oracle" + type = DatabaseType.ORACLE + + async def connect(self) -> Connection: + return Connection() + + run = SchemaExecutor(Provider()).run(block) # pyright: ignore[reportArgumentType] + if runs == 2: + assert await run == [] + else: + with pytest.raises(DrivineError, match="ORA-00054"): + await run + assert ran == runs + + @pytest.mark.parametrize( "failure", [ diff --git a/tests/unit/test_oracle_schema_grammar.py b/tests/unit/test_oracle_schema_grammar.py index 0e0cb1d..9d86da4 100644 --- a/tests/unit/test_oracle_schema_grammar.py +++ b/tests/unit/test_oracle_schema_grammar.py @@ -1,12 +1,13 @@ """Oracle's schema statements where the golden statements hold none of the case, the catalog's key, and its items.""" +import re from typing import Any import pytest -from drivine import range_index +from drivine import range_index, uniqueness_constraint from drivine.backend.oracle.oracle_graph import OracleGraph -from drivine.backend.oracle.oracle_schema_grammar import oracle_schema_grammar +from drivine.backend.oracle.oracle_schema_grammar import BUSY_ATTEMPTS, oracle_schema_grammar from drivine.backend.sql.sql_catalog import catalog_properties, catalog_record from drivine.connection.database_type import DatabaseType from drivine.drivine_error import DrivineError @@ -37,6 +38,23 @@ def test_quotes_a_key_drivine4j_quotes_one_a_json_path_reads_as_a_value_or_that_ assert "(JSON_VALUE(props, ''$.\"a\\\"b\"''))" in shared_index('a"b') +def test_runs_an_index_over_the_node_table_again_while_another_sessions_lock_turns_it_away_up_to_a_bound(): + # Managers making the store key's constraint together were turned away at once by each other's lock on the node + # table (ORA-00054), ran out of attempts, and went on without it. + unique = next( + it + for it in GRAMMAR.create_unique(uniqueness_constraint("DrivineStore", "key")) + if isinstance(it, str) and "CREATE UNIQUE INDEX" in it + ) + for block in (shared_index("name"), unique): + assert "PRAGMA EXCEPTION_INIT(e_busy, -54);" in block + assert f"FOR attempt IN 1 .. {BUSY_ATTEMPTS} LOOP" in block + assert re.search( + r"WHEN e_busy THEN\s+IF attempt = \d+ THEN RAISE; END IF;\s+DBMS_SESSION\.SLEEP\(0\.1\);", block + ) + assert re.search(r"EXECUTE IMMEDIATE 'CREATE[^;]*;\s+RETURN;", block) + + class _Failing: """A provider whose connections fail each statement with the next of `failures`, then return no rows."""