Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/11509#discussion_r55072046
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala ---
@@ -464,215 +378,140 @@ abstract class OutputWriter {
}
}
-/**
- * ::Experimental::
- * A [[BaseRelation]] that provides much of the common code required for
relations that store their
- * data to an HDFS compatible filesystem.
- *
- * For the read path, similar to [[PrunedFilteredScan]], it can eliminate
unneeded columns and
- * filter using selected predicates before producing an RDD containing all
matching tuples as
- * [[Row]] objects. In addition, when reading from Hive style partitioned
tables stored in file
- * systems, it's able to discover partitioning information from the paths
of input directories, and
- * perform partition pruning before start reading the data. Subclasses of
[[HadoopFsRelation()]]
- * must override one of the four `buildScan` methods to implement the read
path.
- *
- * For the write path, it provides the ability to write to both
non-partitioned and partitioned
- * tables. Directory layout of the partitioned tables is compatible with
Hive.
- *
- * @constructor This constructor is for internal uses only. The
[[PartitionSpec]] argument is for
- * implementing metastore table conversion.
- *
- * @param maybePartitionSpec An [[HadoopFsRelation]] can be created with
an optional
- * [[PartitionSpec]], so that partition discovery can be skipped.
- *
- * @since 1.4.0
- */
-@Experimental
-abstract class HadoopFsRelation private[sql](
- maybePartitionSpec: Option[PartitionSpec],
- parameters: Map[String, String])
- extends BaseRelation with FileRelation with Logging {
+case class HadoopFsRelation(
+ sqlContext: SQLContext,
+ location: FileCatalog,
+ partitionSchema: StructType,
+ dataSchema: StructType,
+ bucketSpec: Option[BucketSpec],
+ fileFormat: FileFormat,
+ options: Map[String, String]) extends BaseRelation with FileRelation {
- override def toString: String = getClass.getSimpleName
+ /**
+ * Schema of this relation. It consists of columns appearing in
[[dataSchema]] and all partition
+ * columns not appearing in [[dataSchema]].
+ *
+ * TODO... this is kind of weird since we don't read partition columns
from data when possible
+ *
+ * @since 1.4.0
+ */
+ val schema: StructType = {
+ val dataSchemaColumnNames = dataSchema.map(_.name.toLowerCase).toSet
+ StructType(dataSchema ++ partitionSchema.filterNot { column =>
+ dataSchemaColumnNames.contains(column.name.toLowerCase)
+ })
+ }
- def this() = this(None, Map.empty[String, String])
+ def partitionSchemaOption: Option[StructType] =
+ if (partitionSchema.isEmpty) None else Some(partitionSchema)
+ def partitionSpec: PartitionSpec =
location.partitionSpec(partitionSchemaOption)
- def this(parameters: Map[String, String]) = this(None, parameters)
+ def refresh(): Unit = location.refresh()
- private[sql] def this(maybePartitionSpec: Option[PartitionSpec]) =
- this(maybePartitionSpec, Map.empty[String, String])
+ override def toString: String =
+ s"$fileFormat part: ${partitionSchema.simpleString}, data:
${dataSchema.simpleString}"
- private val hadoopConf = new
Configuration(sqlContext.sparkContext.hadoopConfiguration)
+ /** Returns the list of files that will be read when scanning this
relation. */
+ override def inputFiles: Array[String] =
+ location.allFiles().map(_.getPath.toUri.toString).toArray
+}
- private var _partitionSpec: PartitionSpec = _
+trait FileFormat {
--- End diff --
nit: doc for `FileFormat` and its methods.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
