Skip to content

Rewrite VEProjection to have separate loops for data and validity. #563

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
wants to merge 5 commits into
base: main
Choose a base branch
from
Open

Rewrite VEProjection to have separate loops for data and validity. #563

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
38d1e85
Rewrite VEProjection to have separate loops for data and validity.
wmeddie Apr 21, 2022
9560bb6
Fix C++ code.
wmeddie Apr 21, 2022
7ce7253
_NEC vector all the things.
wmeddie Apr 21, 2022
58ea8f3
Remove dbgen binary.
wmeddie Apr 21, 2022
30435f9
Make _NEC vector pragma explicit.
wmeddie Apr 22, 2022
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
6 changes: 4 additions & 2 deletions src/main/scala/com/nec/spark/agile/core/CodeLines.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
package com.nec.spark.agile.core

import scala.annotation.tailrec
import sourcecode.{FullName, Line}

import scala.annotation.tailrec

object CodeLines {
def parse(source: String): CodeLines = {
CodeLines.from(source.split("\\r?\\n").toList)
Expand All @@ -29,8 +30,9 @@ object CodeLines {
CodeLines.from(s"if ($condition) { ", sub1.indented, "} else {", sub2.indented, "}")
}

def forLoop(counter: String, until: String)(sub: => CodeLines): CodeLines = {
def forLoop(counter: String, until: String, vector: Boolean = false)(sub: => CodeLines): CodeLines = {
CodeLines.from(
if (vector) "#pragma _NEC vector" else "",
s"for (auto $counter = 0; $counter < $until; $counter++) {",
sub.indented,
"}"
Expand Down
23 changes: 18 additions & 5 deletions src/main/scala/com/nec/spark/agile/projection/ProjectionFunction.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
package com.nec.spark.agile.projection

import com.nec.spark.agile.core._
import com.nec.spark.agile.CFunctionGeneration._
import com.nec.spark.agile.StringProducer.FrovedisStringProducer
import com.nec.spark.agile.core._
import com.nec.spark.agile.core.CFunction2.CFunctionArgument
import com.nec.spark.agile.core._

final case class ProjectionFunction(name: String,
data: Seq[CVector],
Expand Down Expand Up @@ -59,10 +58,24 @@ final case class ProjectionFunction(name: String,
"#pragma _NEC vector",
CodeLines.forLoop("i", s"${inputs.head.name}->count") {
List(
s"bool validity = ${cexpr.isNotNullCode.getOrElse("1")};",
s"$outname->data[i] = ${cexpr.cCode};",
s"$outname->set_validity(i, validity);"
s"${outname}->data[i] = ${cexpr.cCode};",
)
},
if (cexpr.isNotNullCode.isEmpty) {
CodeLines.from(
s"size_t vcount = ceil(${inputs.head.name}->count / 64.0);",
CodeLines.forLoop("i", s"vcount", vector = true) {
List(
s"${outname}->validityBuffer[i] = 0xffffffff;"
)
}
)
} else {
CodeLines.forLoop("i", s"${inputs.head.name}->count", vector = true) {
List(
s"$outname->set_validity(i, ${cexpr.isNotNullCode.getOrElse("1")});"
)
}
}
)
}
Expand Down
Binary file removed tests/tpchbench/dbgen/dbgen
View file
Open in desktop
Binary file not shown.
Binary file removed tests/tpchbench/dbgen/qgen
View file
Open in desktop
Binary file not shown.
6 changes: 5 additions & 1 deletion tests/tpchbench/run_ve.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ time $SPARK_HOME/bin/spark-submit \
--conf spark.executorEnv.VE_OMP_NUM_THREADS=1 \
--conf spark.com.nec.spark.aggregate-on-ve=true \
--conf spark.com.nec.spark.sort-on-ve=true \
--conf spark.com.nec.spark.project-on-ve=false \
--conf spark.com.nec.spark.project-on-ve=true \
--conf spark.com.nec.spark.filter-on-ve=true \
--conf spark.com.nec.spark.exchange-on-ve=true \
--conf spark.com.nec.spark.join-on-ve=false \
Expand All @@ -31,3 +31,7 @@ time $SPARK_HOME/bin/spark-submit \
--conf spark.shuffle.partitions=8 \
target/scala-2.12/tpchbench-assembly-0.0.1.jar \
$*

# --conf spark.executorEnv.VE_PROGINF=YES \
# --conf spark.executorEnv.VE_ACC_IO=1 \
# --conf spark.executorEnv.VEO_LOG_DEBUG=1 \