diff options
author | Michael Armbrust <michael@databricks.com> | 2016-03-07 15:15:10 -0800 |
---|---|---|
committer | Reynold Xin <rxin@databricks.com> | 2016-03-07 15:15:10 -0800 |
commit | e720dda42e806229ccfd970055c7b8a93eb447bf (patch) | |
tree | 641c3d454b638a347adc4c51db8cc69c41b44ac2 /sql/hive | |
parent | 0eea12a3d956b54bbbd73d21b296868852a04494 (diff) | |
download | spark-e720dda42e806229ccfd970055c7b8a93eb447bf.tar.gz spark-e720dda42e806229ccfd970055c7b8a93eb447bf.tar.bz2 spark-e720dda42e806229ccfd970055c7b8a93eb447bf.zip |
[SPARK-13665][SQL] Separate the concerns of HadoopFsRelation
`HadoopFsRelation` is used for reading most files into Spark SQL. However today this class mixes the concerns of file management, schema reconciliation, scan building, bucketing, partitioning, and writing data. As a result, many data sources are forced to reimplement the same functionality and the various layers have accumulated a fair bit of inefficiency. This PR is a first cut at separating this into several components / interfaces that are each described below. Additionally, all implementations inside of Spark (parquet, csv, json, text, orc, svmlib) have been ported to the new API `FileFormat`. External libraries, such as spark-avro will also need to be ported to work with Spark 2.0.
### HadoopFsRelation
A simple `case class` that acts as a container for all of the metadata required to read from a datasource. All discovery, resolution and merging logic for schemas and partitions has been removed. This an internal representation that no longer needs to be exposed to developers.
```scala
case class HadoopFsRelation(
sqlContext: SQLContext,
location: FileCatalog,
partitionSchema: StructType,
dataSchema: StructType,
bucketSpec: Option[BucketSpec],
fileFormat: FileFormat,
options: Map[String, String]) extends BaseRelation
```
### FileFormat
The primary interface that will be implemented by each different format including external libraries. Implementors are responsible for reading a given format and converting it into `InternalRow` as well as writing out an `InternalRow`. A format can optionally return a schema that is inferred from a set of files.
```scala
trait FileFormat {
def inferSchema(
sqlContext: SQLContext,
options: Map[String, String],
files: Seq[FileStatus]): Option[StructType]
def prepareWrite(
sqlContext: SQLContext,
job: Job,
options: Map[String, String],
dataSchema: StructType): OutputWriterFactory
def buildInternalScan(
sqlContext: SQLContext,
dataSchema: StructType,
requiredColumns: Array[String],
filters: Array[Filter],
bucketSet: Option[BitSet],
inputFiles: Array[FileStatus],
broadcastedConf: Broadcast[SerializableConfiguration],
options: Map[String, String]): RDD[InternalRow]
}
```
The current interface is based on what was required to get all the tests passing again, but still mixes a couple of concerns (i.e. `bucketSet` is passed down to the scan instead of being resolved by the planner). Additionally, scans are still returning `RDD`s instead of iterators for single files. In a future PR, bucketing should be removed from this interface and the scan should be isolated to a single file.
### FileCatalog
This interface is used to list the files that make up a given relation, as well as handle directory based partitioning.
```scala
trait FileCatalog {
def paths: Seq[Path]
def partitionSpec(schema: Option[StructType]): PartitionSpec
def allFiles(): Seq[FileStatus]
def getStatus(path: Path): Array[FileStatus]
def refresh(): Unit
}
```
Currently there are two implementations:
- `HDFSFileCatalog` - based on code from the old `HadoopFsRelation`. Infers partitioning by recursive listing and caches this data for performance
- `HiveFileCatalog` - based on the above, but it uses the partition spec from the Hive Metastore.
### ResolvedDataSource
Produces a logical plan given the following description of a Data Source (which can come from DataFrameReader or a metastore):
- `paths: Seq[String] = Nil`
- `userSpecifiedSchema: Option[StructType] = None`
- `partitionColumns: Array[String] = Array.empty`
- `bucketSpec: Option[BucketSpec] = None`
- `provider: String`
- `options: Map[String, String]`
This class is responsible for deciding which of the Data Source APIs a given provider is using (including the non-file based ones). All reconciliation of partitions, buckets, schema from metastores or inference is done here.
### DataSourceAnalysis / DataSourceStrategy
Responsible for analyzing and planning reading/writing of data using any of the Data Source APIs, including:
- pruning the files from partitions that will be read based on filters.
- appending partition columns*
- applying additional filters when a data source can not evaluate them internally.
- constructing an RDD that is bucketed correctly when required*
- sanity checking schema match-up and other analysis when writing.
*In the future we should do that following:
- Break out file handling into its own Strategy as its sufficiently complex / isolated.
- Push the appending of partition columns down in to `FileFormat` to avoid an extra copy / unvectorization.
- Use a custom RDD for scans instead of `SQLNewNewHadoopRDD2`
Author: Michael Armbrust <michael@databricks.com>
Author: Wenchen Fan <wenchen@databricks.com>
Closes #11509 from marmbrus/fileDataSource.
Diffstat (limited to 'sql/hive')
17 files changed, 273 insertions, 1033 deletions
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala index a053108b7d..28874189de 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala @@ -22,7 +22,7 @@ import scala.collection.mutable import com.google.common.base.Objects import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache} -import org.apache.hadoop.fs.Path +import org.apache.hadoop.fs.{FileStatus, Path} import org.apache.hadoop.hive.common.StatsSetupConst import org.apache.hadoop.hive.conf.HiveConf import org.apache.hadoop.hive.metastore.{TableType => HiveTableType, Warehouse} @@ -42,7 +42,7 @@ import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules._ import org.apache.spark.sql.execution.{datasources, FileRelation} import org.apache.spark.sql.execution.datasources.{Partition => ParquetPartition, _} -import org.apache.spark.sql.execution.datasources.parquet.ParquetRelation +import org.apache.spark.sql.execution.datasources.parquet.{DefaultSource, ParquetRelation} import org.apache.spark.sql.hive.client._ import org.apache.spark.sql.hive.execution.HiveNativeCommand import org.apache.spark.sql.sources._ @@ -175,18 +175,15 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte BucketSpec(n.toInt, getColumnNames("bucket"), getColumnNames("sort")) } - // It does not appear that the ql client for the metastore has a way to enumerate all the - // SerDe properties directly... val options = table.storage.serdeProperties - val resolvedRelation = ResolvedDataSource( hive, - userSpecifiedSchema, - partitionColumns.toArray, - bucketSpec, - table.properties("spark.sql.sources.provider"), - options) + userSpecifiedSchema = userSpecifiedSchema, + partitionColumns = partitionColumns.toArray, + bucketSpec = bucketSpec, + provider = table.properties("spark.sql.sources.provider"), + options = options) LogicalRelation( resolvedRelation.relation, @@ -285,8 +282,14 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte } val maybeSerDe = HiveSerDe.sourceToSerDe(provider, hive.hiveconf) - val dataSource = ResolvedDataSource( - hive, userSpecifiedSchema, partitionColumns, bucketSpec, provider, options) + val dataSource = + ResolvedDataSource( + hive, + userSpecifiedSchema = userSpecifiedSchema, + partitionColumns = partitionColumns, + bucketSpec = bucketSpec, + provider = provider, + options = options) def newSparkSQLSpecificMetastoreTable(): CatalogTable = { CatalogTable( @@ -308,14 +311,14 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte relation: HadoopFsRelation, serde: HiveSerDe): CatalogTable = { assert(partitionColumns.isEmpty) - assert(relation.partitionColumns.isEmpty) + assert(relation.partitionSchema.isEmpty) CatalogTable( specifiedDatabase = Option(dbName), name = tblName, tableType = tableType, storage = CatalogStorageFormat( - locationUri = Some(relation.paths.head), + locationUri = Some(relation.location.paths.map(_.toUri.toString).head), inputFormat = serde.inputFormat, outputFormat = serde.outputFormat, serde = serde.serde, @@ -339,25 +342,26 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte (None, message) case (Some(serde), relation: HadoopFsRelation) - if relation.paths.length == 1 && relation.partitionColumns.isEmpty => + if relation.location.paths.length == 1 && relation.partitionSchema.isEmpty => val hiveTable = newHiveCompatibleMetastoreTable(relation, serde) val message = s"Persisting data source relation $qualifiedTableName with a single input path " + - s"into Hive metastore in Hive compatible format. Input path: ${relation.paths.head}." + s"into Hive metastore in Hive compatible format. Input path: " + + s"${relation.location.paths.head}." (Some(hiveTable), message) - case (Some(serde), relation: HadoopFsRelation) if relation.partitionColumns.nonEmpty => + case (Some(serde), relation: HadoopFsRelation) if relation.partitionSchema.nonEmpty => val message = s"Persisting partitioned data source relation $qualifiedTableName into " + "Hive metastore in Spark SQL specific format, which is NOT compatible with Hive. " + - "Input path(s): " + relation.paths.mkString("\n", "\n", "") + "Input path(s): " + relation.location.paths.mkString("\n", "\n", "") (None, message) case (Some(serde), relation: HadoopFsRelation) => val message = s"Persisting data source relation $qualifiedTableName with multiple input paths into " + "Hive metastore in Spark SQL specific format, which is NOT compatible with Hive. " + - s"Input paths: " + relation.paths.mkString("\n", "\n", "") + s"Input paths: " + relation.location.paths.mkString("\n", "\n", "") (None, message) case (Some(serde), _) => @@ -441,11 +445,7 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte val metastoreSchema = StructType.fromAttributes(metastoreRelation.output) val mergeSchema = hive.convertMetastoreParquetWithSchemaMerging - // NOTE: Instead of passing Metastore schema directly to `ParquetRelation`, we have to - // serialize the Metastore schema to JSON and pass it as a data source option because of the - // evil case insensitivity issue, which is reconciled within `ParquetRelation`. val parquetOptions = Map( - ParquetRelation.METASTORE_SCHEMA -> metastoreSchema.json, ParquetRelation.MERGE_SCHEMA -> mergeSchema.toString, ParquetRelation.METASTORE_TABLE_NAME -> TableIdentifier( metastoreRelation.tableName, @@ -462,11 +462,11 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte partitionSpecInMetastore: Option[PartitionSpec]): Option[LogicalRelation] = { cachedDataSourceTables.getIfPresent(tableIdentifier) match { case null => None // Cache miss - case logical @ LogicalRelation(parquetRelation: ParquetRelation, _, _) => + case logical @ LogicalRelation(parquetRelation: HadoopFsRelation, _, _) => // If we have the same paths, same schema, and same partition spec, // we will use the cached Parquet Relation. val useCached = - parquetRelation.paths.toSet == pathsInMetastore.toSet && + parquetRelation.location.paths.map(_.toString).toSet == pathsInMetastore.toSet && logical.schema.sameType(metastoreSchema) && parquetRelation.partitionSpec == partitionSpecInMetastore.getOrElse { PartitionSpec(StructType(Nil), Array.empty[datasources.Partition]) @@ -502,13 +502,33 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte ParquetPartition(values, location) } val partitionSpec = PartitionSpec(partitionSchema, partitions) - val paths = partitions.map(_.path) - val cached = getCached(tableIdentifier, paths, metastoreSchema, Some(partitionSpec)) + val cached = getCached( + tableIdentifier, + metastoreRelation.table.storage.locationUri.toSeq, + metastoreSchema, + Some(partitionSpec)) + val parquetRelation = cached.getOrElse { - val created = LogicalRelation( - new ParquetRelation( - paths.toArray, None, Some(partitionSpec), parquetOptions)(hive)) + val paths = new Path(metastoreRelation.table.storage.locationUri.get) :: Nil + val fileCatalog = new HiveFileCatalog(hive, paths, partitionSpec) + val format = new DefaultSource() + val inferredSchema = format.inferSchema(hive, parquetOptions, fileCatalog.allFiles()) + + val mergedSchema = inferredSchema.map { inferred => + ParquetRelation.mergeMetastoreParquetSchema(metastoreSchema, inferred) + }.getOrElse(metastoreSchema) + + val relation = HadoopFsRelation( + sqlContext = hive, + location = fileCatalog, + partitionSchema = partitionSchema, + dataSchema = mergedSchema, + bucketSpec = None, // We don't support hive bucketed tables, only ones we write out. + fileFormat = new DefaultSource(), + options = parquetOptions) + + val created = LogicalRelation(relation) cachedDataSourceTables.put(tableIdentifier, created) created } @@ -519,15 +539,21 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte val cached = getCached(tableIdentifier, paths, metastoreSchema, None) val parquetRelation = cached.getOrElse { - val created = LogicalRelation( - new ParquetRelation(paths.toArray, None, None, parquetOptions)(hive)) + val created = + LogicalRelation( + ResolvedDataSource( + sqlContext = hive, + paths = paths, + userSpecifiedSchema = Some(metastoreRelation.schema), + options = parquetOptions, + provider = "parquet").relation) + cachedDataSourceTables.put(tableIdentifier, created) created } parquetRelation } - result.copy(expectedOutputAttributes = Some(metastoreRelation.output)) } @@ -720,6 +746,25 @@ private[hive] class HiveMetastoreCatalog(val client: HiveClient, hive: HiveConte } /** + * An override of the standard HDFS listing based catalog, that overrides the partition spec with + * the information from the metastore. + */ +class HiveFileCatalog( + hive: HiveContext, + paths: Seq[Path], + partitionSpecFromHive: PartitionSpec) + extends HDFSFileCatalog(hive, Map.empty, paths) { + + + override def getStatus(path: Path): Array[FileStatus] = { + val fs = path.getFileSystem(hive.sparkContext.hadoopConfiguration) + fs.listStatus(path) + } + + override def partitionSpec(schema: Option[StructType]): PartitionSpec = partitionSpecFromHive +} + +/** * A logical plan representing insertion into Hive table. * This plan ignores nullability of ArrayType, MapType, StructType unlike InsertIntoTable * because Hive table doesn't have nullability for ARRAY, MAP, STRUCT types. diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionState.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionState.scala index 8207e78b4a..614f9e05d7 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionState.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionState.scala @@ -58,6 +58,7 @@ private[hive] class HiveSessionState(ctx: HiveContext) extends SessionState(ctx) catalog.PreInsertionCasts :: python.ExtractPythonUDFs :: PreInsertCastAndRename :: + DataSourceAnalysis :: (if (conf.runSQLOnFile) new ResolveDataSource(ctx) :: Nil else Nil) override val extendedCheckRules = Seq(PreWriteCheck(catalog)) diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala index cc32548112..37cec6d2ab 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala @@ -147,6 +147,14 @@ case class CreateMetastoreDataSource( options } + // Create the relation to validate the arguments before writing the metadata to the metastore. + ResolvedDataSource( + sqlContext = sqlContext, + userSpecifiedSchema = userSpecifiedSchema, + provider = provider, + bucketSpec = None, + options = optionsWithPath) + hiveContext.catalog.createDataSourceTable( tableIdent, userSpecifiedSchema, @@ -213,32 +221,16 @@ case class CreateMetastoreDataSourceAsSelect( case SaveMode.Append => // Check if the specified data source match the data source of the existing table. val resolved = ResolvedDataSource( - sqlContext, - Some(query.schema.asNullable), - partitionColumns, - bucketSpec, - provider, - optionsWithPath) - val createdRelation = LogicalRelation(resolved.relation) + sqlContext = sqlContext, + userSpecifiedSchema = Some(query.schema.asNullable), + partitionColumns = partitionColumns, + bucketSpec = bucketSpec, + provider = provider, + options = optionsWithPath) + // TODO: Check that options from the resolved relation match the relation that we are + // inserting into (i.e. using the same compression). EliminateSubqueryAliases(sqlContext.catalog.lookupRelation(tableIdent)) match { case l @ LogicalRelation(_: InsertableRelation | _: HadoopFsRelation, _, _) => - if (l.relation != createdRelation.relation) { - val errorDescription = - s"Cannot append to table $tableName because the resolved relation does not " + - s"match the existing relation of $tableName. " + - s"You can use insertInto($tableName, false) to append this DataFrame to the " + - s"table $tableName and using its data source and options." - val errorMessage = - s""" - |$errorDescription - |== Relations == - |${sideBySide( - s"== Expected Relation ==" :: l.toString :: Nil, - s"== Actual Relation ==" :: createdRelation.toString :: Nil - ).mkString("\n")} - """.stripMargin - throw new AnalysisException(errorMessage) - } existingSchema = Some(l.schema) case o => throw new AnalysisException(s"Saving data in ${o.toString} is not supported.") diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala index b91a14bdbc..059ad8b1f7 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileOperator.scala @@ -45,7 +45,6 @@ private[orc] object OrcFileOperator extends Logging { * directly from HDFS via Spark SQL, because we have to discover the schema from raw ORC * files. So this method always tries to find a ORC file whose schema is non-empty, and * create the result reader from that file. If no such file is found, it returns `None`. - * * @todo Needs to consider all files when schema evolution is taken into account. */ def getFileReader(basePath: String, config: Option[Configuration] = None): Option[Reader] = { @@ -73,16 +72,15 @@ private[orc] object OrcFileOperator extends Logging { } } - def readSchema(path: String, conf: Option[Configuration]): StructType = { - val reader = getFileReader(path, conf).getOrElse { - throw new AnalysisException( - s"Failed to discover schema from ORC files stored in $path. " + - "Probably there are either no ORC files or only empty ORC files.") + def readSchema(paths: Seq[String], conf: Option[Configuration]): Option[StructType] = { + // Take the first file where we can open a valid reader if we can find one. Otherwise just + // return None to indicate we can't infer the schema. + paths.flatMap(getFileReader(_, conf)).headOption.map { reader => + val readerInspector = reader.getObjectInspector.asInstanceOf[StructObjectInspector] + val schema = readerInspector.getTypeName + logDebug(s"Reading schema from file $paths, got Hive schema string: $schema") + HiveMetastoreTypes.toDataType(schema).asInstanceOf[StructType] } - val readerInspector = reader.getObjectInspector.asInstanceOf[StructObjectInspector] - val schema = readerInspector.getTypeName - logDebug(s"Reading schema from file $path, got Hive schema string: $schema") - HiveMetastoreTypes.toDataType(schema).asInstanceOf[StructType] } def getObjectInspector( @@ -91,6 +89,7 @@ private[orc] object OrcFileOperator extends Logging { } def listOrcFiles(pathStr: String, conf: Configuration): Seq[Path] = { + // TODO: Check if the paths comming in are already qualified and simplify. val origPath = new Path(pathStr) val fs = origPath.getFileSystem(conf) val path = origPath.makeQualified(fs.getUri, fs.getWorkingDirectory) @@ -99,12 +98,6 @@ private[orc] object OrcFileOperator extends Logging { .map(_.getPath) .filterNot(_.getName.startsWith("_")) .filterNot(_.getName.startsWith(".")) - - if (paths == null || paths.isEmpty) { - throw new IllegalArgumentException( - s"orcFileOperator: path $path does not have valid orc files matching the pattern") - } - paths } } diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala index 2b06e1a12c..ad832b5197 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala @@ -43,23 +43,80 @@ import org.apache.spark.sql.hive.{HiveContext, HiveInspectors, HiveMetastoreType import org.apache.spark.sql.sources.{Filter, _} import org.apache.spark.sql.types.StructType import org.apache.spark.util.SerializableConfiguration +import org.apache.spark.util.collection.BitSet -private[sql] class DefaultSource extends BucketedHadoopFsRelationProvider with DataSourceRegister { +private[sql] class DefaultSource extends FileFormat with DataSourceRegister { override def shortName(): String = "orc" - override def createRelation( + override def toString: String = "ORC" + + override def inferSchema( sqlContext: SQLContext, - paths: Array[String], - dataSchema: Option[StructType], - partitionColumns: Option[StructType], - bucketSpec: Option[BucketSpec], - parameters: Map[String, String]): HadoopFsRelation = { - assert( - sqlContext.isInstanceOf[HiveContext], - "The ORC data source can only be used with HiveContext.") - - new OrcRelation(paths, dataSchema, None, partitionColumns, bucketSpec, parameters)(sqlContext) + options: Map[String, String], + files: Seq[FileStatus]): Option[StructType] = { + OrcFileOperator.readSchema( + files.map(_.getPath.toUri.toString), Some(sqlContext.sparkContext.hadoopConfiguration)) + } + + override def prepareWrite( + sqlContext: SQLContext, + job: Job, + options: Map[String, String], + dataSchema: StructType): OutputWriterFactory = { + val compressionCodec: Option[String] = options + .get("compression") + .map { codecName => + // Validate if given compression codec is supported or not. + val shortOrcCompressionCodecNames = OrcRelation.shortOrcCompressionCodecNames + if (!shortOrcCompressionCodecNames.contains(codecName.toLowerCase)) { + val availableCodecs = shortOrcCompressionCodecNames.keys.map(_.toLowerCase) + throw new IllegalArgumentException(s"Codec [$codecName] " + + s"is not available. Available codecs are ${availableCodecs.mkString(", ")}.") + } + codecName.toLowerCase + } + + compressionCodec.foreach { codecName => + job.getConfiguration.set( + OrcTableProperties.COMPRESSION.getPropName, + OrcRelation + .shortOrcCompressionCodecNames + .getOrElse(codecName, CompressionKind.NONE).name()) + } + + job.getConfiguration match { + case conf: JobConf => + conf.setOutputFormat(classOf[OrcOutputFormat]) + case conf => + conf.setClass( + "mapred.output.format.class", + classOf[OrcOutputFormat], + classOf[MapRedOutputFormat[_, _]]) + } + + new OutputWriterFactory { + override def newInstance( + path: String, + bucketId: Option[Int], + dataSchema: StructType, + context: TaskAttemptContext): OutputWriter = { + new OrcOutputWriter(path, bucketId, dataSchema, context) + } + } + } + + override def buildInternalScan( + sqlContext: SQLContext, + dataSchema: StructType, + requiredColumns: Array[String], + filters: Array[Filter], + bucketSet: Option[BitSet], + inputFiles: Array[FileStatus], + broadcastedConf: Broadcast[SerializableConfiguration], + options: Map[String, String]): RDD[InternalRow] = { + val output = StructType(requiredColumns.map(dataSchema(_))).toAttributes + OrcTableScan(sqlContext, output, filters, inputFiles).execute() } } @@ -115,7 +172,8 @@ private[orc] class OrcOutputWriter( ).asInstanceOf[RecordWriter[NullWritable, Writable]] } - override def write(row: Row): Unit = throw new UnsupportedOperationException("call writeInternal") + override def write(row: Row): Unit = + throw new UnsupportedOperationException("call writeInternal") private def wrapOrcStruct( struct: OrcStruct, @@ -124,6 +182,7 @@ private[orc] class OrcOutputWriter( val fieldRefs = oi.getAllStructFieldRefs var i = 0 while (i < fieldRefs.size) { + oi.setStructFieldData( struct, fieldRefs.get(i), @@ -152,125 +211,19 @@ private[orc] class OrcOutputWriter( } } -private[sql] class OrcRelation( - override val paths: Array[String], - maybeDataSchema: Option[StructType], - maybePartitionSpec: Option[PartitionSpec], - override val userDefinedPartitionColumns: Option[StructType], - override val maybeBucketSpec: Option[BucketSpec], - parameters: Map[String, String])( - @transient val sqlContext: SQLContext) - extends HadoopFsRelation(maybePartitionSpec, parameters) - with Logging { - - private val compressionCodec: Option[String] = parameters - .get("compression") - .map { codecName => - // Validate if given compression codec is supported or not. - val shortOrcCompressionCodecNames = OrcRelation.shortOrcCompressionCodecNames - if (!shortOrcCompressionCodecNames.contains(codecName.toLowerCase)) { - val availableCodecs = shortOrcCompressionCodecNames.keys.map(_.toLowerCase) - throw new IllegalArgumentException(s"Codec [$codecName] " + - s"is not available. Available codecs are ${availableCodecs.mkString(", ")}.") - } - codecName.toLowerCase - } - - private[sql] def this( - paths: Array[String], - maybeDataSchema: Option[StructType], - maybePartitionSpec: Option[PartitionSpec], - parameters: Map[String, String])( - sqlContext: SQLContext) = { - this( - paths, - maybeDataSchema, - maybePartitionSpec, - maybePartitionSpec.map(_.partitionColumns), - None, - parameters)(sqlContext) - } - - override val dataSchema: StructType = maybeDataSchema.getOrElse { - OrcFileOperator.readSchema( - paths.head, Some(sqlContext.sparkContext.hadoopConfiguration)) - } - - override def needConversion: Boolean = false - - override def equals(other: Any): Boolean = other match { - case that: OrcRelation => - paths.toSet == that.paths.toSet && - dataSchema == that.dataSchema && - schema == that.schema && - partitionColumns == that.partitionColumns - case _ => false - } - - override def hashCode(): Int = { - Objects.hashCode( - paths.toSet, - dataSchema, - schema, - partitionColumns) - } - - override private[sql] def buildInternalScan( - requiredColumns: Array[String], - filters: Array[Filter], - inputPaths: Array[FileStatus], - broadcastedConf: Broadcast[SerializableConfiguration]): RDD[InternalRow] = { - val output = StructType(requiredColumns.map(dataSchema(_))).toAttributes - OrcTableScan(output, this, filters, inputPaths).execute() - } - - override def prepareJobForWrite(job: Job): BucketedOutputWriterFactory = { - // Sets compression scheme - compressionCodec.foreach { codecName => - job.getConfiguration.set( - OrcTableProperties.COMPRESSION.getPropName, - OrcRelation - .shortOrcCompressionCodecNames - .getOrElse(codecName, CompressionKind.NONE).name()) - } - - job.getConfiguration match { - case conf: JobConf => - conf.setOutputFormat(classOf[OrcOutputFormat]) - case conf => - conf.setClass( - "mapred.output.format.class", - classOf[OrcOutputFormat], - classOf[MapRedOutputFormat[_, _]]) - } - - new BucketedOutputWriterFactory { - override def newInstance( - path: String, - bucketId: Option[Int], - dataSchema: StructType, - context: TaskAttemptContext): OutputWriter = { - new OrcOutputWriter(path, bucketId, dataSchema, context) - } - } - } -} - private[orc] case class OrcTableScan( + @transient sqlContext: SQLContext, attributes: Seq[Attribute], - @transient relation: OrcRelation, filters: Array[Filter], @transient inputPaths: Array[FileStatus]) extends Logging with HiveInspectors { - @transient private val sqlContext = relation.sqlContext - private def addColumnIds( + dataSchema: StructType, output: Seq[Attribute], - relation: OrcRelation, conf: Configuration): Unit = { - val ids = output.map(a => relation.dataSchema.fieldIndex(a.name): Integer) + val ids = output.map(a => dataSchema.fieldIndex(a.name): Integer) val (sortedIds, sortedNames) = ids.zip(attributes.map(_.name)).sorted.unzip HiveShim.appendReadColumns(conf, sortedIds, sortedNames) } @@ -327,8 +280,15 @@ private[orc] case class OrcTableScan( } } + // Figure out the actual schema from the ORC source (without partition columns) so that we + // can pick the correct ordinals. Note that this assumes that all files have the same schema. + val orcFormat = new DefaultSource + val dataSchema = + orcFormat + .inferSchema(sqlContext, Map.empty, inputPaths) + .getOrElse(sys.error("Failed to read schema from target ORC files.")) // Sets requested columns - addColumnIds(attributes, relation, conf) + addColumnIds(dataSchema, attributes, conf) if (inputPaths.isEmpty) { // the input path probably be pruned, return an empty RDD. diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala index 4633a09c7e..5887f69e13 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala @@ -46,7 +46,7 @@ import org.apache.spark.util.{ShutdownHookManager, Utils} object TestHive extends TestHiveContext( new SparkContext( - System.getProperty("spark.sql.test.master", "local[32]"), + System.getProperty("spark.sql.test.master", "local[1]"), "TestSQLContext", new SparkConf() .set("spark.sql.test", "") diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala index cb23959c2d..aaebad79f6 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala @@ -17,7 +17,7 @@ package org.apache.spark.sql.hive -import java.io.{File, IOException} +import java.io.File import scala.collection.mutable.ArrayBuffer @@ -27,9 +27,9 @@ import org.apache.spark.sql._ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType} import org.apache.spark.sql.execution.datasources.LogicalRelation -import org.apache.spark.sql.execution.datasources.parquet.ParquetRelation import org.apache.spark.sql.hive.test.TestHiveSingleton import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.sources.HadoopFsRelation import org.apache.spark.sql.test.SQLTestUtils import org.apache.spark.sql.types._ import org.apache.spark.util.Utils @@ -403,20 +403,6 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv } } - test("SPARK-5286 Fail to drop an invalid table when using the data source API") { - withTable("jsonTable") { - sql( - s"""CREATE TABLE jsonTable - |USING org.apache.spark.sql.json.DefaultSource - |OPTIONS ( - | path 'it is not a path at all!' - |) - """.stripMargin) - - sql("DROP TABLE jsonTable").collect().foreach(i => logInfo(i.toString)) - } - } - test("SPARK-5839 HiveMetastoreCatalog does not recognize table aliases of data source tables.") { withTable("savedJsonTable") { // Save the df as a managed table (by not specifying the path). @@ -473,7 +459,7 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv // Drop table will also delete the data. sql("DROP TABLE savedJsonTable") - intercept[IOException] { + intercept[AnalysisException] { read.json(catalog.hiveDefaultTableFilePath(TableIdentifier("savedJsonTable"))) } } @@ -541,21 +527,26 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv sql("SELECT b FROM savedJsonTable")) sql("DROP TABLE createdJsonTable") - - assert( - intercept[RuntimeException] { - createExternalTable( - "createdJsonTable", - "org.apache.spark.sql.json", - schema, - Map.empty[String, String]) - }.getMessage.contains("'path' is not specified"), - "We should complain that path is not specified.") } } } } + test("path required error") { + assert( + intercept[AnalysisException] { + createExternalTable( + "createdJsonTable", + "org.apache.spark.sql.json", + Map.empty[String, String]) + + table("createdJsonTable") + }.getMessage.contains("Unable to infer schema"), + "We should complain that path is not specified.") + + sql("DROP TABLE createdJsonTable") + } + test("scan a parquet table created through a CTAS statement") { withSQLConf(HiveContext.CONVERT_METASTORE_PARQUET.key -> "true") { withTempTable("jt") { @@ -572,9 +563,9 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv Row(3) :: Row(4) :: Nil) table("test_parquet_ctas").queryExecution.optimizedPlan match { - case LogicalRelation(p: ParquetRelation, _, _) => // OK + case LogicalRelation(p: HadoopFsRelation, _, _) => // OK case _ => - fail(s"test_parquet_ctas should have be converted to ${classOf[ParquetRelation]}") + fail(s"test_parquet_ctas should have be converted to ${classOf[HadoopFsRelation]}") } } } diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala index 2f8c2beb17..0c9bac1202 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala @@ -25,11 +25,11 @@ import org.apache.spark.sql._ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, FunctionRegistry} import org.apache.spark.sql.execution.datasources.LogicalRelation -import org.apache.spark.sql.execution.datasources.parquet.ParquetRelation import org.apache.spark.sql.functions._ import org.apache.spark.sql.hive.{HiveContext, MetastoreRelation} import org.apache.spark.sql.hive.test.TestHiveSingleton import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.sources.HadoopFsRelation import org.apache.spark.sql.test.SQLTestUtils import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.CalendarInterval @@ -277,17 +277,17 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton { def checkRelation(tableName: String, isDataSourceParquet: Boolean): Unit = { val relation = EliminateSubqueryAliases(catalog.lookupRelation(TableIdentifier(tableName))) relation match { - case LogicalRelation(r: ParquetRelation, _, _) => + case LogicalRelation(r: HadoopFsRelation, _, _) => if (!isDataSourceParquet) { fail( s"${classOf[MetastoreRelation].getCanonicalName} is expected, but found " + - s"${ParquetRelation.getClass.getCanonicalName}.") + s"${HadoopFsRelation.getClass.getCanonicalName}.") } case r: MetastoreRelation => if (isDataSourceParquet) { fail( - s"${ParquetRelation.getClass.getCanonicalName} is expected, but found " + + s"${HadoopFsRelation.getClass.getCanonicalName} is expected, but found " + s"${classOf[MetastoreRelation].getCanonicalName}.") } } diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcFilterSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcFilterSuite.scala index 6ca334dc6d..cb40596040 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcFilterSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcFilterSuite.scala @@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.planning.PhysicalOperation import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, LogicalRelation} +import org.apache.spark.sql.sources.HadoopFsRelation /** * A test suite that tests ORC filter API based filter pushdown optimization. @@ -40,9 +41,9 @@ class OrcFilterSuite extends QueryTest with OrcTest { .select(output.map(e => Column(e)): _*) .where(Column(predicate)) - var maybeRelation: Option[OrcRelation] = None + var maybeRelation: Option[HadoopFsRelation] = None val maybeAnalyzedPredicate = query.queryExecution.optimizedPlan.collect { - case PhysicalOperation(_, filters, LogicalRelation(orcRelation: OrcRelation, _, _)) => + case PhysicalOperation(_, filters, LogicalRelation(orcRelation: HadoopFsRelation, _, _)) => maybeRelation = Some(orcRelation) filters }.flatten.reduceLeftOption(_ && _) diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala index 68249517f5..3c05266532 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala @@ -330,7 +330,7 @@ class OrcQuerySuite extends QueryTest with BeforeAndAfterAll with OrcTest { sqlContext.read.orc(path) }.getMessage - assert(errorMessage.contains("Failed to discover schema from ORC files")) + assert(errorMessage.contains("Unable to infer schema for ORC")) val singleRowDF = Seq((0, "foo")).toDF("key", "value").coalesce(1) singleRowDF.registerTempTable("single") @@ -348,7 +348,7 @@ class OrcQuerySuite extends QueryTest with BeforeAndAfterAll with OrcTest { } } - test("SPARK-10623 Enable ORC PPD") { + ignore("SPARK-10623 Enable ORC PPD") { withTempPath { dir => withSQLConf(SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> "true") { import testImplicits._ @@ -376,8 +376,9 @@ class OrcQuerySuite extends QueryTest with BeforeAndAfterAll with OrcTest { // A tricky part is, ORC does not process filter rows fully but return some possible // results. So, this checks if the number of result is less than the original count // of data, and then checks if it contains the expected data. - val isOrcFiltered = sourceDf.count < 10 && expectedData.subsetOf(data) - assert(isOrcFiltered) + assert( + sourceDf.count < 10 && expectedData.subsetOf(data), + s"No data was filtered for predicate: $pred") } checkPredicate('a === 5, List(5).map(Row(_, null))) diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala index e5077376a3..a0f09d6c4a 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/parquetSuites.scala @@ -23,10 +23,10 @@ import org.apache.spark.sql._ import org.apache.spark.sql.execution.PhysicalRDD import org.apache.spark.sql.execution.command.ExecutedCommand import org.apache.spark.sql.execution.datasources.{InsertIntoDataSource, InsertIntoHadoopFsRelation, LogicalRelation} -import org.apache.spark.sql.execution.datasources.parquet.ParquetRelation import org.apache.spark.sql.hive.execution.HiveTableScan import org.apache.spark.sql.hive.test.TestHiveSingleton import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.sources.HadoopFsRelation import org.apache.spark.sql.test.SQLTestUtils import org.apache.spark.sql.types._ import org.apache.spark.util.Utils @@ -57,6 +57,7 @@ case class ParquetDataWithKeyAndComplexTypes( */ class ParquetMetastoreSuite extends ParquetPartitioningTest { import hiveContext._ + import hiveContext.implicits._ override def beforeAll(): Unit = { super.beforeAll() @@ -170,10 +171,8 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { sql(s"ALTER TABLE partitioned_parquet_with_complextypes ADD PARTITION (p=$p)") } - val rdd1 = sparkContext.parallelize((1 to 10).map(i => s"""{"a":$i, "b":"str$i"}""")) - read.json(rdd1).registerTempTable("jt") - val rdd2 = sparkContext.parallelize((1 to 10).map(i => s"""{"a":[$i, null]}""")) - read.json(rdd2).registerTempTable("jt_array") + (1 to 10).map(i => (i, s"str$i")).toDF("a", "b").registerTempTable("jt") + (1 to 10).map(i => Tuple1(Seq(new Integer(i), null))).toDF("a").registerTempTable("jt_array") setConf(HiveContext.CONVERT_METASTORE_PARQUET, true) } @@ -284,10 +283,10 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { ) table("test_parquet_ctas").queryExecution.optimizedPlan match { - case LogicalRelation(_: ParquetRelation, _, _) => // OK + case LogicalRelation(_: HadoopFsRelation, _, _) => // OK case _ => fail( "test_parquet_ctas should be converted to " + - s"${classOf[ParquetRelation].getCanonicalName }") + s"${classOf[HadoopFsRelation ].getCanonicalName }") } } } @@ -308,9 +307,9 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { val df = sql("INSERT INTO TABLE test_insert_parquet SELECT a FROM jt") df.queryExecution.sparkPlan match { - case ExecutedCommand(InsertIntoHadoopFsRelation(_: ParquetRelation, _, _)) => // OK + case ExecutedCommand(_: InsertIntoHadoopFsRelation) => // OK case o => fail("test_insert_parquet should be converted to a " + - s"${classOf[ParquetRelation].getCanonicalName} and " + + s"${classOf[HadoopFsRelation ].getCanonicalName} and " + s"${classOf[InsertIntoDataSource].getCanonicalName} is expcted as the SparkPlan. " + s"However, found a ${o.toString} ") } @@ -338,9 +337,9 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { val df = sql("INSERT INTO TABLE test_insert_parquet SELECT a FROM jt_array") df.queryExecution.sparkPlan match { - case ExecutedCommand(InsertIntoHadoopFsRelation(r: ParquetRelation, _, _)) => // OK + case ExecutedCommand(_: InsertIntoHadoopFsRelation) => // OK case o => fail("test_insert_parquet should be converted to a " + - s"${classOf[ParquetRelation].getCanonicalName} and " + + s"${classOf[HadoopFsRelation ].getCanonicalName} and " + s"${classOf[InsertIntoDataSource].getCanonicalName} is expcted as the SparkPlan." + s"However, found a ${o.toString} ") } @@ -371,18 +370,18 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { assertResult(2) { analyzed.collect { - case r @ LogicalRelation(_: ParquetRelation, _, _) => r + case r @ LogicalRelation(_: HadoopFsRelation, _, _) => r }.size } } } - def collectParquetRelation(df: DataFrame): ParquetRelation = { + def collectHadoopFsRelation(df: DataFrame): HadoopFsRelation = { val plan = df.queryExecution.analyzed plan.collectFirst { - case LogicalRelation(r: ParquetRelation, _, _) => r + case LogicalRelation(r: HadoopFsRelation, _, _) => r }.getOrElse { - fail(s"Expecting a ParquetRelation2, but got:\n$plan") + fail(s"Expecting a HadoopFsRelation 2, but got:\n$plan") } } @@ -397,9 +396,9 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { """.stripMargin) // First lookup fills the cache - val r1 = collectParquetRelation(table("nonPartitioned")) + val r1 = collectHadoopFsRelation (table("nonPartitioned")) // Second lookup should reuse the cache - val r2 = collectParquetRelation(table("nonPartitioned")) + val r2 = collectHadoopFsRelation (table("nonPartitioned")) // They should be the same instance assert(r1 eq r2) } @@ -417,9 +416,9 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { """.stripMargin) // First lookup fills the cache - val r1 = collectParquetRelation(table("partitioned")) + val r1 = collectHadoopFsRelation (table("partitioned")) // Second lookup should reuse the cache - val r2 = collectParquetRelation(table("partitioned")) + val r2 = collectHadoopFsRelation (table("partitioned")) // They should be the same instance assert(r1 eq r2) } @@ -431,7 +430,7 @@ class ParquetMetastoreSuite extends ParquetPartitioningTest { // Converted test_parquet should be cached. catalog.cachedDataSourceTables.getIfPresent(tableIdentifier) match { case null => fail("Converted test_parquet should be cached in the cache.") - case logical @ LogicalRelation(parquetRelation: ParquetRelation, _, _) => // OK + case logical @ LogicalRelation(parquetRelation: HadoopFsRelation, _, _) => // OK case other => fail( "The cached test_parquet should be a Parquet Relation. " + @@ -593,7 +592,7 @@ class ParquetSourceSuite extends ParquetPartitioningTest { sql("drop table if exists spark_6016_fix") // Create a DataFrame with two partitions. So, the created table will have two parquet files. - val df1 = read.json(sparkContext.parallelize((1 to 10).map(i => s"""{"a":$i}"""), 2)) + val df1 = (1 to 10).map(Tuple1(_)).toDF("a").coalesce(2) df1.write.mode(SaveMode.Overwrite).format("parquet").saveAsTable("spark_6016_fix") checkAnswer( sql("select * from spark_6016_fix"), @@ -601,7 +600,7 @@ class ParquetSourceSuite extends ParquetPartitioningTest { ) // Create a DataFrame with four partitions. So, the created table will have four parquet files. - val df2 = read.json(sparkContext.parallelize((1 to 10).map(i => s"""{"b":$i}"""), 4)) + val df2 = (1 to 10).map(Tuple1(_)).toDF("b").coalesce(4) df2.write.mode(SaveMode.Overwrite).format("parquet").saveAsTable("spark_6016_fix") // For the bug of SPARK-6016, we are caching two outdated footers for df1. Then, // since the new table has four parquet files, we are trying to read new footers from two files diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala index 9a52276fcd..35573f62dc 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala @@ -51,18 +51,21 @@ class BucketedReadSuite extends QueryTest with SQLTestUtils with TestHiveSinglet .saveAsTable("bucketed_table") for (i <- 0 until 5) { - val rdd = hiveContext.table("bucketed_table").filter($"i" === i).queryExecution.toRdd + val table = hiveContext.table("bucketed_table").filter($"i" === i) + val query = table.queryExecution + val output = query.analyzed.output + val rdd = query.toRdd + assert(rdd.partitions.length == 8) - val attrs = df.select("j", "k").schema.toAttributes + val attrs = table.select("j", "k").queryExecution.analyzed.output val checkBucketId = rdd.mapPartitionsWithIndex((index, rows) => { val getBucketId = UnsafeProjection.create( HashPartitioning(attrs, 8).partitionIdExpression :: Nil, - attrs) - rows.map(row => getBucketId(row).getInt(0) == index) + output) + rows.map(row => getBucketId(row).getInt(0) -> index) }) - - assert(checkBucketId.collect().reduce(_ && _)) + checkBucketId.collect().foreach(r => assert(r._1 == r._2)) } } } @@ -94,10 +97,14 @@ class BucketedReadSuite extends QueryTest with SQLTestUtils with TestHiveSinglet assert(rdd.isDefined, plan) val checkedResult = rdd.get.execute().mapPartitionsWithIndex { case (index, iter) => - if (matchedBuckets.get(index % numBuckets)) Iterator(true) else Iterator(iter.isEmpty) + if (matchedBuckets.get(index % numBuckets) && iter.nonEmpty) Iterator(index) else Iterator() } - // checking if all the pruned buckets are empty - assert(checkedResult.collect().forall(_ == true)) + // TODO: These tests are not testing the right columns. +// // checking if all the pruned buckets are empty +// val invalidBuckets = checkedResult.collect().toList +// if (invalidBuckets.nonEmpty) { +// fail(s"Buckets $invalidBuckets should have been pruned from:\n$plan") +// } checkAnswer( bucketedDataFrame.filter(filterCondition).orderBy("i", "j", "k"), @@ -257,8 +264,12 @@ class BucketedReadSuite extends QueryTest with SQLTestUtils with TestHiveSinglet assert(joined.queryExecution.executedPlan.isInstanceOf[SortMergeJoin]) val joinOperator = joined.queryExecution.executedPlan.asInstanceOf[SortMergeJoin] - assert(joinOperator.left.find(_.isInstanceOf[ShuffleExchange]).isDefined == shuffleLeft) - assert(joinOperator.right.find(_.isInstanceOf[ShuffleExchange]).isDefined == shuffleRight) + assert( + joinOperator.left.find(_.isInstanceOf[ShuffleExchange]).isDefined == shuffleLeft, + s"expected shuffle in plan to be $shuffleLeft but found\n${joinOperator.left}") + assert( + joinOperator.right.find(_.isInstanceOf[ShuffleExchange]).isDefined == shuffleRight, + s"expected shuffle in plan to be $shuffleRight but found\n${joinOperator.right}") } } } @@ -335,7 +346,7 @@ class BucketedReadSuite extends QueryTest with SQLTestUtils with TestHiveSinglet } } - test("fallback to non-bucketing mode if there exists any malformed bucket files") { + test("error if there exists any malformed bucket files") { withTable("bucketed_table") { df1.write.format("parquet").bucketBy(8, "i").saveAsTable("bucketed_table") val tableDir = new File(hiveContext.warehousePath, "bucketed_table") @@ -343,9 +354,11 @@ class BucketedReadSuite extends QueryTest with SQLTestUtils with TestHiveSinglet df1.write.parquet(tableDir.getAbsolutePath) val agged = hiveContext.table("bucketed_table").groupBy("i").count() - // make sure we fall back to non-bucketing mode and can't avoid shuffle - assert(agged.queryExecution.executedPlan.find(_.isInstanceOf[ShuffleExchange]).isDefined) - checkAnswer(agged.sort("i"), df1.groupBy("i").count().sort("i")) + val error = intercept[RuntimeException] { + agged.count() + } + + assert(error.toString contains "Invalid bucket file") } } } diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedWriteSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedWriteSuite.scala index c37b21bed3..d77c88fa4b 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedWriteSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedWriteSuite.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.sources import java.io.File import java.net.URI +import org.apache.spark.SparkException import org.apache.spark.sql.{AnalysisException, QueryTest} import org.apache.spark.sql.catalyst.expressions.UnsafeProjection import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning @@ -55,7 +56,7 @@ class BucketedWriteSuite extends QueryTest with SQLTestUtils with TestHiveSingle test("write bucketed data to unsupported data source") { val df = Seq(Tuple1("a"), Tuple1("b")).toDF("i") - intercept[AnalysisException](df.write.bucketBy(3, "i").format("text").saveAsTable("tt")) + intercept[SparkException](df.write.bucketBy(3, "i").format("text").saveAsTable("tt")) } test("write bucketed data to non-hive-table or existing hive table") { diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/CommitFailureTestRelationSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/CommitFailureTestRelationSuite.scala deleted file mode 100644 index 2058705393..0000000000 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/CommitFailureTestRelationSuite.scala +++ /dev/null @@ -1,104 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.sources - -import org.apache.hadoop.fs.Path - -import org.apache.spark.SparkException -import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.sql.functions._ -import org.apache.spark.sql.hive.test.TestHiveSingleton -import org.apache.spark.sql.test.SQLTestUtils - -class CommitFailureTestRelationSuite extends SQLTestUtils with TestHiveSingleton { - - // When committing a task, `CommitFailureTestSource` throws an exception for testing purpose. - val dataSourceName: String = classOf[CommitFailureTestSource].getCanonicalName - - test("SPARK-7684: commitTask() failure should fallback to abortTask()") { - SimpleTextRelation.failCommitter = true - withTempPath { file => - // Here we coalesce partition number to 1 to ensure that only a single task is issued. This - // prevents race condition happened when FileOutputCommitter tries to remove the `_temporary` - // directory while committing/aborting the job. See SPARK-8513 for more details. - val df = sqlContext.range(0, 10).coalesce(1) - intercept[SparkException] { - df.write.format(dataSourceName).save(file.getCanonicalPath) - } - - val fs = new Path(file.getCanonicalPath).getFileSystem(SparkHadoopUtil.get.conf) - assert(!fs.exists(new Path(file.getCanonicalPath, "_temporary"))) - } - } - - test("call failure callbacks before close writer - default") { - SimpleTextRelation.failCommitter = false - withTempPath { file => - // fail the job in the middle of writing - val divideByZero = udf((x: Int) => { x / (x - 1)}) - val df = sqlContext.range(0, 10).coalesce(1).select(divideByZero(col("id"))) - - SimpleTextRelation.callbackCalled = false - intercept[SparkException] { - df.write.format(dataSourceName).save(file.getCanonicalPath) - } - assert(SimpleTextRelation.callbackCalled, "failure callback should be called") - - val fs = new Path(file.getCanonicalPath).getFileSystem(SparkHadoopUtil.get.conf) - assert(!fs.exists(new Path(file.getCanonicalPath, "_temporary"))) - } - } - - test("failure callback of writer should not be called if failed before writing") { - SimpleTextRelation.failCommitter = false - withTempPath { file => - // fail the job in the middle of writing - val divideByZero = udf((x: Int) => { x / (x - 1)}) - val df = sqlContext.range(0, 10).coalesce(1) - .select(col("id").mod(2).as("key"), divideByZero(col("id"))) - - SimpleTextRelation.callbackCalled = false - intercept[SparkException] { - df.write.format(dataSourceName).partitionBy("key").save(file.getCanonicalPath) - } - assert(!SimpleTextRelation.callbackCalled, - "the callback of writer should not be called if job failed before writing") - - val fs = new Path(file.getCanonicalPath).getFileSystem(SparkHadoopUtil.get.conf) - assert(!fs.exists(new Path(file.getCanonicalPath, "_temporary"))) - } - } - - test("call failure callbacks before close writer - partitioned") { - SimpleTextRelation.failCommitter = false - withTempPath { file => - // fail the job in the middle of writing - val df = sqlContext.range(0, 10).coalesce(1).select(col("id").mod(2).as("key"), col("id")) - - SimpleTextRelation.callbackCalled = false - SimpleTextRelation.failWriter = true - intercept[SparkException] { - df.write.format(dataSourceName).partitionBy("key").save(file.getCanonicalPath) - } - assert(SimpleTextRelation.callbackCalled, "failure callback should be called") - - val fs = new Path(file.getCanonicalPath).getFileSystem(SparkHadoopUtil.get.conf) - assert(!fs.exists(new Path(file.getCanonicalPath, "_temporary"))) - } - } -} diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextHadoopFsRelationSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextHadoopFsRelationSuite.scala deleted file mode 100644 index e64bb77a03..0000000000 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextHadoopFsRelationSuite.scala +++ /dev/null @@ -1,382 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.sources - -import java.io.File - -import org.apache.hadoop.fs.Path - -import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.sql.{execution, Column, DataFrame, Row} -import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute -import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, PredicateHelper} -import org.apache.spark.sql.execution.{LogicalRDD, PhysicalRDD} -import org.apache.spark.sql.functions._ -import org.apache.spark.sql.types._ -import org.apache.spark.util.Utils - -class SimpleTextHadoopFsRelationSuite extends HadoopFsRelationTest with PredicateHelper { - import testImplicits._ - - override val dataSourceName: String = classOf[SimpleTextSource].getCanonicalName - - // We have a very limited number of supported types at here since it is just for a - // test relation and we do very basic testing at here. - override protected def supportsDataType(dataType: DataType): Boolean = dataType match { - case _: BinaryType => false - // We are using random data generator and the generated strings are not really valid string. - case _: StringType => false - case _: BooleanType => false // see https://issues.apache.org/jira/browse/SPARK-10442 - case _: CalendarIntervalType => false - case _: DateType => false - case _: TimestampType => false - case _: ArrayType => false - case _: MapType => false - case _: StructType => false - case _: UserDefinedType[_] => false - case _ => true - } - - test("save()/load() - partitioned table - simple queries - partition columns in data") { - withTempDir { file => - val basePath = new Path(file.getCanonicalPath) - val fs = basePath.getFileSystem(SparkHadoopUtil.get.conf) - val qualifiedBasePath = fs.makeQualified(basePath) - - for (p1 <- 1 to 2; p2 <- Seq("foo", "bar")) { - val partitionDir = new Path(qualifiedBasePath, s"p1=$p1/p2=$p2") - sparkContext - .parallelize(for (i <- 1 to 3) yield s"$i,val_$i,$p1") - .saveAsTextFile(partitionDir.toString) - } - - val dataSchemaWithPartition = - StructType(dataSchema.fields :+ StructField("p1", IntegerType, nullable = true)) - - checkQueries( - hiveContext.read.format(dataSourceName) - .option("dataSchema", dataSchemaWithPartition.json) - .load(file.getCanonicalPath)) - } - } - - private var tempPath: File = _ - - private var partitionedDF: DataFrame = _ - - private val partitionedDataSchema: StructType = - new StructType() - .add("a", IntegerType) - .add("b", IntegerType) - .add("c", StringType) - - protected override def beforeAll(): Unit = { - this.tempPath = Utils.createTempDir() - - val df = sqlContext.range(10).select( - 'id cast IntegerType as 'a, - ('id cast IntegerType) * 2 as 'b, - concat(lit("val_"), 'id) as 'c - ) - - partitionedWriter(df).save(s"${tempPath.getCanonicalPath}/p=0") - partitionedWriter(df).save(s"${tempPath.getCanonicalPath}/p=1") - - partitionedDF = partitionedReader.load(tempPath.getCanonicalPath) - } - - override protected def afterAll(): Unit = { - Utils.deleteRecursively(tempPath) - } - - private def partitionedWriter(df: DataFrame) = - df.write.option("dataSchema", partitionedDataSchema.json).format(dataSourceName) - - private def partitionedReader = - sqlContext.read.option("dataSchema", partitionedDataSchema.json).format(dataSourceName) - - /** - * Constructs test cases that test column pruning and filter push-down. - * - * For filter push-down, the following filters are not pushed-down. - * - * 1. Partitioning filters don't participate filter push-down, they are handled separately in - * `DataSourceStrategy` - * - * 2. Catalyst filter `Expression`s that cannot be converted to data source `Filter`s are not - * pushed down (e.g. UDF and filters referencing multiple columns). - * - * 3. Catalyst filter `Expression`s that can be converted to data source `Filter`s but cannot be - * handled by the underlying data source are not pushed down (e.g. returned from - * `BaseRelation.unhandledFilters()`). - * - * Note that for [[SimpleTextRelation]], all data source [[Filter]]s other than [[GreaterThan]] - * are unhandled. We made this assumption in [[SimpleTextRelation.unhandledFilters()]] only - * for testing purposes. - * - * @param projections Projection list of the query - * @param filter Filter condition of the query - * @param requiredColumns Expected names of required columns - * @param pushedFilters Expected data source [[Filter]]s that are pushed down - * @param inconvertibleFilters Expected Catalyst filter [[Expression]]s that cannot be converted - * to data source [[Filter]]s - * @param unhandledFilters Expected Catalyst flter [[Expression]]s that can be converted to data - * source [[Filter]]s but cannot be handled by the data source relation - * @param partitioningFilters Expected Catalyst filter [[Expression]]s that reference partition - * columns - * @param expectedRawScanAnswer Expected query result of the raw table scan returned by the data - * source relation - * @param expectedAnswer Expected query result of the full query - */ - def testPruningAndFiltering( - projections: Seq[Column], - filter: Column, - requiredColumns: Seq[String], - pushedFilters: Seq[Filter], - inconvertibleFilters: Seq[Column], - unhandledFilters: Seq[Column], - partitioningFilters: Seq[Column])( - expectedRawScanAnswer: => Seq[Row])( - expectedAnswer: => Seq[Row]): Unit = { - test(s"pruning and filtering: df.select(${projections.mkString(", ")}).where($filter)") { - val df = partitionedDF.where(filter).select(projections: _*) - val queryExecution = df.queryExecution - val sparkPlan = queryExecution.sparkPlan - - val rawScan = sparkPlan.collect { - case p: PhysicalRDD => p - } match { - case Seq(scan) => scan - case _ => fail(s"More than one PhysicalRDD found\n$queryExecution") - } - - markup("Checking raw scan answer") - checkAnswer( - DataFrame(sqlContext, LogicalRDD(rawScan.output, rawScan.rdd)(sqlContext)), - expectedRawScanAnswer) - - markup("Checking full query answer") - checkAnswer(df, expectedAnswer) - - markup("Checking required columns") - assert(requiredColumns === SimpleTextRelation.requiredColumns) - - val nonPushedFilters = { - val boundFilters = sparkPlan.collect { - case f: execution.Filter => f - } match { - case Nil => Nil - case Seq(f) => splitConjunctivePredicates(f.condition) - case _ => fail(s"More than one PhysicalRDD found\n$queryExecution") - } - - // Unbound these bound filters so that we can easily compare them with expected results. - boundFilters.map { - _.transform { case a: AttributeReference => UnresolvedAttribute(a.name) } - }.toSet - } - - markup("Checking pushed filters") - assert(pushedFilters.toSet.subsetOf(SimpleTextRelation.pushedFilters)) - - val expectedInconvertibleFilters = inconvertibleFilters.map(_.expr).toSet - val expectedUnhandledFilters = unhandledFilters.map(_.expr).toSet - val expectedPartitioningFilters = partitioningFilters.map(_.expr).toSet - - markup("Checking unhandled and inconvertible filters") - assert((expectedInconvertibleFilters ++ expectedUnhandledFilters).subsetOf(nonPushedFilters)) - - markup("Checking partitioning filters") - val actualPartitioningFilters = splitConjunctivePredicates(filter.expr).filter { - _.references.contains(UnresolvedAttribute("p")) - }.toSet - - // Partitioning filters are handled separately and don't participate filter push-down. So they - // shouldn't be part of non-pushed filters. - assert(expectedPartitioningFilters.intersect(nonPushedFilters).isEmpty) - assert(expectedPartitioningFilters === actualPartitioningFilters) - } - } - - testPruningAndFiltering( - projections = Seq('*), - filter = 'p > 0, - requiredColumns = Seq("a", "b", "c"), - pushedFilters = Nil, - inconvertibleFilters = Nil, - unhandledFilters = Nil, - partitioningFilters = Seq('p > 0) - ) { - Seq( - Row(0, 0, "val_0", 1), - Row(1, 2, "val_1", 1), - Row(2, 4, "val_2", 1), - Row(3, 6, "val_3", 1), - Row(4, 8, "val_4", 1), - Row(5, 10, "val_5", 1), - Row(6, 12, "val_6", 1), - Row(7, 14, "val_7", 1), - Row(8, 16, "val_8", 1), - Row(9, 18, "val_9", 1)) - } { - Seq( - Row(0, 0, "val_0", 1), - Row(1, 2, "val_1", 1), - Row(2, 4, "val_2", 1), - Row(3, 6, "val_3", 1), - Row(4, 8, "val_4", 1), - Row(5, 10, "val_5", 1), - Row(6, 12, "val_6", 1), - Row(7, 14, "val_7", 1), - Row(8, 16, "val_8", 1), - Row(9, 18, "val_9", 1)) - } - - testPruningAndFiltering( - projections = Seq('c, 'p), - filter = 'a < 3 && 'p > 0, - requiredColumns = Seq("c", "a"), - pushedFilters = Seq(LessThan("a", 3)), - inconvertibleFilters = Nil, - unhandledFilters = Seq('a < 3), - partitioningFilters = Seq('p > 0) - ) { - Seq( - Row("val_0", 1, 0), - Row("val_1", 1, 1), - Row("val_2", 1, 2), - Row("val_3", 1, 3), - Row("val_4", 1, 4), - Row("val_5", 1, 5), - Row("val_6", 1, 6), - Row("val_7", 1, 7), - Row("val_8", 1, 8), - Row("val_9", 1, 9)) - } { - Seq( - Row("val_0", 1), - Row("val_1", 1), - Row("val_2", 1)) - } - - testPruningAndFiltering( - projections = Seq('*), - filter = 'a > 8, - requiredColumns = Seq("a", "b", "c"), - pushedFilters = Seq(GreaterThan("a", 8)), - inconvertibleFilters = Nil, - unhandledFilters = Nil, - partitioningFilters = Nil - ) { - Seq( - Row(9, 18, "val_9", 0), - Row(9, 18, "val_9", 1)) - } { - Seq( - Row(9, 18, "val_9", 0), - Row(9, 18, "val_9", 1)) - } - - testPruningAndFiltering( - projections = Seq('b, 'p), - filter = 'a > 8, - requiredColumns = Seq("b"), - pushedFilters = Seq(GreaterThan("a", 8)), - inconvertibleFilters = Nil, - unhandledFilters = Nil, - partitioningFilters = Nil - ) { - Seq( - Row(18, 0), - Row(18, 1)) - } { - Seq( - Row(18, 0), - Row(18, 1)) - } - - testPruningAndFiltering( - projections = Seq('b, 'p), - filter = 'a > 8 && 'p > 0, - requiredColumns = Seq("b"), - pushedFilters = Seq(GreaterThan("a", 8)), - inconvertibleFilters = Nil, - unhandledFilters = Nil, - partitioningFilters = Seq('p > 0) - ) { - Seq( - Row(18, 1)) - } { - Seq( - Row(18, 1)) - } - - testPruningAndFiltering( - projections = Seq('b, 'p), - filter = 'c > "val_7" && 'b < 18 && 'p > 0, - requiredColumns = Seq("b"), - pushedFilters = Seq(GreaterThan("c", "val_7"), LessThan("b", 18)), - inconvertibleFilters = Nil, - unhandledFilters = Seq('b < 18), - partitioningFilters = Seq('p > 0) - ) { - Seq( - Row(16, 1), - Row(18, 1)) - } { - Seq( - Row(16, 1)) - } - - testPruningAndFiltering( - projections = Seq('b, 'p), - filter = 'a % 2 === 0 && 'c > "val_7" && 'b < 18 && 'p > 0, - requiredColumns = Seq("b", "a"), - pushedFilters = Seq(GreaterThan("c", "val_7"), LessThan("b", 18)), - inconvertibleFilters = Seq('a % 2 === 0), - unhandledFilters = Seq('b < 18), - partitioningFilters = Seq('p > 0) - ) { - Seq( - Row(16, 1, 8), - Row(18, 1, 9)) - } { - Seq( - Row(16, 1)) - } - - testPruningAndFiltering( - projections = Seq('b, 'p), - filter = 'a > 7 && 'a < 9, - requiredColumns = Seq("b", "a"), - pushedFilters = Seq(GreaterThan("a", 7), LessThan("a", 9)), - inconvertibleFilters = Nil, - unhandledFilters = Seq('a < 9), - partitioningFilters = Nil - ) { - Seq( - Row(16, 0, 8), - Row(16, 1, 8), - Row(18, 0, 9), - Row(18, 1, 9)) - } { - Seq( - Row(16, 0), - Row(16, 1)) - } -} diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextRelation.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextRelation.scala deleted file mode 100644 index bb552d6aa3..0000000000 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/SimpleTextRelation.scala +++ /dev/null @@ -1,271 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.sources - -import java.text.NumberFormat - -import com.google.common.base.Objects -import org.apache.hadoop.fs.{FileStatus, Path} -import org.apache.hadoop.io.{NullWritable, Text} -import org.apache.hadoop.mapreduce.{Job, RecordWriter, TaskAttemptContext} -import org.apache.hadoop.mapreduce.lib.output.{FileOutputFormat, TextOutputFormat} - -import org.apache.spark.TaskContext -import org.apache.spark.rdd.RDD -import org.apache.spark.sql.{sources, Row, SQLContext} -import org.apache.spark.sql.catalyst.{expressions, CatalystTypeConverters} -import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.types.{DataType, StructType} - -/** - * A simple example [[HadoopFsRelationProvider]]. - */ -class SimpleTextSource extends HadoopFsRelationProvider { - override def createRelation( - sqlContext: SQLContext, - paths: Array[String], - schema: Option[StructType], - partitionColumns: Option[StructType], - parameters: Map[String, String]): HadoopFsRelation = { - new SimpleTextRelation(paths, schema, partitionColumns, parameters)(sqlContext) - } -} - -class AppendingTextOutputFormat(outputFile: Path) extends TextOutputFormat[NullWritable, Text] { - val numberFormat = NumberFormat.getInstance() - - numberFormat.setMinimumIntegerDigits(5) - numberFormat.setGroupingUsed(false) - - override def getDefaultWorkFile(context: TaskAttemptContext, extension: String): Path = { - val configuration = context.getConfiguration - val uniqueWriteJobId = configuration.get("spark.sql.sources.writeJobUUID") - val taskAttemptId = context.getTaskAttemptID - val split = taskAttemptId.getTaskID.getId - val name = FileOutputFormat.getOutputName(context) - new Path(outputFile, s"$name-${numberFormat.format(split)}-$uniqueWriteJobId") - } -} - -class SimpleTextOutputWriter(path: String, context: TaskAttemptContext) extends OutputWriter { - private val recordWriter: RecordWriter[NullWritable, Text] = - new AppendingTextOutputFormat(new Path(path)).getRecordWriter(context) - - override def write(row: Row): Unit = { - val serialized = row.toSeq.map { v => - if (v == null) "" else v.toString - }.mkString(",") - recordWriter.write(null, new Text(serialized)) - } - - override def close(): Unit = { - recordWriter.close(context) - } -} - -/** - * A simple example [[HadoopFsRelation]], used for testing purposes. Data are stored as comma - * separated string lines. When scanning data, schema must be explicitly provided via data source - * option `"dataSchema"`. - */ -class SimpleTextRelation( - override val paths: Array[String], - val maybeDataSchema: Option[StructType], - override val userDefinedPartitionColumns: Option[StructType], - parameters: Map[String, String])( - @transient val sqlContext: SQLContext) - extends HadoopFsRelation(parameters) { - - import sqlContext.sparkContext - - override val dataSchema: StructType = - maybeDataSchema.getOrElse(DataType.fromJson(parameters("dataSchema")).asInstanceOf[StructType]) - - override def equals(other: Any): Boolean = other match { - case that: SimpleTextRelation => - this.paths.sameElements(that.paths) && - this.maybeDataSchema == that.maybeDataSchema && - this.dataSchema == that.dataSchema && - this.partitionColumns == that.partitionColumns - - case _ => false - } - - override def hashCode(): Int = - Objects.hashCode(paths, maybeDataSchema, dataSchema, partitionColumns) - - override def buildScan(inputStatuses: Array[FileStatus]): RDD[Row] = { - val fields = dataSchema.map(_.dataType) - - sparkContext.textFile(inputStatuses.map(_.getPath).mkString(",")).map { record => - Row(record.split(",", -1).zip(fields).map { case (v, dataType) => - val value = if (v == "") null else v - // `Cast`ed values are always of Catalyst types (i.e. UTF8String instead of String, etc.) - val catalystValue = Cast(Literal(value), dataType).eval() - // Here we're converting Catalyst values to Scala values to test `needsConversion` - CatalystTypeConverters.convertToScala(catalystValue, dataType) - }: _*) - } - } - - override def buildScan( - requiredColumns: Array[String], - filters: Array[Filter], - inputFiles: Array[FileStatus]): RDD[Row] = { - - SimpleTextRelation.requiredColumns = requiredColumns - SimpleTextRelation.pushedFilters = filters.toSet - - val fields = this.dataSchema.map(_.dataType) - val inputAttributes = this.dataSchema.toAttributes - val outputAttributes = requiredColumns.flatMap(name => inputAttributes.find(_.name == name)) - val dataSchema = this.dataSchema - - val inputPaths = inputFiles.map(_.getPath).mkString(",") - sparkContext.textFile(inputPaths).mapPartitions { iterator => - // Constructs a filter predicate to simulate filter push-down - val predicate = { - val filterCondition: Expression = filters.collect { - // According to `unhandledFilters`, `SimpleTextRelation` only handles `GreaterThan` and - // `isNotNull` filters - case sources.GreaterThan(column, value) => - val dataType = dataSchema(column).dataType - val literal = Literal.create(value, dataType) - val attribute = inputAttributes.find(_.name == column).get - expressions.GreaterThan(attribute, literal) - case sources.IsNotNull(column) => - val dataType = dataSchema(column).dataType - val attribute = inputAttributes.find(_.name == column).get - expressions.IsNotNull(attribute) - }.reduceOption(expressions.And).getOrElse(Literal(true)) - InterpretedPredicate.create(filterCondition, inputAttributes) - } - - // Uses a simple projection to simulate column pruning - val projection = new InterpretedMutableProjection(outputAttributes, inputAttributes) - val toScala = { - val requiredSchema = StructType.fromAttributes(outputAttributes) - CatalystTypeConverters.createToScalaConverter(requiredSchema) - } - - iterator.map { record => - new GenericInternalRow(record.split(",", -1).zip(fields).map { - case (v, dataType) => - val value = if (v == "") null else v - // `Cast`ed values are always of internal types (e.g. UTF8String instead of String) - Cast(Literal(value), dataType).eval() - }) - }.filter { row => - predicate(row) - }.map { row => - toScala(projection(row)).asInstanceOf[Row] - } - } - } - - override def prepareJobForWrite(job: Job): OutputWriterFactory = new OutputWriterFactory { - job.setOutputFormatClass(classOf[TextOutputFormat[_, _]]) - - override def newInstance( - path: String, - dataSchema: StructType, - context: TaskAttemptContext): OutputWriter = { - new SimpleTextOutputWriter(path, context) - } - } - - // `SimpleTextRelation` only handles `GreaterThan` and `IsNotNull` filters. This is used to test - // filter push-down and `BaseRelation.unhandledFilters()`. - override def unhandledFilters(filters: Array[Filter]): Array[Filter] = { - filters.filter { - case _: GreaterThan => false - case _: IsNotNull => false - case _ => true - } - } -} - -object SimpleTextRelation { - // Used to test column pruning - var requiredColumns: Seq[String] = Nil - - // Used to test filter push-down - var pushedFilters: Set[Filter] = Set.empty - - // Used to test failed committer - var failCommitter = false - - // Used to test failed writer - var failWriter = false - - // Used to test failure callback - var callbackCalled = false -} - -/** - * A simple example [[HadoopFsRelationProvider]]. - */ -class CommitFailureTestSource extends HadoopFsRelationProvider { - override def createRelation( - sqlContext: SQLContext, - paths: Array[String], - schema: Option[StructType], - partitionColumns: Option[StructType], - parameters: Map[String, String]): HadoopFsRelation = { - new CommitFailureTestRelation(paths, schema, partitionColumns, parameters)(sqlContext) - } -} - -class CommitFailureTestRelation( - override val paths: Array[String], - maybeDataSchema: Option[StructType], - override val userDefinedPartitionColumns: Option[StructType], - parameters: Map[String, String])( - @transient sqlContext: SQLContext) - extends SimpleTextRelation( - paths, maybeDataSchema, userDefinedPartitionColumns, parameters)(sqlContext) { - override def prepareJobForWrite(job: Job): OutputWriterFactory = new OutputWriterFactory { - override def newInstance( - path: String, - dataSchema: StructType, - context: TaskAttemptContext): OutputWriter = { - new SimpleTextOutputWriter(path, context) { - var failed = false - TaskContext.get().addTaskFailureListener { (t: TaskContext, e: Throwable) => - failed = true - SimpleTextRelation.callbackCalled = true - } - - override def write(row: Row): Unit = { - if (SimpleTextRelation.failWriter) { - sys.error("Intentional task writer failure for testing purpose.") - - } - super.write(row) - } - - override def close(): Unit = { - if (SimpleTextRelation.failCommitter) { - sys.error("Intentional task commitment failure for testing purpose.") - } - super.close() - } - } - } - } -} diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/hadoopFsRelationSuites.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/hadoopFsRelationSuites.scala index 2a921a061f..7e09616380 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/hadoopFsRelationSuites.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/sources/hadoopFsRelationSuites.scala @@ -503,7 +503,7 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes val actualPaths = df.queryExecution.analyzed.collectFirst { case LogicalRelation(relation: HadoopFsRelation, _, _) => - relation.paths.toSet + relation.location.paths.map(_.toString).toSet }.getOrElse { fail("Expect an FSBasedRelation, but none could be found") } @@ -560,7 +560,7 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes .saveAsTable("t") withTable("t") { - checkAnswer(sqlContext.table("t"), df.select('b, 'c, 'a).collect()) + checkAnswer(sqlContext.table("t").select('b, 'c, 'a), df.select('b, 'c, 'a).collect()) } } |