Fixes neo4j/apoc#126: apoc.periodic.submit fails with schema operatio…

…ns (extended) (#3211) * Fixes neo4j/apoc#126: apoc.periodic.submit fails with schema operations (extended) * added test * updated extended.txt
neo4j-contrib · Mar 28, 2023 · 4e99124 · 4e99124
1 parent 63c2c99
commit 4e99124
Show file tree

Hide file tree

Showing 3 changed files with 111 additions and 0 deletions.
diff --git a/extended/src/main/java/apoc/periodic/PeriodicExtended.java b/extended/src/main/java/apoc/periodic/PeriodicExtended.java
@@ -0,0 +1,46 @@
+package apoc.periodic;
+
+import apoc.Description;
+import apoc.Extended;
+import apoc.Pools;
+import apoc.util.Util;
+import org.neo4j.graphdb.GraphDatabaseService;
+import org.neo4j.logging.Log;
+import org.neo4j.procedure.Context;
+import org.neo4j.procedure.Mode;
+import org.neo4j.procedure.Name;
+import org.neo4j.procedure.Procedure;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.neo4j.graphdb.QueryExecutionType.QueryType.READ_ONLY;
+import static org.neo4j.graphdb.QueryExecutionType.QueryType.READ_WRITE;
+import static org.neo4j.graphdb.QueryExecutionType.QueryType.WRITE;
+import static org.neo4j.graphdb.QueryExecutionType.QueryType.SCHEMA_WRITE;
+
+@Extended
+public class PeriodicExtended {
+
+    @Context
+    public GraphDatabaseService db;
+
+    @Context
+    public Log log;
+
+    @Context
+    public Pools pools;
+
+    @Procedure(mode = Mode.SCHEMA)
+    @Description("apoc.periodic.submitSchema(name, statement, $config) - equivalent to apoc.periodic.submit which can also accept schema operations")
+    public Stream<PeriodicUtils.JobInfo> submitSchema(@Name("name") String name, @Name("statement") String statement, @Name(value = "params", defaultValue = "{}") Map<String,Object> config) {
+        validateQuery(statement);
+        return PeriodicUtils.submitProc(name, statement, config, db, log, pools);
+    }
+
+    private void validateQuery(String statement) {
+        Util.validateQuery(db, statement,
+                READ_ONLY, WRITE, READ_WRITE, SCHEMA_WRITE);
+    }
+
+}
diff --git a/extended/src/main/resources/extended.txt b/extended/src/main/resources/extended.txt
@@ -101,6 +101,7 @@ apoc.nlp.gcp.classify.graph
 apoc.nlp.gcp.classify.stream
 apoc.nlp.gcp.entities.graph
 apoc.nlp.gcp.entities.stream
+apoc.periodic.submitSchema
 apoc.redis.append
 apoc.redis.configGet
 apoc.redis.configSet

diff --git a/extended/src/test/java/apoc/periodic/PeriodicExtendedTest.java b/extended/src/test/java/apoc/periodic/PeriodicExtendedTest.java
@@ -0,0 +1,64 @@
+package apoc.periodic;
+
+import apoc.util.TestUtil;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.neo4j.test.rule.DbmsRule;
+import org.neo4j.test.rule.ImpermanentDbmsRule;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import static apoc.util.TestUtil.testCall;
+import static org.junit.Assert.assertEquals;
+import static org.neo4j.test.assertion.Assert.assertEventually;
+
+public class PeriodicExtendedTest {
+
+    @Rule
+    public DbmsRule db = new ImpermanentDbmsRule();
+
+    @Before
+    public void initDb() {
+        TestUtil.registerProcedure(db, PeriodicExtended.class, Periodic.class);
+    }
+
+    @Test
+    public void testSubmitSchema() {
+        testCall(db, "CALL apoc.periodic.submitSchema('subSchema','CREATE INDEX periodicIdx FOR (n:Bar) ON (n.first_name, n.last_name)')",
+                (row) -> {
+                    assertEquals("subSchema", row.get("name"));
+                    assertEquals(false, row.get("done"));
+                });
+
+        assertEventually(() -> db.executeTransactionally("SHOW INDEXES YIELD name WHERE name = 'periodicIdx' RETURN count(*) AS count",
+                        Collections.emptyMap(),
+                        (res) -> res.<Long>columnAs("count").next()),
+                val -> val == 1L, 15L, TimeUnit.SECONDS);
+
+        testCall(db, "CALL apoc.periodic.list()", (row) -> {
+            assertEquals("subSchema", row.get("name"));
+            assertEquals(true, row.get("done"));
+        });
+    }
+
+    @Test
+    public void testSubmitSchemaWithWriteOperation() {
+        testCall(db, "CALL apoc.periodic.submitSchema('subSchema','CREATE (:SchemaLabel)')",
+                (row) -> {
+                    assertEquals("subSchema", row.get("name"));
+                    assertEquals(false, row.get("done"));
+                });
+
+        assertEventually(() -> db.executeTransactionally("MATCH (n:SchemaLabel) RETURN count(n) AS count",
+                        Collections.emptyMap(),
+                        (res) -> res.<Long>columnAs("count").next()),
+                val -> val == 1L, 15L, TimeUnit.SECONDS);
+
+        testCall(db, "CALL apoc.periodic.list()", (row) -> {
+            assertEquals("subSchema", row.get("name"));
+            assertEquals(true, row.get("done"));
+        });
+    }
+}