From 90120b77eec9c77f71855786e41e6837f47fd4fc Mon Sep 17 00:00:00 2001 From: xiaowu297 <155960537+xiaowu297@users.noreply.github.com> Date: Mon, 23 Dec 2024 16:36:46 +0800 Subject: [PATCH 1/7] Update LogarithmicFunctions.scala MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 增加一个截断函数，默认保留整数，提供可选参数指定截断到的小数位数 --- .../org/grapheco/lynx/procedure/LogarithmicFunctions.scala | 5 +++++ 1 file changed, 5 insertions(+) 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 + } } From b324d527f03fcbfc97ecbd7a13b518f5435de85d Mon Sep 17 00:00:00 2001 From: xiaowu297 <155960537+xiaowu297@users.noreply.github.com> Date: Mon, 23 Dec 2024 16:42:19 +0800 Subject: [PATCH 2/7] Update LogarithmicFunctions.scala MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上一个描述错误，应该是power不应该放在时间类函数下 From 4bce81825b7845bdfff8cb30b9b08d659bec5a54 Mon Sep 17 00:00:00 2001 From: xiaowu297 <155960537+xiaowu297@users.noreply.github.com> Date: Mon, 23 Dec 2024 17:53:41 +0800 Subject: [PATCH 3/7] Update ApplyPushDownRule.scala MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 添加了 if apply.right.isEmpty 条件，直接判断 Apply 节点是否无效，简化了移除逻辑，同时修改了val returnItemNames = A.schema.map(_._1).toSet，因为如果 A.schema 包含大量字段，Set 的查找性能更优。 --- .../org/grapheco/lynx/optimizer/ApplyPushDownRule.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 From 7eda841961f22d25c30609eca2d868ba9176b007 Mon Sep 17 00:00:00 2001 From: xiaowu297 <155960537+xiaowu297@users.noreply.github.com> Date: Mon, 23 Dec 2024 17:54:36 +0800 Subject: [PATCH 4/7] Update JoinTableSizeEstimateRule.scala MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 具体优化内容： （1）apply方法:在apply方法内模式匹配方式不够简洁明了，后续如果需要扩展对其他类型节点的判断逻辑，容易使match代码块变得臃肿,因此新增了两个私有辅助方法，让apply方法内的模式匹配逻辑更清晰。 （2）estimateNodeRow方法：代码中有较多的嵌套match和比较复杂的类型判断及取值操作，使用collect方法来更简洁地处理属性提取部分，同时使用flatten方法将可能嵌套的结果扁平化处理。 （3）joinRecursion方法：代码中对于parent.children.head和parent.children.last的处理逻辑基本重复，,对此提取出了一个私有方法，将获取基础表的公共操作封装在一个私有方法。 --- .../optimizer/JoinTableSizeEstimateRule.scala | 85 ++++++++----------- 1 file changed, 35 insertions(+), 50 deletions(-) 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 } } From fd309329e1e8316738668fd44ac9b2649f303c61 Mon Sep 17 00:00:00 2001 From: xiaowu297 <155960537+xiaowu297@users.noreply.github.com> Date: Mon, 23 Dec 2024 17:55:25 +0800 Subject: [PATCH 5/7] Update LynxRecord.scala MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 具体优化内容： （1）类型转换安全性：map(_.asInstanceOf[LynxX]) 中未进行类型检查，若类型不匹配，程序会抛出 异常，因此使用模式匹配方法，避免程序崩溃。 （2）toMap方法：使用 keys.zip(values)，若 cols.keys 和 values 的长度不一致，会导致抛出异常或数据丢失，因此修改为根据 cols 的映射来生成键值对。 （3）增加数据类型支持：除去文中提到的数据类型，还增加了date以及list两种数据类型供使用。 （4）去除列名大小写干扰：有时列名的大小写可能不一致，因此提供了一个大小写无关的访问选项getIgnoreCase（）。 （5）增加日志记录logger：当无法找到列名或发生错误时，记录日志以便排查问题。 --- .../scala/org/grapheco/lynx/LynxRecord.scala | 58 ++++++++++++++++--- 1 file changed, 50 insertions(+), 8 deletions(-) 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) +} } From 3fd45d933f1c8834ca64897601681062c3362918 Mon Sep 17 00:00:00 2001 From: xiaowu297 <155960537+xiaowu297@users.noreply.github.com> Date: Mon, 23 Dec 2024 17:56:45 +0800 Subject: [PATCH 6/7] Update NumericFunctions.scala MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 增加了对数据的截断函数trunc，支持可选参数指定保留的小数位数，默认截断到整数位 --- .../grapheco/lynx/procedure/NumericFunctions.scala | 11 +++++++++++ 1 file changed, 11 insertions(+) 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()) + } + } } From 0f2769fbf017a443d2d15f8b13f763c87b4092ca Mon Sep 17 00:00:00 2001 From: xiaowu297 <155960537+xiaowu297@users.noreply.github.com> Date: Mon, 23 Dec 2024 17:57:23 +0800 Subject: [PATCH 7/7] Update TrigonometricFunctions.scala MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 增加了弧度转化为角度的函数degrees；增加了角度转化为弧度的函数radians --- .../lynx/procedure/TrigonometricFunctions.scala | 10 ++++++++++ 1 file changed, 10 insertions(+) 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()) + } }