Skip to content
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

[SPARK-14596][SQL] Remove not used SqlNewHadoopRDD and some more unused imports #12354

Closed
wants to merge 5 commits into from
Closed

[SPARK-14596][SQL] Remove not used SqlNewHadoopRDD and some more unused imports #12354

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8b8e961
Remove SqlNewHadoopRDD and unused imports
HyukjinKwon Apr 13, 2016
a291332
Rename SqlNewHadoopRDDState to FileScanRDDState
HyukjinKwon Apr 13, 2016
3f8e878
Update indentation
HyukjinKwon Apr 13, 2016
9c5893d
Remove the change of file name and add some more changes for removing…
HyukjinKwon Apr 14, 2016
6cb3547
Rename SqlNewHadoopRDDState to InputFileNameHolder
HyukjinKwon Apr 14, 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
6 changes: 3 additions & 3 deletions ...ache/spark/rdd/SqlNewHadoopRDDState.scala → ...g/apache/spark/rdd/FileScanRDDState.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,10 @@ package org.apache.spark.rdd
import org.apache.spark.unsafe.types.UTF8String

/**
* State for SqlNewHadoopRDD objects. This is split this way because of the package splits.
* TODO: Move/Combine this with org.apache.spark.sql.datasources.SqlNewHadoopRDD
* State for FileScanRDD objects. This is split this way because of the package splits.
* TODO: Move/Combine this with org.apache.spark.sql.datasources.FileScanRDD
*/
private[spark] object SqlNewHadoopRDDState {
private[spark] object FileScanRDDState {
/**
* The thread variable for the name of the current file being read. This is used by
* the InputFileName function in Spark SQL.
Expand Down
6 changes: 3 additions & 3 deletions core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,8 +220,8 @@ class HadoopRDD[K, V](

// Sets the thread local variable for the file's name
split.inputSplit.value match {
case fs: FileSplit => SqlNewHadoopRDDState.setInputFileName(fs.getPath.toString)
case _ => SqlNewHadoopRDDState.unsetInputFileName()
case fs: FileSplit => FileScanRDDState.setInputFileName(fs.getPath.toString)
Copy link
Contributor

Choose a reason for hiding this comment

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

@liancheng @yhuai Do you still use HadoopRDD to read data source relation? If not, I think we don't need to update file name here anymore.

Copy link
Contributor

Choose a reason for hiding this comment

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

I think it is used by HiveTableScan

Copy link
Member Author

Choose a reason for hiding this comment

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

@cloud-fan I see. This is used in https://github.com/apache/spark/blob/d6dc12ef0146ae409834c78737c116050961f350/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InputFileName.scala during code generation.

Should I remove the change of file name?

Copy link
Contributor

Choose a reason for hiding this comment

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

i see, how about renaming it to InputFileNameHolder?

Copy link
Member Author

Choose a reason for hiding this comment

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

Sure. Thanks.

case _ => FileScanRDDState.unsetInputFileName()
}

// Find a function that will return the FileSystem bytes read by this thread. Do this before
Expand Down Expand Up @@ -271,7 +271,7 @@ class HadoopRDD[K, V](

override def close() {
if (reader != null) {
SqlNewHadoopRDDState.unsetInputFileName()
FileScanRDDState.unsetInputFileName()
// Close the reader and release it. Note: it's very important that we don't close the
// reader more than once, since that exposes us to MAPREDUCE-5918 when running against
// Hadoop 1.x and older Hadoop 2.x releases. That bug can lead to non-deterministic
Expand Down
4 changes: 2 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InputFileName.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.sql.catalyst.expressions

import org.apache.spark.rdd.SqlNewHadoopRDDState
import org.apache.spark.rdd.FileScanRDDState
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
import org.apache.spark.sql.types.{DataType, StringType}
Expand All @@ -40,7 +40,7 @@ case class InputFileName() extends LeafExpression with Nondeterministic {
override protected def initInternal(): Unit = {}

override protected def evalInternal(input: InternalRow): UTF8String = {
SqlNewHadoopRDDState.getInputFileName()
FileScanRDDState.getInputFileName()
}

override def genCode(ctx: CodegenContext, ev: ExprCode): String = {
Expand Down
10 changes: 5 additions & 5 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.apache.spark.sql.execution.datasources

import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.rdd.{RDD, SqlNewHadoopRDDState}
import org.apache.spark.rdd.{FileScanRDDState, RDD}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.catalyst.InternalRow

Expand Down Expand Up @@ -50,7 +50,7 @@ class FileScanRDD(
@transient val sqlContext: SQLContext,
readFunction: (PartitionedFile) => Iterator[InternalRow],
@transient val filePartitions: Seq[FilePartition])
extends RDD[InternalRow](sqlContext.sparkContext, Nil) {
extends RDD[InternalRow](sqlContext.sparkContext, Nil) {

override def compute(split: Partition, context: TaskContext): Iterator[InternalRow] = {
val iterator = new Iterator[Object] with AutoCloseable {
Expand All @@ -65,17 +65,17 @@ class FileScanRDD(
if (files.hasNext) {
val nextFile = files.next()
logInfo(s"Reading File $nextFile")
SqlNewHadoopRDDState.setInputFileName(nextFile.filePath)
FileScanRDDState.setInputFileName(nextFile.filePath)
currentIterator = readFunction(nextFile)
hasNext
} else {
SqlNewHadoopRDDState.unsetInputFileName()
FileScanRDDState.unsetInputFileName()
false
}
}

override def close() = {
SqlNewHadoopRDDState.unsetInputFileName()
FileScanRDDState.unsetInputFileName()
}
}

Expand Down
282 changes: 0 additions & 282 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SqlNewHadoopRDD.scala
View file
Open in desktop

This file was deleted.

1 change: 0 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.config._
import org.apache.spark.network.util.ByteUnit
import org.apache.spark.sql.catalyst.CatalystConf
import org.apache.spark.util.Utils

////////////////////////////////////////////////////////////////////////////////////////////////////
// This file defines the configuration options for Spark SQL.
Expand Down
Loading