Skip to content

优化了一些常见的DataFrame操作方法；优化了过滤和匹配图数据库中节点的方法 #148

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
seer-xjc wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

优化了一些常见的DataFrame操作方法；优化了过滤和匹配图数据库中节点的方法 #148

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 34 additions & 35 deletions src/main/scala/org/grapheco/lynx/dataframe/DefaultDataFrameOperator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,17 @@ import org.opencypher.v9_0.expressions.Expression
*/
class DefaultDataFrameOperator(expressionEvaluator: ExpressionEvaluator) extends DataFrameOperator {

private def prepareColumns(df: DataFrame, columns: Seq[(String, Option[String])]): (Map[String, LynxType], Map[String, Int], Seq[Int]) = {
val sourceSchema = df.schema.toMap
val columnNameIndex = df.columnsName.zipWithIndex.toMap
val usedIndices = columns.map(_._1).map(columnNameIndex)
(sourceSchema, columnNameIndex, usedIndices)
}
override def select(df: DataFrame, columns: Seq[(String, Option[String])]): DataFrame = {
val sourceSchema: Map[String, LynxType] = df.schema.toMap
val columnNameIndex: Map[String, Int] = df.columnsName.zipWithIndex.toMap
val newSchema: Seq[(String, LynxType)] = columns.map(column => column._2.getOrElse(column._1) -> sourceSchema(column._1))
val usedIndex: Seq[Int] = columns.map(_._1).map(columnNameIndex)
val (sourceSchema, _, usedIndices) = prepareColumns(df, columns)

@HC-teemo HC-teemo Dec 20, 2024

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems to be just extracting these few operations, and it's not that different.

val newSchema = columns.map(col => col._2.getOrElse(col._1) -> sourceSchema(col._1))

DataFrame(newSchema, () => df.records.map(row => usedIndex.map(row.apply)))
DataFrame(newSchema, () => df.records.map(row => usedIndices.map(row.apply)))
}

override def filter(df: DataFrame, predicate: Seq[LynxValue] => Boolean)(ctx: ExpressionContext): DataFrame =
Expand All @@ -42,28 +46,24 @@ class DefaultDataFrameOperator(expressionEvaluator: ExpressionEvaluator) extends


override def groupBy(df: DataFrame, groupings: Seq[(String, Expression)], aggregations: Seq[(String, Expression)])(ctx: ExpressionContext): DataFrame = {
// match (n:nothislabel) return count(n)
val newSchema = (groupings ++ aggregations).map(col =>
col._1 -> expressionEvaluator.typeOf(col._2, df.schema.toMap)
)
val newSchema = (groupings ++ aggregations).map(col => col._1 -> expressionEvaluator.typeOf(col._2, df.schema.toMap))
val columnsName = df.columnsName

DataFrame(newSchema, () => {
if (groupings.nonEmpty) {
df.records.map { record =>
val recordCtx = ctx.withVars(columnsName.zip(record).toMap)
groupings.map(col => expressionEvaluator.eval(col._2)(recordCtx)) -> recordCtx
} // (groupingValue: Seq[LynxValue] -> recordCtx: ExpressionContext)
.toSeq.groupBy(_._1) // #group by 'groupingValue'.
.mapValues(_.map(_._2)) // #trans to: (groupingValue: Seq[LynxValue] -> recordsCtx: Seq[ExpressionContext])
.map { case (groupingValue, recordsCtx) => // #aggragate: (groupingValues & aggregationValues): Seq[LynxValue]
groupingValue ++ {
aggregations.map { case (name, expr) => expressionEvaluator.aggregateEval(expr)(recordsCtx) }
}
}.toIterator
val groupingValues = groupings.map(col => expressionEvaluator.eval(col._2)(recordCtx))
groupingValues -> recordCtx
}.groupBy(_._1)
.view.flatMap { case (groupingValue, records) =>
Iterator(groupingValue ++ aggregations.map { case (_, expr) =>
expressionEvaluator.aggregateEval(expr)(records.map(_._2))
})
}.iterator
} else {
val allRecordsCtx = df.records.map { record => ctx.withVars(columnsName.zip(record).toMap) }.toSeq
Iterator(aggregations.map { case (name, expr) => expressionEvaluator.aggregateEval(expr)(allRecordsCtx) })
val allRecordsCtx = df.records.map(record => ctx.withVars(columnsName.zip(record).toMap)).toSeq
Iterator(aggregations.map { case (_, expr) => expressionEvaluator.aggregateEval(expr)(allRecordsCtx) })
}
})
}
Expand All @@ -89,25 +89,24 @@ class DefaultDataFrameOperator(expressionEvaluator: ExpressionEvaluator) extends

override def orderBy(df: DataFrame, sortItem: Seq[(Expression, Boolean)])(ctx: ExpressionContext): DataFrame = {
val columnsName = df.columnsName
DataFrame(df.schema, () => df.records.toSeq
.sortWith { (A, B) =>
val ctxA = ctx.withVars(columnsName.zip(A).toMap)
val ctxB = ctx.withVars(columnsName.zip(B).toMap)
val sortValue = sortItem.map {
case (exp, asc) =>
(expressionEvaluator.eval(exp)(ctxA), expressionEvaluator.eval(exp)(ctxB), asc)
}
_ascCmp(sortValue.toIterator)
}.toIterator)
val sortedRecords = df.records.toSeq.sortWith { (A, B) =>
val ctxA = ctx.withVars(columnsName.zip(A).toMap)
val ctxB = ctx.withVars(columnsName.zip(B).toMap)
val sortValues = sortItem.map { case (exp, asc) =>
val valueA = expressionEvaluator.eval(exp)(ctxA)
val valueB = expressionEvaluator.eval(exp)(ctxB)
(valueA.compareTo(valueB), asc)
}
sortValues.exists { case (cmp, asc) => cmp != 0 && (cmp > 0) == asc }

@HC-teemo HC-teemo Dec 20, 2024

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It doesn't seem to make much sense here. Need to return asc judgement on the first different value. For example the following example: [cmp,asc]={(0,t),(1,f),(1,t)}. should return false at (1,f), but the exists function will return true, because (1,t).

}
DataFrame(df.schema, () => sortedRecords.iterator)
}

private def _ascCmp(sortValue: Iterator[(LynxValue, LynxValue, Boolean)]): Boolean = {

@HC-teemo HC-teemo Dec 20, 2024

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The function _ascCmp does not seem to be used.

while (sortValue.hasNext) {
val (valueOfA, valueOfB, asc) = sortValue.next()
sortValue.find { case (valueOfA, valueOfB, asc) =>
val comparable = valueOfA.compareTo(valueOfB)
if(comparable != 0) return comparable > 0 != asc
}
false
comparable != 0 && comparable > 0 != asc
}.isDefined
}

}
107 changes: 41 additions & 66 deletions src/main/scala/org/grapheco/lynx/runner/NodeFilter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,74 +44,49 @@ case class NodeFilter(labels: Seq[LynxNodeLabel],
properties: Map[LynxPropertyKey, LynxValue],
propOps: Map[LynxPropertyKey, PropOp] = Map.empty) {

def matches(node: LynxNode, ignoreProps: LynxPropertyKey*): Boolean = {
val newProperties: Map[LynxPropertyKey, LynxValue] = properties -- ignoreProps
if (propOps.isEmpty) { // Compatible with empty propOps. e.g.PPTRelationshipScan
labels.forall(node.labels.contains) &&
newProperties.forall {
case (propertyName, value: LynxValue) => node.property(propertyName).exists(value.equals)
}
}
else {
labels.forall(node.labels.contains) &&
newProperties.forall {
case (propertyName, value) =>
if (!propOps.contains(propertyName)) {
false
} else {
val maybeOp: Option[PropOp] = propOps.get(propertyName)
maybeOp match {
case Some(EQUAL) => {
node.property(propertyName).exists(value.equals)
}
case Some(NOT_EQUAL) => {
if (node.property(propertyName).exists(value.equals)) false else true
}
case Some(IN) => {
val lynxValues: List[LynxValue] = value.asInstanceOf[LynxList].value
lynxValues.foreach(
f => if (node.property(propertyName).get.valueEq(f)) {
return true
})
false
}
case Some(LESS_THAN) =>
val nodeValue = node.property(propertyName)
if (nodeValue.isEmpty) false else nodeValue.get.<(value)
case Some(LESS_THAN_OR_EQUAL) => {
val nodeValue = node.property(propertyName)
if (nodeValue.isEmpty) false else nodeValue.get.<=(value)
}
case Some(GREATER_THAN) => {
val nodeValue = node.property(propertyName)
if (nodeValue.isEmpty) false else {
val bool = nodeValue.get.>(value)
bool
}
}
case Some(GREATER_THAN_OR_EQUAL) => {
val nodeValue = node.property(propertyName)
if (nodeValue.isEmpty) false else nodeValue.get.>=(value)
}
case Some(CONTAINS) => {
val lynxValue: LynxValue = node.property(propertyName).get
lynxValue.value match {
case lynxStr: String =>
value.value match {
case str: String =>
lynxStr.contains(str)
case _ =>
false
}
case _ =>
false
}
}
case _ => throw new scala.Exception("unexpected PropOp")
}
// Helper function to check property matching based on the operation
private def matchesProperty(node: LynxNode, propertyName: LynxPropertyKey, value: LynxValue, op: PropOp): Boolean = {
node.property(propertyName) match {
case Some(nodeValue) =>
op match {
case EQUAL => nodeValue.equals(value)
case NOT_EQUAL => !nodeValue.equals(value)
case LESS_THAN => nodeValue.<(value)
case LESS_THAN_OR_EQUAL => nodeValue.<=(value)
case GREATER_THAN => nodeValue.>(value)
case GREATER_THAN_OR_EQUAL => nodeValue.>=(value)
case CONTAINS => nodeValue.value match {
case lynxStr: String => value.value match {
case str: String => lynxStr.contains(str)
case _ => false
}
case _ => false
case _ => false
}
case IN => value match {
case lynxList: LynxList => lynxList.value.exists(v => nodeValue.valueEq(v))
case _ => false
}
case _ => throw new scala.Exception(s"Unsupported PropOp: $op")
}
case None => false
}
}

// Core matching logic
def matches(node: LynxNode, ignoreProps: LynxPropertyKey*): Boolean = {
val newProperties = properties -- ignoreProps
val hasMatchingLabels = labels.forall(node.labels.contains)

// If no specific property operations, just check labels and simple equality for properties
if (propOps.isEmpty) {
hasMatchingLabels && newProperties.forall {
case (propertyName, value) => node.property(propertyName).contains(value)
}
} else {
hasMatchingLabels && newProperties.forall {
case (propertyName, value) =>
propOps.get(propertyName).exists(matchesProperty(node, propertyName, value, _))
}
}
}
}