Skip to content

[SPARK-4131] [Follow-up] Support "Writing data into the filesystem from queries" #19225

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
gatorsmile wants to merge 2 commits into from
Closed

[SPARK-4131] [Follow-up] Support "Writing data into the filesystem from queries" #19225

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b19ee89
Merge remote-tracking branch 'upstream/master' into refactorSPARK-4131
gatorsmile Sep 13, 2017
299cebb
fix
gatorsmile Sep 13, 2017
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
fix
  • Loading branch information
@gatorsmile
gatorsmile committed Sep 13, 2017
commit 299cebb2be933c689f33c0cd96e1a0f7e0f91623
6 changes: 5 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -801,7 +801,11 @@ object DDLUtils {
val HIVE_PROVIDER = "hive"

def isHiveTable(table: CatalogTable): Boolean = {
table.provider.isDefined && table.provider.get.toLowerCase(Locale.ROOT) == HIVE_PROVIDER
isHiveTable(table.provider)
}

def isHiveTable(provider: Option[String]): Boolean = {
provider.isDefined && provider.get.toLowerCase(Locale.ROOT) == HIVE_PROVIDER
}

def isDatasourceTable(table: CatalogTable): Boolean = {
Expand Down
3 changes: 1 addition & 2 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveStrategies.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,8 +160,7 @@ object HiveAnalysis extends Rule[LogicalPlan] {
CreateHiveTableAsSelectCommand(tableDesc, query, mode)

case InsertIntoDir(isLocal, storage, provider, child, overwrite)
if provider.isDefined && provider.get.toLowerCase(Locale.ROOT) == DDLUtils.HIVE_PROVIDER =>

if DDLUtils.isHiveTable(provider) =>
val outputPath = new Path(storage.locationUri.get)
if (overwrite) DDLUtils.verifyNotReadPath(child, outputPath)

Expand Down
6 changes: 4 additions & 2 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/SaveAsHiveFile.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import org.apache.hadoop.conf.Configuration

import org.apache.spark.internal.io.FileCommitProtocol
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.command.DataWritingCommand
Expand All @@ -36,7 +37,8 @@ private[hive] trait SaveAsHiveFile extends DataWritingCommand {
hadoopConf: Configuration,
fileSinkConf: FileSinkDesc,
outputLocation: String,
partitionAttributes: Seq[Attribute] = Nil): Unit = {
customPartitionLocations: Map[TablePartitionSpec, String] = Map.empty,
partitionAttributes: Seq[Attribute] = Nil): Set[String] = {

val isCompressed = hadoopConf.get("hive.exec.compress.output", "false").toBoolean
if (isCompressed) {
Expand All @@ -62,7 +64,7 @@ private[hive] trait SaveAsHiveFile extends DataWritingCommand {
plan = plan,
fileFormat = new HiveFileFormat(fileSinkConf),
committer = committer,
outputSpec = FileFormatWriter.OutputSpec(outputLocation, Map.empty),
outputSpec = FileFormatWriter.OutputSpec(outputLocation, customPartitionLocations),
hadoopConf = hadoopConf,
partitionColumns = partitionAttributes,
bucketSpec = None,
Expand Down