Optimized ApplyPushDownRule, LynxRecord and JoinTableSizeEstimateRule

src/main/scala/org/grapheco/lynx/LynxRecord.scala

@@ -3,25 +3,67 @@ package org.grapheco.lynx
 import org.grapheco.lynx.types.LynxValue
 import org.grapheco.lynx.types.property.{LynxBoolean, LynxFloat, LynxInteger, LynxNull, LynxString}
 import org.grapheco.lynx.types.structural.{LynxNode, LynxRelationship}
+import org.slf4j.LoggerFactory
+
+private val logger = LoggerFactory.getLogger(classOf[LynxRecord])
 
 case class LynxRecord(cols: Map[String, Int], values: Seq[LynxValue]){
   def apply(columnName: String): LynxValue = get(columnName).getOrElse(LynxNull)
 
-  def get(columnName: String): Option[LynxValue] = cols.get(columnName).map(values.apply)
+ def get(columnName: String): Option[LynxValue] = {
+  val result = cols.get(columnName).map(values.apply)
+  if (result.isEmpty) logger.debug(s"Column '$columnName' not found.")
+  result
+}
 
   def get(columnIndex: Int): Option[LynxValue] = values.lift(columnIndex)
 
-  def getAsString(columnName: String): Option[LynxString] = get(columnName).map(_.asInstanceOf[LynxString])
+def getAsString(columnName: String): Option[LynxString] = get(columnName) match {
+  case Some(value: LynxString) => Some(value)
+  case _ => None
+}
 
-  def getAsInt(columnName: String): Option[LynxInteger] = get(columnName).map(_.asInstanceOf[LynxInteger])
+def getAsDate(columnName: String): Option[LynxDate] = get(columnName) match {
+  case Some(value: LynxDate) => Some(value)
+  case _ => None
+}
 
-  def getAsDouble(columnName: String): Option[LynxFloat] = get(columnName).map(_.asInstanceOf[LynxFloat])
+def getAsList(columnName: String): Option[Seq[LynxValue]] = get(columnName) match {
+  case Some(value: LynxList) => Some(value.elements)
+  case _ => None
+}
 
-  def getAsBoolean(columnName: String): Option[LynxBoolean] = get(columnName).map(_.asInstanceOf[LynxBoolean])
+def getAsInt(columnName: String): Option[LynxInteger] = get(columnName) match {
+  case Some(value: LynxInteger) => Some(value)
+  case _ => None
+}
 
-  def getAsNode(columnName: String): Option[LynxNode] = get(columnName).map(_.asInstanceOf[LynxNode])
+  def getAsDouble(columnName: String): Option[LynxFloat] = get(columnName)  match {
+  case Some(value: LynxFloat) => Some(value)
+  case _ => None
+}
 
-  def getAsRelationship(columnName: String): Option[LynxRelationship] = get(columnName).map(_.asInstanceOf[LynxRelationship])
+  def getAsBoolean(columnName: String): Option[LynxBoolean] = get(columnName)  match {
+  case Some(value:LynxBoolean) => Some(value)
+  case _ => None
+}
 
-  def toMap: Map[String, LynxValue] = cols.keys.zip(values).toMap
+  def getAsNode(columnName: String): Option[LynxNode] = get(columnName)  match {
+  case Some(value:LynxNode) => Some(value)
+  case _ => None
+}
+
+  def getAsRelationship(columnName: String): Option[LynxRelationship] = get(columnName)  match {
+  case Some(value:LynxRelationship) => Some(value)
+  case _ => None
+}
+
+def getIgnoreCase(columnName: String): Option[LynxValue] = {
+  cols.find { case (key, _) => key.equalsIgnoreCase(columnName) }
+      .flatMap { case (_, index) => values.lift(index) }
+}
+
+  def toMap: Map[String, LynxValue] = cols.map { case (key, index) =>
+  key -> values.lift(index).getOrElse(LynxNull)
+}
 }

src/main/scala/org/grapheco/lynx/optimizer/ApplyPushDownRule.scala

@@ -19,7 +19,8 @@ object ApplyPushDownRule extends PhysicalPlanOptimizerRule {
       ║
      [A]
      */
-    case apply: Apply => apply
+    // case apply: Apply => apply
+    case apply: Apply if apply.right.isEmpty => apply.left
   }
 
   private val APPLY_PUSH_DOWN: PartialFunction[PhysicalPlan, PhysicalPlan] = {
@@ -38,7 +39,8 @@ object ApplyPushDownRule extends PhysicalPlanOptimizerRule {
     case apply:Apply =>
       val A = apply.left
       val B = apply.right
-      val returnItemNames = A.get.schema.map(_._1)
+      // val returnItemNames = A.get.schema.map(_._1)
+      val returnItemNames = A.get.schema.map(_._1).toSet
 
       while (apply.right.isDefined
 //        && apply.right.get.children.length==1

src/main/scala/org/grapheco/lynx/optimizer/JoinTableSizeEstimateRule.scala

@@ -15,44 +15,38 @@ object JoinTableSizeEstimateRule extends PhysicalPlanOptimizerRule {
 
   override def apply(plan: PhysicalPlan, ppc: PhysicalPlannerContext): PhysicalPlan = optimizeBottomUp(plan,
     {
-      case pnode: PhysicalPlan => {
-        pnode.children match {
-          case Seq(pj@Join(filterExpr, isSingleMatch, joinType)) => {
-            val res = joinRecursion(pj, ppc, isSingleMatch)
-            pnode.withChildren(Seq(res))
-          }
-          case _ => pnode
-        }
+      case pnode: PhysicalPlan if isJoinNode(pnode) => {
+        val joinedPlan = joinRecursion(pnode.asInstanceOf[Join], ppc, getIsSingleMatch(pnode))
+        pnode.withChildren(Seq(joinedPlan))
       }
+      case pnode => pnode
     }
   )
 
+  private def isJoinNode(plan: PhysicalPlan): Boolean = plan match {
+    case Join(_, _, _) => true
+    case _ => false
+  }
+
+  private def getIsSingleMatch(plan: PhysicalPlan): Boolean = plan match {
+    case Join(_, isSingleMatch, _) => isSingleMatch
+    case _ => false
+  }
+
   def estimateNodeRow(pattern: NodePattern, graphModel: GraphModel): Long = {
-    val countMap = mutable.Map[String, Long]()
-    val labels = pattern.labels.map(l => l.name)
-    val prop = pattern.properties.map({
-      case MapExpression(items) => {
-        items.map(
-          p => {
-            p._2 match {
-              case b: Literal => (p._1.name, b.value)
-              case _ => (p._1.name, null)
-            }
-          }
-        )
+    val labels = pattern.labels.map(_.name)
+    val prop = pattern.properties.collect {
+      case MapExpression(items) => items.map {
+        case p if p._2.isInstanceOf[Literal] => (p._1.name, p._2.value)
+        case _ => (p._1.name, null)
       }
-      case _ => return 0
-    })
+    }.flatten
 
     if (labels.nonEmpty) {
-      val minLabelAndCount = labels.map(label => (label, graphModel._helper.estimateNodeLabel(label))).minBy(f => f._2)
-
-      if (prop.isDefined) {
-        prop.get.map(f => graphModel._helper.estimateNodeProperty(minLabelAndCount._1, f._1, f._2)).min
-      }
-      else minLabelAndCount._2
-    }
-    else graphModel.statistics.numNode
+      val minLabelAndCount = labels.map(label => (label, graphModel._helper.estimateNodeLabel(label))).minBy(_._2)
+      val estimatedPropCount = if (prop.nonEmpty) prop.map(f => graphModel._helper.estimateNodeProperty(minLabelAndCount._1, f._1, f._2)).min else minLabelAndCount._2
+      estimatedPropCount
+    } else graphModel.statistics.numNode
   }
 
   def estimateRelationshipRow(rel: RelationshipPattern, left: NodePattern, right: NodePattern, graphModel: GraphModel): Long = {
@@ -75,30 +69,21 @@ object JoinTableSizeEstimateRule extends PhysicalPlanOptimizerRule {
   }
 
   def joinRecursion(parent: Join, ppc: PhysicalPlannerContext, isSingleMatch: Boolean): PhysicalPlan = {
-    val t1 = parent.children.head
-    val t2 = parent.children.last
-
-    val table1 = t1 match {
-      case pj@Join(filterExpr, isSingleMatch, joinType) => joinRecursion(pj, ppc, isSingleMatch)
-      case pm@Merge(mergeSchema, mergeOps, onMatch, onCreate) => {
-        val res = joinRecursion(pm.children.head.asInstanceOf[Join], ppc, isSingleMatch)
-        pm.withChildren(Seq(res))
-      }
-      case _ => t1
-    }
-    val table2 = t2 match {
-      case pj@Join(filterExpr, isSingleMatch, joinType) => joinRecursion(pj, ppc, isSingleMatch)
-      case pm@Merge(mergeSchema, mergeOps, onMatch, onCreate) => {
-        val res = joinRecursion(pm.children.head.asInstanceOf[Join], ppc, isSingleMatch)
-        pm.withChildren(Seq(res))
-      }
-      case _ => t2
-    }
+    val table1 = getBaseTable(parent.children.head, ppc, isSingleMatch)
+    val table2 = getBaseTable(parent.children.last, ppc, isSingleMatch)
 
     if ((table1.isInstanceOf[NodeScan] || table1.isInstanceOf[RelationshipScan])
       && (table2.isInstanceOf[NodeScan] || table2.isInstanceOf[RelationshipScan])) {
       estimateTableSize(parent, table1, table2, ppc)
+    } else Join(parent.filterExpr, parent.isSingleMatch, parent.joinType)(table1, table2, ppc)
+  }
+
+  private def getBaseTable(plan: PhysicalPlan, ppc: PhysicalPlannerContext, isSingleMatch: Boolean): PhysicalPlan = plan match {
+    case pj@Join(_, _, _) => joinRecursion(pj, ppc, isSingleMatch)
+    case pm@Merge(mergeSchema, mergeOps, onMatch, onCreate) => {
+      val res = joinRecursion(pm.children.head.asInstanceOf[Join], ppc, isSingleMatch)
+      pm.withChildren(Seq(res))
     }
-    else Join(parent.filterExpr, parent.isSingleMatch, parent.joinType)(table1, table2, ppc)
+    case other => other
   }
 }

src/main/scala/org/grapheco/lynx/procedure/LogarithmicFunctions.scala

@@ -41,4 +41,9 @@ class LogarithmicFunctions {
   def sqrt(x: LynxNumber): Double = {
     math.sqrt(x.number.doubleValue())
   }
+
+  @LynxProcedure(name = "power")
+  def power(x: LynxInteger, n: LynxInteger): Int = {
+    math.pow(x.value, n.value).toInt
+  }
 }

src/main/scala/org/grapheco/lynx/procedure/NumericFunctions.scala

@@ -50,4 +50,15 @@ class NumericFunctions {
   def sign(x: LynxNumber): Double = {
     math.signum(x.number.doubleValue())
   }
+
+  @LynxProcedure(name = "trunc")
+  def trunc(x: LynxNumber, precision: Option[LynxNumber] = None): Double = {
+    precision match {
+      case Some(p) =>
+        val scale = math.pow(10, p.number.doubleValue()).toDouble
+        math.floor(x.number.doubleValue() * scale) / scale
+      case None =>
+        if (x.number.doubleValue() >= 0) math.floor(x.number.doubleValue()) else math.ceil(x.number.doubleValue())
+    }
+  }
 }

src/main/scala/org/grapheco/lynx/procedure/TrigonometricFunctions.scala

@@ -87,4 +87,14 @@ class TrigonometricFunctions {
   def tanh(x: LynxNumber): Double = {
     math.tanh(x.number.doubleValue())
   }
+
+  @LynxProcedure(name = "degrees")
+  def degrees(x: LynxNumber): Double = {
+    math.toDegrees(x.number.doubleValue())
+  }
+
+  @LynxProcedure(name = "radians")
+  def radians(x: LynxNumber): Double = {
+    math.toRadians(x.number.doubleValue())
+  }
 }