Skip to content

[SPARK-17480][SQL][FOLLOWUP] Fix more instances which calls List.length/size which is O(n) #15093

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

Closed
HyukjinKwon wants to merge 6 commits into from
Closed

[SPARK-17480][SQL][FOLLOWUP] Fix more instances which calls List.length/size which is O(n) #15093

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
fbc3419
Fix more same instances
HyukjinKwon Sep 14, 2016
ecd3dde
Last instance
HyukjinKwon Sep 14, 2016
996e292
Revert unrelated changes and use funtional transformation for a case
HyukjinKwon Sep 14, 2016
cb536a4
Address comment
HyukjinKwon Sep 15, 2016
f5e8131
Fix comments
HyukjinKwon Sep 17, 2016
8a3d293
Correct the comments more
HyukjinKwon Sep 17, 2016
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
7 changes: 1 addition & 6 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1658,19 +1658,14 @@ class Analyzer(
// Third, for every Window Spec, we add a Window operator and set currentChild as the
// child of it.
var currentChild = child
var i = 0
while (i < groupedWindowExpressions.size) {
val ((partitionSpec, orderSpec), windowExpressions) = groupedWindowExpressions(i)
groupedWindowExpressions.foreach { case ((partitionSpec, orderSpec), windowExpressions) =>
Copy link
Member Author

@HyukjinKwon HyukjinKwon Sep 15, 2016 •
edited
Loading

Choose a reason for hiding this comment

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

Hi @srowen , It seems groupedWindowExpressions is Seq[((Seq[Expression], Seq[SortOrder]), ArrayBuffer[NamedExpression])] but the desired output is LogicalPlan. So, it seems I can't directly use fold or reduce simply. So, could this just use foreach rather than others? Otherwise, please teach me for a better expression.

Copy link
Member

@srowen srowen Sep 15, 2016

Choose a reason for hiding this comment

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

Well currentChild is a Window, and that's the only thing whose computation is changed here. How about:

      val currentChild =
        groupedWindowExpressions.foldLeft(child) {
          case (last, ((partitionSpec, orderSpec), windowExpressions)) =>
            Window(windowExpressions, partitionSpec, orderSpec, last)
        }

Copy link
Member Author

@HyukjinKwon HyukjinKwon Sep 15, 2016

Choose a reason for hiding this comment

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

Ah, that's clear and better. I will try. Thanks!

// Set currentChild to the newly created Window operator.
currentChild =
Window(
windowExpressions,
partitionSpec,
orderSpec,
currentChild)

// Move to next Window Spec.
i += 1
}

// Finally, we create a Project to output currentChild's output
Expand Down
3 changes: 2 additions & 1 deletion ...yst/src/main/scala/org/apache/spark/sql/catalyst/expressions/conditionalExpressions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,8 @@ abstract class CaseWhenBase(

override def eval(input: InternalRow): Any = {
var i = 0
while (i < branches.size) {
val size = branches.size
while (i < size) {
if (java.lang.Boolean.TRUE.equals(branches(i)._1.eval(input))) {
return branches(i)._2.eval(input)
}
Expand Down
3 changes: 2 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ordering.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,8 @@ class InterpretedOrdering(ordering: Seq[SortOrder]) extends Ordering[InternalRow

def compare(a: InternalRow, b: InternalRow): Int = {
var i = 0
while (i < ordering.size) {
val size = ordering.size
while (i < size) {
val order = ordering(i)
val left = order.child.eval(a)
val right = order.child.eval(b)
Expand Down
10 changes: 5 additions & 5 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/QuantileSummaries.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,18 +91,18 @@ class QuantileSummaries(
var sampleIdx = 0
// The index of the sample currently being inserted.
var opsIdx: Int = 0
while(opsIdx < sorted.length) {
while (opsIdx < sorted.length) {
val currentSample = sorted(opsIdx)
// Add all the samples before the next observation.
while(sampleIdx < sampled.size && sampled(sampleIdx).value <= currentSample) {
while (sampleIdx < sampled.length && sampled(sampleIdx).value <= currentSample) {
newSamples += sampled(sampleIdx)
sampleIdx += 1
}

// If it is the first one to insert, of if it is the last one
currentCount += 1
val delta =
if (newSamples.isEmpty || (sampleIdx == sampled.size && opsIdx == sorted.length - 1)) {
if (newSamples.isEmpty || (sampleIdx == sampled.length && opsIdx == sorted.length - 1)) {
0
} else {
math.floor(2 * relativeError * currentCount).toInt
Expand All @@ -114,7 +114,7 @@ class QuantileSummaries(
}

// Add all the remaining existing samples
while(sampleIdx < sampled.size) {
while (sampleIdx < sampled.length) {
newSamples += sampled(sampleIdx)
sampleIdx += 1
}
Expand Down Expand Up @@ -195,7 +195,7 @@ class QuantileSummaries(
// Minimum rank at current sample
var minRank = 0
var i = 1
while (i < sampled.size - 1) {
while (i < sampled.length - 1) {
val curSample = sampled(i)
minRank += curSample.g
val maxRank = minRank + curSample.delta
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,7 @@ object JdbcUtils extends Logging {
val bytes = rs.getBytes(pos + 1)
var ans = 0L
var j = 0
while (j < bytes.size) {
while (j < bytes.length) {
ans = 256 * ans + (255 & bytes(j))
j = j + 1
}
Expand Down
6 changes: 4 additions & 2 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -703,7 +703,8 @@ private[hive] trait HiveInspectors {
// 1. create the pojo (most likely) object
val result = x.create()
var i = 0
while (i < fieldRefs.size) {
val size = fieldRefs.size
while (i < size) {
// 2. set the property for the pojo
val tpe = structType(i).dataType
x.setStructFieldData(
Expand All @@ -720,7 +721,8 @@ private[hive] trait HiveInspectors {
val row = a.asInstanceOf[InternalRow]
val result = new java.util.ArrayList[AnyRef](fieldRefs.size)
var i = 0
while (i < fieldRefs.size) {
val size = fieldRefs.size
while (i < size) {
val tpe = structType(i).dataType
result.add(wrap(row.get(i, tpe), fieldRefs.get(i).getFieldObjectInspector, tpe))
i += 1
Expand Down
3 changes: 2 additions & 1 deletion sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,8 @@ private[hive] object HadoopTableReader extends HiveInspectors with Logging {
iterator.map { value =>
val raw = converter.convert(rawDeser.deserialize(value))
var i = 0
while (i < fieldRefs.length) {
val length = fieldRefs.length
while (i < length) {
val fieldValue = soi.getStructFieldData(raw, fieldRefs(i))
if (fieldValue == null) {
mutableRow.setNullAt(fieldOrdinals(i))
Expand Down
3 changes: 2 additions & 1 deletion sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,8 @@ private[hive] case class HiveGenericUDF(
returnInspector // Make sure initialized.

var i = 0
while (i < children.length) {
val length = children.length
while (i < length) {
val idx = i
deferredObjects(i).asInstanceOf[DeferredObjectAdapter]
.set(() => children(idx).eval(input))
Expand Down
6 changes: 4 additions & 2 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,8 @@ private[orc] class OrcSerializer(dataSchema: StructType, conf: Configuration)
row: InternalRow): Unit = {
val fieldRefs = oi.getAllStructFieldRefs
var i = 0
while (i < fieldRefs.size) {
val size = fieldRefs.size
while (i < size) {

oi.setStructFieldData(
struct,
Expand Down Expand Up @@ -289,7 +290,8 @@ private[orc] object OrcRelation extends HiveInspectors {
iterator.map { value =>
val raw = deserializer.deserialize(value)
var i = 0
while (i < fieldRefs.length) {
val length = fieldRefs.length
while (i < length) {
val fieldValue = oi.getStructFieldData(raw, fieldRefs(i))
if (fieldValue == null) {
mutableRow.setNullAt(fieldOrdinals(i))
Expand Down