Skip to content

[SPARK-6844][SQL] Clean up accumulators used in InMemoryRelation when it is uncached #5475

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
viirya wants to merge 5 commits into from
Closed

[SPARK-6844][SQL] Clean up accumulators used in InMemoryRelation when it is uncached #5475

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1c3b06e
Clean up accumulators used in InMemoryRelation when it is uncached.
viirya Apr 11, 2015
26c9bb6
Add configuration to enable in-memory table scan accumulators.
viirya Apr 12, 2015
78af229
Merge remote-tracking branch 'upstream/master' into cache_memory_leak
viirya Apr 14, 2015
dc1d5d5
For comments.
viirya Apr 15, 2015
0b41235
fix style.
viirya Apr 15, 2015
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
Prev Previous commit
Next Next commit
Add configuration to enable in-memory table scan accumulators.
  • Loading branch information
@viirya
viirya committed Apr 12, 2015
commit 26c9bb61b70391f2408bbb0846a8e486baf3ec4d
22 changes: 17 additions & 5 deletions sql/core/src/main/scala/org/apache/spark/sql/columnar/InMemoryColumnarTableScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -282,14 +282,24 @@ private[sql] case class InMemoryColumnarTableScan(
}
}

lazy val enableAccumulators: Boolean =
sqlContext.getConf("spark.sql.inMemoryTableScanStatistics.enable", "false").toBoolean

// Accumulators used for testing purposes
val (readPartitions, readBatches) = relation.applyScanAccumulators(sparkContext)
lazy val (readPartitions, readBatches) =
if (enableAccumulators) {
relation.applyScanAccumulators(sparkContext)
} else {
(null, null)
}

private val inMemoryPartitionPruningEnabled = sqlContext.conf.inMemoryPartitionPruning

override def execute(): RDD[Row] = {
readPartitions.setValue(0)
readBatches.setValue(0)
if (enableAccumulators) {
readPartitions.setValue(0)
readBatches.setValue(0)
}

relation.cachedColumnBuffers.mapPartitions { cachedBatchIterator =>
val partitionFilter = newPredicate(
Expand Down Expand Up @@ -339,7 +349,7 @@ private[sql] case class InMemoryColumnarTableScan(
}
}

if (rows.hasNext) {
if (rows.hasNext && enableAccumulators) {
readPartitions += 1
}

Expand All @@ -358,7 +368,9 @@ private[sql] case class InMemoryColumnarTableScan(
logInfo(s"Skipping partition based on stats $statsString")
false
} else {
readBatches += 1
if (enableAccumulators) {
readBatches += 1
}
true
}
}
Expand Down
2 changes: 2 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/columnar/PartitionBatchPruningSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ class PartitionBatchPruningSuite extends FunSuite with BeforeAndAfterAll with Be

// Enable in-memory partition pruning
setConf(SQLConf.IN_MEMORY_PARTITION_PRUNING, "true")
// Enable in-memory table scan accumulators
setConf("spark.sql.inMemoryTableScanStatistics.enable", "true")
Copy link
Member

@gatorsmile gatorsmile Jan 12, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

Why not setting spark.sql.inMemoryTableScanStatistics.enable back false in afterAll?

}

override protected def afterAll(): Unit = {
Expand Down