feat: introduce global await for index creation (#790)

Right before data is being written and right after scripts have been invoked we now  call
cypher procedure "db.awaitIndexes".

This ensures that any custom indexes created from custom scripts will have leeway room to
finish construction before we write data.

Can be tweaked using option "index.await.timeout". Default 300, set to 0 for old behaviour
i.e. no wait.

Author: nord-erik

common/src/main/scala/org/neo4j/spark/util/Neo4jOptions.scala

@@ -134,6 +134,8 @@ class Neo4jOptions(private val options: java.util.Map[String, String]) extends S
     )
   }
 
+  val indexAwait = getParameter(INDEX_AWAIT_TIMEOUT_SEC, DEFAULT_INDEX_AWAIT_TIMEOUT_SEC.toString).toInt
+
   val query: Neo4jQueryOptions = (
     getParameter(QUERY.toString.toLowerCase),
     getParameter(LABELS.toString.toLowerCase),
@@ -564,6 +566,9 @@ object Neo4jOptions {
   // map aggregation
   val SCHEMA_MAP_GROUP_DUPLICATE_KEYS = "schema.map.group.duplicate.keys"
 
+  // index options
+  val INDEX_AWAIT_TIMEOUT_SEC = "index.await.timeout"
+
   // partitions
   val PARTITIONS = "partitions"
 
@@ -651,6 +656,8 @@ object Neo4jOptions {
 
   val DEFAULT_MAP_GROUP_DUPLICATE_KEYS = false
 
+  val DEFAULT_INDEX_AWAIT_TIMEOUT_SEC = 300
+
   var DEFAULT_AUTH_PARAMETERS: Map[String, String] =
     Seq("username", "password", "ticket", "principal", "credentials", "realm", "scheme", "token")
       .map(name => name -> DEFAULT_EMPTY).toMap

spark-3/src/main/scala/org/neo4j/spark/writer/Neo4jBatchWriter.scala

@@ -43,6 +43,11 @@ class Neo4jBatchWriter(
     val scriptResult = schemaService.execute(neo4jOptions.script)
     schemaService.close()
 
+    if (neo4jOptions.indexAwait > 0) {
+      val session = driverCache.getOrCreate().session(neo4jOptions.session.toNeo4jSession())
+      session.run(s"CALL db.awaitIndexes(${neo4jOptions.indexAwait})").consume()
+    }
+
     new Neo4jDataWriterFactory(
       neo4j,
       jobId,