diff --git a/src/main/scala/org/grapheco/lynx/LynxRecord.scala b/src/main/scala/org/grapheco/lynx/LynxRecord.scala index ad0bad81..a994a612 100644 --- a/src/main/scala/org/grapheco/lynx/LynxRecord.scala +++ b/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) +} } diff --git a/src/main/scala/org/grapheco/lynx/optimizer/ApplyPushDownRule.scala b/src/main/scala/org/grapheco/lynx/optimizer/ApplyPushDownRule.scala index 2ec7fb0e..a0a38d4d 100644 --- a/src/main/scala/org/grapheco/lynx/optimizer/ApplyPushDownRule.scala +++ b/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 diff --git a/src/main/scala/org/grapheco/lynx/optimizer/JoinTableSizeEstimateRule.scala b/src/main/scala/org/grapheco/lynx/optimizer/JoinTableSizeEstimateRule.scala index 63cd9c37..0cc0bbea 100644 --- a/src/main/scala/org/grapheco/lynx/optimizer/JoinTableSizeEstimateRule.scala +++ b/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 } } diff --git a/src/main/scala/org/grapheco/lynx/procedure/LogarithmicFunctions.scala b/src/main/scala/org/grapheco/lynx/procedure/LogarithmicFunctions.scala index fbababfe..71ac48e2 100644 --- a/src/main/scala/org/grapheco/lynx/procedure/LogarithmicFunctions.scala +++ b/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 + } } diff --git a/src/main/scala/org/grapheco/lynx/procedure/NumericFunctions.scala b/src/main/scala/org/grapheco/lynx/procedure/NumericFunctions.scala index 8f6b31a3..0eb0baae 100644 --- a/src/main/scala/org/grapheco/lynx/procedure/NumericFunctions.scala +++ b/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()) + } + } } diff --git a/src/main/scala/org/grapheco/lynx/procedure/TrigonometricFunctions.scala b/src/main/scala/org/grapheco/lynx/procedure/TrigonometricFunctions.scala index a1ea9e0a..f84289bb 100644 --- a/src/main/scala/org/grapheco/lynx/procedure/TrigonometricFunctions.scala +++ b/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()) + } }