Key type wrapper.

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/utils/NodeKey.scala

@@ -0,0 +1,35 @@
+package net.sansa_stack.inference.flink.utils
+
+import org.apache.jena.graph.Node
+import org.apache.jena.sparql.util.NodeComparator
+
+/**
+  * Key type wrapper for Jena `Node` objects.
+  * It basically makes Node comparable which is necessary to be handles as key in Flink.
+  *
+  * @author Lorenz Buehmann
+  */
+class NodeKey(val node: Node) extends Comparable[NodeKey] with Equals {
+
+  override def compareTo(o: NodeKey): Int = {
+    val other = o.node
+    if (node == null)
+      if (other == null) 0 else -1
+    else
+    if (other == null) 1 else new NodeComparator().compare(node, other)
+  }
+
+  override def canEqual(that: Any): Boolean = that.isInstanceOf[NodeKey]
+
+  override def hashCode(): Int = 31 * node.##
+
+  override def equals(that: Any): Boolean =
+    that match {
+      case key: NodeKey => (this eq key) || (key.canEqual(this) && hashCode == key.hashCode)
+      case _ => false
+    }
+}
+
+object NodeKey {
+  def apply(node: Node): NodeKey = new NodeKey(node)
+}

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/utils/key/Key.scala

@@ -0,0 +1,18 @@
+package net.sansa_stack.inference.flink.utils.key
+
+/**
+  * Base of a tuple-like generic key.
+  *
+  * @tparam T The type of the concrete key type.
+  */
+abstract class Key[T <: Key[T]] extends Comparable[T] {
+  /**
+    * Gets the i-th element of the tuple-like key.
+    *
+    * @param pos The position.
+    * @return The element at that key position;
+    */
+  def get(pos: Int): Any
+}
+
+

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/utils/key/Key1.scala

@@ -0,0 +1,36 @@
+package net.sansa_stack.inference.flink.utils.key
+
+/**
+  * A key with one key field.
+  *
+  * @tparam T1 The type of the field.
+  */
+class Key1[T1 <: Comparable[T1]](val value1: T1) extends Key[Key1[T1]] with Equals {
+
+  def get(pos: Int): Any = pos match {
+    case 0 =>
+      value1
+    case _ =>
+      throw new IndexOutOfBoundsException
+  }
+
+  override def hashCode: Int = if (value1 == null) 0 else value1.hashCode
+
+  override def canEqual(that: Any): Boolean = that.isInstanceOf[Key1[T1]]
+
+  override def equals(obj: Any): Boolean =
+    obj match {
+      case that: Key1[T1] => (this eq that) || (this.canEqual(that) && (value1 == that.value1))
+      case _ => false
+    }
+
+  override def toString: String = s"Key1 ($value1)"
+
+  def compareTo(o: Key1[T1]): Int = {
+    val other = o.value1
+    if (value1 == null)
+      if (other == null) 0 else -1
+    else
+      if (other == null) 1 else value1.compareTo(other)
+  }
+}

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/utils/key/Key2.scala

@@ -0,0 +1,53 @@
+package net.sansa_stack.inference.flink.utils.key
+
+/**
+  * A key with two key fields.
+  *
+  * @tparam T1 The type of the first field.
+  * @tparam T2 The type of the second field.
+  */
+class Key2[T1 <: Comparable[T1], T2 <: Comparable[T2]](val value1: T1, val value2: T2)
+  extends Key[Key2[T1, T2]]
+    with Equals {
+
+  def get(pos: Int): Any = pos match {
+    case 0 =>
+      value1
+    case 1 =>
+      value2
+    case _ =>
+      throw new IndexOutOfBoundsException
+  }
+
+  override def hashCode: Int = {
+    val c1: Int = if (value1 == null) 0 else value1.hashCode
+    val c2: Int = if (value2 == null) 0 else value2.hashCode
+    c1 * 17 + c2 * 31
+  }
+
+  override def canEqual(that: Any): Boolean = that.isInstanceOf[Key2[T1, T2]]
+
+  override def equals(obj: Any): Boolean =
+    obj match {
+      case that: Key2[T1, T2] => (this eq that) || (this.canEqual(that) && (value1 == that.value1) && (value2 == that.value2))
+      case _ => false
+    }
+
+  override def toString: String = s"Key2 ($value1, $value2)"
+
+  def compareTo(o: Key2[T1, T2]): Int = {
+    val other1 = o.value1
+    val other2 = o.value2
+
+    val c1 = if (value1 == null)
+      if (other1 == null) 0 else -1
+        else
+      if (other1 == null) 1 else value1.compareTo(other1)
+
+    if(c1 != 0) c1 else
+      if (value2 == null)
+        if (other2 == null) 0 else -1
+      else
+      if (other2 == null) 1 else value2.compareTo(other2)
+  }
+}

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/utils/key/Key3.scala

@@ -0,0 +1,69 @@
+package net.sansa_stack.inference.flink.utils.key
+
+/**
+  * A key with two key fields.
+  *
+  * @tparam T1 The type of the first field.
+  * @tparam T2 The type of the second field.
+  * @tparam T3 The type of the third field.
+  */
+class Key3[T1 <: Comparable[T1], T2 <: Comparable[T2], T3 <: Comparable[T3]](val value1: T1, val value2: T2, val value3: T3)
+  extends Key[Key3[T1, T2, T3]]
+    with Equals {
+
+  def get(pos: Int): Any = pos match {
+    case 0 =>
+      value1
+    case 1 =>
+      value2
+    case 2 =>
+      value3
+    case _ =>
+      throw new IndexOutOfBoundsException
+  }
+
+  override def hashCode: Int = {
+    val c1: Int = if (value1 == null) 0 else value1.hashCode
+    val c2: Int = if (value2 == null) 0 else value2.hashCode
+    val c3: Int = if (value3 == null) 0 else value3.hashCode
+    c1 * 17 + c2 * 31 + c3 * 47
+  }
+
+  override def canEqual(that: Any): Boolean = that.isInstanceOf[Key3[T1, T2, T3]]
+
+  override def equals(obj: Any): Boolean =
+    obj match {
+      case that: Key3[T1, T2, T3] =>
+        (this eq that) || (this.canEqual(that) && (value1 == that.value1) && (value2 == that.value2) && (value3 == that.value3))
+      case _ => false
+    }
+
+  override def toString: String = s"Key3 ($value1, $value2, $value3)"
+
+  def compareTo(o: Key3[T1, T2, T3]): Int = {
+    val other1 = o.value1
+    val other2 = o.value2
+    val other3 = o.value3
+
+    val c1 = if (value1 == null)
+      if (other1 == null) 0 else -1
+    else
+      if (other1 == null) 1 else value1.compareTo(other1)
+
+    if(c1 != 0) c1 else {
+      val c2 = if (value2 == null)
+        if (other2 == null) 0 else -1
+      else
+        if (other2 == null) 1 else value2.compareTo(other2)
+
+      if(c2 != 0) c2 else {
+        if (value3 == null)
+          if (other3 == null) 0 else -1
+        else
+          if (other3 == null) 1 else value3.compareTo(other3)
+      }
+
+    }
+
+  }
+}