More analysis of backward chaining best practices.

Author: LorenzBuehmann

sansa-inference-common/pom.xml

@@ -97,8 +97,8 @@
   </dependencies>
 
   <build>
-    <sourceDirectory>src/main/scala,src/main/resources</sourceDirectory>
-    <testSourceDirectory>src/test/scala</testSourceDirectory>
+    <!--<sourceDirectory>src/main/scala,src/main/resources</sourceDirectory>-->
+    <!--<testSourceDirectory>src/test/scala</testSourceDirectory>-->
     <!--<outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>-->
     <!--<testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory>-->
     <plugins>

sansa-inference-common/src/main/scala/net/sansa_stack/inference/rules/plan/SimpleSQLGenerator.scala

@@ -22,7 +22,7 @@ class SimpleSQLGenerator(val sqlSchema: SQLSchema = SQLSchemaDefault) extends SQ
   var idx = 0
 
   def generateSQLQuery(rule: Rule): String = {
-    info(s"Rule:\n$rule")
+    debug(s"Rule:\n$rule")
 
     reset()
 

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/backwardchaining/BackwardChainingReasonerDataframe.scala

@@ -1,18 +1,19 @@
 package net.sansa_stack.inference.spark.backwardchaining
 
-import net.sansa_stack.inference.rules.RuleSets
-import net.sansa_stack.inference.rules.plan.{SimplePlanGenerator, SimpleSQLGenerator, TriplesSchema}
-import net.sansa_stack.inference.spark.backwardchaining.tree.{AndNode, OrNode}
-import net.sansa_stack.inference.spark.data.loader.RDFGraphLoader
-import net.sansa_stack.inference.utils.{Logging, TripleUtils}
-import net.sansa_stack.inference.utils.RuleUtils._
-import org.apache.jena.graph.{NodeFactory, Triple}
+import org.apache.jena.graph.{Node, NodeFactory, Triple}
 import org.apache.jena.reasoner.TriplePattern
 import org.apache.jena.reasoner.rulesys.Rule
 import org.apache.jena.reasoner.rulesys.impl.BindingVector
-import org.apache.jena.vocabulary.RDF
+import org.apache.jena.sparql.util.FmtUtils
+import org.apache.jena.vocabulary.{RDF, RDFS}
 import org.apache.spark.sql.{Dataset, SparkSession}
-import net.sansa_stack.inference.utils.TripleUtils._
+
+import net.sansa_stack.inference.rules.RuleSets
+import net.sansa_stack.inference.rules.plan.SimpleSQLGenerator
+import net.sansa_stack.inference.spark.backwardchaining.tree.{AndNode, OrNode}
+import net.sansa_stack.inference.spark.data.loader.RDFGraphLoader
+import net.sansa_stack.inference.utils.RuleUtils._
+import net.sansa_stack.inference.utils.{Logging, TripleUtils}
 
 
 //case class RDFTriple(s: Node, p: Node, o: Node)
@@ -26,24 +27,31 @@ class BackwardChainingReasonerDataframe(
                                          val rules: Set[Rule],
                                          val graph: Dataset[RDFTriple]) extends Logging {
 
+  import org.apache.spark.sql.functions._
+
+  val precomputeSchema: Boolean = true
+
+  var schema: Map[Node, Dataset[RDFTriple]] = Map()
+
   def isEntailed(triple: Triple): Boolean = {
     isEntailed(new TriplePattern(triple))
   }
 
   def isEntailed(tp: TriplePattern): Boolean = {
 
+    if (precomputeSchema) schema = extractWithIndex(graph)
+
     val tree = buildTree(new AndNode(tp), Seq())
     println(tree.toString)
 
     val triples = processTree(tree)
     triples.explain(true)
-    println(triples.count())
+    println(triples.distinct().count())
 
     false
   }
 
-  import org.apache.spark.sql.functions._
-  val planGenerator = new SimplePlanGenerator(TriplesSchema.get())
+
 
   private def processTree(tree: AndNode): Dataset[RDFTriple] = {
     // 1. look for asserted triples in the graph
@@ -58,10 +66,11 @@ class BackwardChainingReasonerDataframe(
       // first process the children, i.e. we get the data for each triple pattern in the body of the rule
       val childrenTriples: Seq[Dataset[RDFTriple]] = child.children.map(processTree(_))
 
-      val union = childrenTriples.reduce(_ union _)
+      val baseTriples = if (childrenTriples.size > 1) childrenTriples.reduce(_ union _) else childrenTriples.head
+
 
       // then apply the rule on the UNION of the children data
-      applyRule(child.element, union)
+      applyRule(child.element, baseTriples)
     })
 
     var triples = assertedTriples
@@ -72,14 +81,25 @@ class BackwardChainingReasonerDataframe(
   }
 
   private def lookup(tp: TriplePattern): Dataset[RDFTriple] = {
+
+    val terminological = TripleUtils.isTerminological(tp.asTriple())
+
+    var filteredGraph =
+      if (terminological) {
+        schema.getOrElse(tp.getPredicate, graph)
+      } else {
+        graph
+      }
+
+    info(s"Lookup data for $tp")
     val s = tp.getSubject.toString()
     val p = tp.getPredicate.toString()
     val o = tp.getObject.toString()
-    var filteredGraph = graph
+
     if(tp.getSubject.isConcrete) {
       filteredGraph = filteredGraph.filter(t => t.s.equals(s))
     }
-    if(tp.getPredicate.isConcrete) {
+    if(!terminological && tp.getPredicate.isConcrete) {
       filteredGraph = filteredGraph.filter(t => t.p.equals(p))
     }
     if(tp.getObject.isConcrete) {
@@ -147,7 +167,7 @@ class BackwardChainingReasonerDataframe(
   private def applyRule(rule: Rule, dataset: Dataset[RDFTriple]): Dataset[RDFTriple] = {
     // convert to SQL
     val sqlGenerator = new SimpleSQLGenerator()
-    val sql = sqlGenerator.generateSQLQuery(rule)
+    var sql = sqlGenerator.generateSQLQuery(rule)
 //    val sql =
 //      """
 //        |SELECT rel0.s, 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type' AS p, 'http://swat.cse.lehigh.edu/onto/univ-bench.owl#Person' AS o
@@ -156,12 +176,49 @@ class BackwardChainingReasonerDataframe(
 //      """.stripMargin
 
     // generate logical execution plan
-    val planGenerator = new SimplePlanGenerator(TriplesSchema.get())
-    val plan = planGenerator.generateLogicalPlan(rule)
+//    val planGenerator = new SimplePlanGenerator(TriplesSchema.get())
+//    val plan = planGenerator.generateLogicalPlan(rule)
 
+    val tableName = s"TRIPLES_${rule.getName}"
+    sql = sql.replace("TRIPLES", tableName)
+    println(s"SQL NEW: $sql")
+    dataset.createOrReplaceTempView(tableName)
     dataset.sparkSession.sql(sql).as[RDFTriple]
   }
 
+  val properties = Set(RDFS.subClassOf, RDFS.subPropertyOf, RDFS.domain, RDFS.range).map(p => p.asNode())
+  val DUMMY_VAR = NodeFactory.createVariable("VAR");
+
+  /**
+    * Computes the triples for each schema property p, e.g. `rdfs:subClassOf` and returns it as mapping from p
+    * to the [[Dataset]] containing the triples.
+    *
+    * @param graph the RDF graph
+    * @return a mapping from the corresponding schema property to the Dataframe of s-o pairs
+    */
+  def extractWithIndex(graph: Dataset[RDFTriple]): Map[Node, Dataset[RDFTriple]] = {
+    log.info("Started schema extraction...")
+
+    // for each schema property p
+    val index =
+      properties.map { p =>
+        // get triples (s, p, o)
+        var triples = lookup(new TriplePattern(DUMMY_VAR, p, DUMMY_VAR))
+
+        // broadcast the triples
+        triples = broadcast(triples)
+
+        // register as a table
+        triples.createOrReplaceTempView(FmtUtils.stringForNode(p).replace(":", "_"))
+
+        // add to index
+        (p -> triples)
+      }
+    log.info("Finished schema extraction.")
+
+    index.toMap
+  }
+
 
 }
 
@@ -182,6 +239,7 @@ object BackwardChainingReasonerDataframe {
       .config("spark.default.parallelism", parallelism)
       .config("spark.ui.showConsoleProgress", "false")
       .config("spark.sql.shuffle.partitions", parallelism)
+      .config("spark.sql.autoBroadcastJoinThreshold", "10485760")
       .getOrCreate()
 
     import session.implicits._
@@ -192,9 +250,32 @@ object BackwardChainingReasonerDataframe {
       .triples.map(t => RDFTriple(t.getSubject.toString(), t.getPredicate.toString(), t.getObject.toString()))
     val graph = session.createDataset(triples).cache()
     graph.createOrReplaceTempView("TRIPLES")
+    import org.apache.spark.sql.functions._
+    val domain = graph.filter(t => t.p == RDFS.domain.toString)
+    broadcast(domain).createOrReplaceTempView("DOMAIN")
+
+    val query =
+      """
+        |SELECT rel0.s AS s, 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type' AS p, 'http://swat.cse.lehigh.edu/onto/univ-bench.owl#Person' AS o
+        | FROM DOMAIN rel1 JOIN TRIPLES rel0 ON rel1.s=rel0.p
+        | WHERE rel1.o='http://swat.cse.lehigh.edu/onto/univ-bench.owl#Person' AND rel1.p='http://www.w3.org/2000/01/rdf-schema#domain'
+        | UNION
+        | SELECT *
+        | FROM TRIPLES
+        | WHERE p ='http://www.w3.org/1999/02/22-rdf-syntax-ns#type' AND o ='http://swat.cse.lehigh.edu/onto/univ-bench.owl#Person'
+      """.stripMargin
+
+    val ds = session.sql(query)
+    ds.explain()
+    println(ds.distinct().count())
+
 
     val rules = RuleSets.RDFS_SIMPLE
-      .filter(r => Seq("rdfs2", "rdfs3").contains(r.getName))
+      .filter(r => Seq(
+        "rdfs2"
+//        , "rdfs3"
+//        , "rdfs9"
+      ).contains(r.getName))
 
     val tp = Triple.create(
       NodeFactory.createVariable("s"),
@@ -206,6 +287,8 @@ object BackwardChainingReasonerDataframe {
     println(reasoner.isEntailed(tp))
 
     session.stop()
+
+
   }
 }
 

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/backwardchaining/tree/AndOrTree.scala

@@ -18,9 +18,11 @@ abstract class Node[T, C <: Node[_, _]](val element: T, var children: Seq[C] = S
   override def toString: String = print(0)
 
   def print(indent: Int): String = {
-    renderElement() + "\n" + children.map(c => "---" * indent + c.print(indent + 1)).mkString("\n")
+    indentS(renderElement(), indent) + "\n" + children.map(c => "---" * indent + c.print(indent + 1)).mkString("\n")
   }
 
+  def indentS(s: String, i: Int): String = "---" * i + s
+
   def renderElement(): String = element.toString
 }