You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2023/04/12 03:03:49 UTC
[spark] branch master updated: [SPARK-43039][SQL] Support custom fields in the file source _metadata column
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new a31ac0492a5 [SPARK-43039][SQL] Support custom fields in the file source _metadata column
a31ac0492a5 is described below
commit a31ac0492a57e74aed42144aeb34d9c9f08a0595
Author: Ryan Johnson <ry...@databricks.com>
AuthorDate: Wed Apr 12 11:03:28 2023 +0800
[SPARK-43039][SQL] Support custom fields in the file source _metadata column
### What changes were proposed in this pull request?
Allow `FileFormat` instances to define the schema of the `_metadata` column they expose.
### Why are the changes needed?
Today, the schema of the file source `_metadata` column depends on the file format (e.g. parquet file format supports `_metadata.row_index`) but this is hard-wired into the `FileFormat` itself. Not only is this an ugly design, it also prevents custom file formats from adding their own fields to the `_metadata` column.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit tests.
Closes #40677 from ryan-johnson-databricks/generalized-metadata-cols.
Authored-by: Ryan Johnson <ry...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../catalyst/expressions/namedExpressions.scala | 57 +++-
.../sql/catalyst/types/PhysicalDataType.scala | 2 +-
.../spark/sql/execution/DataSourceScanExec.scala | 5 +-
.../spark/sql/execution/PartitionedFileUtil.scala | 13 +-
.../sql/execution/datasources/FileFormat.scala | 78 ++---
.../sql/execution/datasources/FileIndex.scala | 21 +-
.../sql/execution/datasources/FileScanRDD.scala | 49 ++-
.../execution/datasources/FileSourceStrategy.scala | 22 +-
.../execution/datasources/LogicalRelation.scala | 3 +-
.../datasources/PartitioningAwareFileIndex.scala | 4 +-
.../sql/execution/datasources/RowIndexUtil.scala | 40 ---
.../datasources/parquet/ParquetFileFormat.scala | 14 +
.../datasources/parquet/ParquetRowIndexUtil.scala | 26 +-
.../datasources/v2/parquet/ParquetScan.scala | 6 +-
.../execution/streaming/StreamingRelation.scala | 3 +-
.../FileSourceCustomMetadataStructSuite.scala | 354 +++++++++++++++++++++
.../ParquetFileMetadataStructRowIndexSuite.scala} | 59 ++--
.../datasources/parquet/ParquetRowIndexSuite.scala | 8 +-
.../spark/sql/hive/HiveMetastoreCatalog.scala | 2 +-
19 files changed, 591 insertions(+), 175 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala
index 246ecbd6158..048f379f8a3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala
@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.plans.logical.EventTimeWatermark
import org.apache.spark.sql.catalyst.trees.TreePattern
import org.apache.spark.sql.catalyst.trees.TreePattern._
+import org.apache.spark.sql.catalyst.types._
import org.apache.spark.sql.catalyst.util.{quoteIfNeeded, METADATA_COL_ATTR_KEY}
import org.apache.spark.sql.types._
import org.apache.spark.util.collection.BitSet
@@ -554,6 +555,27 @@ object FileSourceMetadataAttribute {
metadata.getBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY)
}
+ /**
+ * True if the given data type is supported in file source metadata attributes.
+ *
+ * The set of supported types is limited by [[ColumnVectorUtils.populate]], which the constant
+ * file metadata implementation relies on. In general, types that can be partition columns are
+ * supported (including most primitive types). Notably unsupported types include [[ObjectType]],
+ * [[UserDefinedType]], and the complex types ([[StructType]], [[MapType]], [[ArrayType]]).
+ */
+ def isSupportedType(dataType: DataType): Boolean = dataType.physicalDataType match {
+ // PhysicalPrimitiveType covers: Boolean, Byte, Double, Float, Integer, Long, Null, Short
+ case _: PhysicalPrimitiveType | _: PhysicalDecimalType => true
+ case PhysicalBinaryType | PhysicalStringType | PhysicalCalendarIntervalType => true
+ case _ => false
+ }
+
+ /** Returns the type unchanged if valid; otherwise throws [[IllegalArgumentException]]. */
+ def validateType(dataType: DataType): DataType = {
+ require(isSupportedType(dataType), s"Unsupported data type: $dataType")
+ dataType
+ }
+
private def removeInternalMetadata(metadata: Metadata) = new MetadataBuilder()
.withMetadata(metadata)
.remove(METADATA_COL_ATTR_KEY)
@@ -574,17 +596,18 @@ object FileSourceConstantMetadataStructField {
/** Constructs a new metadata struct field of the given type; nullable by default */
def apply(name: String, dataType: DataType, nullable: Boolean = true): StructField =
- StructField(name, dataType, nullable, metadata(name))
+ StructField(name, FileSourceMetadataAttribute.validateType(dataType), nullable, metadata(name))
def unapply(field: StructField): Option[StructField] =
- if (isValid(field.metadata)) Some(field) else None
+ if (isValid(field.dataType, field.metadata)) Some(field) else None
def metadata(name: String): Metadata = new MetadataBuilder()
.withMetadata(FileSourceMetadataAttribute.metadata(name))
.putBoolean(FILE_SOURCE_CONSTANT_METADATA_COL_ATTR_KEY, value = true)
.build()
- def isValid(metadata: Metadata): Boolean = {
+ def isValid(dataType: DataType, metadata: Metadata): Boolean = {
+ FileSourceMetadataAttribute.isSupportedType(dataType) &&
FileSourceMetadataAttribute.isValid(metadata) &&
metadata.contains(FILE_SOURCE_CONSTANT_METADATA_COL_ATTR_KEY) &&
metadata.getBoolean(FILE_SOURCE_CONSTANT_METADATA_COL_ATTR_KEY)
@@ -597,8 +620,10 @@ object FileSourceConstantMetadataStructField {
* usually appended to the output and not generated per row.
*/
object FileSourceConstantMetadataAttribute {
- def unapply(attr: AttributeReference): Option[AttributeReference] =
- if (FileSourceConstantMetadataStructField.isValid(attr.metadata)) Some(attr) else None
+ def unapply(attr: AttributeReference): Option[AttributeReference] = {
+ val valid = FileSourceConstantMetadataStructField.isValid(attr.dataType, attr.metadata)
+ if (valid) Some(attr) else None
+ }
}
/**
@@ -626,24 +651,27 @@ object FileSourceGeneratedMetadataStructField {
name: String,
internalName: String,
dataType: DataType,
- nullable: Boolean = true): StructField =
- StructField(name, dataType, nullable, metadata(name, internalName))
+ nullable: Boolean = true): StructField = {
+ val dt = FileSourceMetadataAttribute.validateType(dataType)
+ StructField(name, dt, nullable, metadata(name, internalName))
+ }
def unapply(field: StructField): Option[(StructField, String)] =
- getInternalNameIfValid(field.metadata).map(field -> _)
+ getInternalNameIfValid(field.dataType, field.metadata).map(field -> _)
def metadata(name: String, internalName: String): Metadata = new MetadataBuilder()
.withMetadata(FileSourceMetadataAttribute.metadata(name))
.putString(FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY, internalName)
.build()
- def isValid(metadata: Metadata): Boolean = {
+ def isValid(dataType: DataType, metadata: Metadata): Boolean = {
+ FileSourceMetadataAttribute.isSupportedType(dataType) &&
FileSourceMetadataAttribute.isValid(metadata) &&
metadata.contains(FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY)
}
- def getInternalNameIfValid(metadata: Metadata): Option[String] = {
- if (isValid(metadata)) {
+ def getInternalNameIfValid(dataType: DataType, metadata: Metadata): Option[String] = {
+ if (isValid(dataType, metadata)) {
Some(metadata.getString(FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY))
} else None
}
@@ -654,6 +682,9 @@ object FileSourceGeneratedMetadataStructField {
* the `metadata` struct that maps to some internal column the scanner returns.
*/
object FileSourceGeneratedMetadataAttribute {
- def unapply(attr: AttributeReference): Option[(AttributeReference, String)] =
- FileSourceGeneratedMetadataStructField.getInternalNameIfValid(attr.metadata).map(attr -> _)
+ def unapply(attr: AttributeReference): Option[(AttributeReference, String)] = {
+ FileSourceGeneratedMetadataStructField
+ .getInternalNameIfValid(attr.dataType, attr.metadata)
+ .map(attr -> _)
+ }
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
index 6059ee291f1..d893a5448ec 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/PhysicalDataType.scala
@@ -60,7 +60,7 @@ object PhysicalDataType {
def ordering(dt: DataType): Ordering[Any] = apply(dt).ordering.asInstanceOf[Ordering[Any]]
}
-trait PhysicalPrimitiveType
+sealed trait PhysicalPrimitiveType
sealed abstract class PhysicalNumericType extends PhysicalDataType {
// Unfortunately we can't get this implicitly as that breaks Spark Serialization. In order for
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
index 0f4b8c563d2..0400a2b6abc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partition
import org.apache.spark.sql.catalyst.util.{truncatedString, CaseInsensitiveMap}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources._
-import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat => ParquetSource}
+import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat => ParquetSource, ParquetRowIndexUtil}
import org.apache.spark.sql.execution.datasources.v2.PushedDownOperators
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
import org.apache.spark.sql.execution.vectorized.ConstantColumnVector
@@ -703,7 +703,8 @@ case class FileSourceScanExec(
relation.sparkSession, relation.options, filePath) &&
// SPARK-39634: Allow file splitting in combination with row index generation once
// the fix for PARQUET-2161 is available.
- !RowIndexUtil.isNeededForSchema(requiredSchema)
+ (!relation.fileFormat.isInstanceOf[ParquetSource]
+ || !ParquetRowIndexUtil.isNeededForSchema(requiredSchema))
PartitionedFileUtil.splitFiles(
sparkSession = relation.sparkSession,
file = file,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/PartitionedFileUtil.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/PartitionedFileUtil.scala
index fd5f2f25c0b..dcc2114b54d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/PartitionedFileUtil.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/PartitionedFileUtil.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.execution.datasources._
object PartitionedFileUtil {
def splitFiles(
sparkSession: SparkSession,
- file: FileStatus,
+ file: FileStatusWithMetadata,
filePath: Path,
isSplitable: Boolean,
maxSplitBytes: Long,
@@ -36,9 +36,9 @@ object PartitionedFileUtil {
(0L until file.getLen by maxSplitBytes).map { offset =>
val remaining = file.getLen - offset
val size = if (remaining > maxSplitBytes) maxSplitBytes else remaining
- val hosts = getBlockHosts(getBlockLocations(file), offset, size)
+ val hosts = getBlockHosts(getBlockLocations(file.fileStatus), offset, size)
PartitionedFile(partitionValues, SparkPath.fromPath(filePath), offset, size, hosts,
- file.getModificationTime, file.getLen)
+ file.getModificationTime, file.getLen, file.metadata)
}
} else {
Seq(getPartitionedFile(file, filePath, partitionValues))
@@ -46,18 +46,19 @@ object PartitionedFileUtil {
}
def getPartitionedFile(
- file: FileStatus,
+ file: FileStatusWithMetadata,
filePath: Path,
partitionValues: InternalRow): PartitionedFile = {
- val hosts = getBlockHosts(getBlockLocations(file), 0, file.getLen)
+ val hosts = getBlockHosts(getBlockLocations(file.fileStatus), 0, file.getLen)
PartitionedFile(partitionValues, SparkPath.fromPath(filePath), 0, file.getLen, hosts,
- file.getModificationTime, file.getLen)
+ file.getModificationTime, file.getLen, file.metadata)
}
private def getBlockLocations(file: FileStatus): Array[BlockLocation] = file match {
case f: LocatedFileStatus => f.getBlockLocations
case f => Array.empty[BlockLocation]
}
+
// Given locations of all blocks of a single file, `blockLocations`, and an `(offset, length)`
// pair that represents a segment of the same file, find out the block that contains the largest
// fraction the segment, and returns location hosts of that block. If no such block can be found,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala
index daaebf738c8..72e3e1e2406 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala
@@ -27,10 +27,9 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.Filter
-import org.apache.spark.sql.types.{DataType, LongType, StringType, StructField, StructType, TimestampType}
+import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
@@ -165,6 +164,17 @@ trait FileFormat {
}
}
+ /**
+ * Create a file metadata struct column containing fields supported by the given file format.
+ */
+ def createFileMetadataCol(): AttributeReference = {
+ // Strip out the fields' metadata to avoid exposing it to the user. [[FileSourceStrategy]]
+ // avoids confusion by mapping back to [[metadataSchemaFields]].
+ val fields = metadataSchemaFields
+ .map(FileSourceMetadataAttribute.cleanupFileSourceMetadataInformation)
+ FileSourceMetadataAttribute(FileFormat.METADATA_NAME, StructType(fields), nullable = false)
+ }
+
/**
* Returns whether this format supports the given [[DataType]] in read/write path.
* By default all data types are supported.
@@ -176,6 +186,23 @@ trait FileFormat {
* By default all field name is supported.
*/
def supportFieldName(name: String): Boolean = true
+
+ /**
+ * All fields the file format's _metadata struct defines.
+ *
+ * Each field's metadata should define [[METADATA_COL_ATTR_KEY]],
+ * [[FILE_SOURCE_METADATA_COL_ATTR_KEY]], and either
+ * [[FILE_SOURCE_CONSTANT_METADATA_COL_ATTR_KEY]] or
+ * [[FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY]] as appropriate.
+ *
+ * Constant attributes will be extracted automatically from
+ * [[PartitionedFile.extraConstantMetadataColumnValues]], while generated metadata columns always
+ * map to some hidden/internal column the underslying reader provides.
+ *
+ * NOTE: It is not possible to change the semantics of the base metadata fields by overriding this
+ * method. Technically, a file format could choose suppress them, but that is not recommended.
+ */
+ def metadataSchemaFields: Seq[StructField] = FileFormat.BASE_METADATA_FIELDS
}
object FileFormat {
@@ -192,15 +219,6 @@ object FileFormat {
val FILE_MODIFICATION_TIME = "file_modification_time"
- val ROW_INDEX = "row_index"
-
- // A name for a temporary column that holds row indexes computed by the file format reader
- // until they can be placed in the _metadata struct.
- val ROW_INDEX_TEMPORARY_COLUMN_NAME = s"_tmp_metadata_$ROW_INDEX"
-
- val ROW_INDEX_FIELD = FileSourceGeneratedMetadataStructField(
- ROW_INDEX, ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = false)
-
val METADATA_NAME = "_metadata"
/**
@@ -223,27 +241,6 @@ object FileFormat {
FileSourceConstantMetadataStructField(FILE_BLOCK_LENGTH, LongType, nullable = false),
FileSourceConstantMetadataStructField(FILE_MODIFICATION_TIME, TimestampType, nullable = false))
- /**
- * Supported metadata fields of the given [[FileFormat]].
- */
- def metadataSchemaFields(fileFormat: FileFormat): Seq[StructField] = fileFormat match {
- case _: ParquetFileFormat =>
- BASE_METADATA_FIELDS :+ ROW_INDEX_FIELD
- case _ =>
- BASE_METADATA_FIELDS
- }
-
- /**
- * Create a file metadata struct column containing fields supported by the given [[FileFormat]].
- */
- def createFileMetadataCol(fileFormat: FileFormat): AttributeReference = {
- // Strip out the fields' metadata to avoid exposing it to the user. [[FileSourceStrategy]]
- // avoids confusion by mapping back to [[metadataSchemaFields]].
- val fields = metadataSchemaFields(fileFormat)
- .map(FileSourceMetadataAttribute.cleanupFileSourceMetadataInformation)
- FileSourceMetadataAttribute(FileFormat.METADATA_NAME, StructType(fields))
- }
-
// create an internal row given required metadata fields and file information
def createMetadataInternalRow(
fieldNames: Seq[String],
@@ -253,7 +250,7 @@ object FileFormat {
// We are not aware of `FILE_BLOCK_START` and `FILE_BLOCK_LENGTH` before splitting files
assert(!fieldNames.contains(FILE_BLOCK_START) && !fieldNames.contains(FILE_BLOCK_LENGTH))
updateMetadataInternalRow(new GenericInternalRow(fieldNames.length), fieldNames,
- filePath, fileSize, 0L, fileSize, fileModificationTime)
+ filePath, fileSize, 0L, fileSize, fileModificationTime, Map.empty)
}
// update an internal row given required metadata fields and file information
@@ -264,9 +261,11 @@ object FileFormat {
fileSize: Long,
fileBlockStart: Long,
fileBlockLength: Long,
- fileModificationTime: Long): InternalRow = {
+ fileModificationTime: Long,
+ otherConstantMetadataColumnValues: Map[String, Any]): InternalRow = {
fieldNames.zipWithIndex.foreach { case (name, i) =>
name match {
+ // NOTE: The base metadata fields are hard-wired here and cannot be overridden.
case FILE_PATH => row.update(i, UTF8String.fromString(filePath.toString))
case FILE_NAME => row.update(i, UTF8String.fromString(filePath.getName))
case FILE_SIZE => row.update(i, fileSize)
@@ -276,10 +275,13 @@ object FileFormat {
// the modificationTime from the file is in millisecond,
// while internally, the TimestampType `file_modification_time` is stored in microsecond
row.update(i, fileModificationTime * 1000L)
- case ROW_INDEX =>
- // Do nothing. Only the metadata fields that have identical values for each row of the
- // file are set by this function, while fields that have different values (such as row
- // index) are set separately.
+ case other =>
+ // Other metadata columns use the file-provided value (if any). Automatically convert raw
+ // values (including nulls) to literals as a courtesy.
+ Literal(otherConstantMetadataColumnValues.get(other).orNull) match {
+ case Literal(null, _) => row.setNullAt(i)
+ case literal => row.update(i, literal.value)
+ }
}
}
row
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala
index 57521ef5a59..5331f29bc5b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndex.scala
@@ -23,11 +23,30 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.types.StructType
+/**
+ * A file status augmented with optional metadata, which tasks and file readers can use however they
+ * see fit. For example, a custom [[FileIndex]] and [[FileFormat]] working together could expose
+ * this extra metadata as file-constant fields of the file source metadata column.
+ */
+case class FileStatusWithMetadata(fileStatus: FileStatus, metadata: Map[String, Any] = Map.empty) {
+ // Wrapper methods to improve source compatibility in code that still expects a [[FileStatus]].
+ def getPath: Path = fileStatus.getPath
+ def getLen: Long = fileStatus.getLen
+ def getModificationTime: Long = fileStatus.getModificationTime
+ def isDirectory: Boolean = fileStatus.isDirectory
+}
+
/**
* A collection of data files from a partitioned relation, along with the partition values in the
* form of an [[InternalRow]].
*/
-case class PartitionDirectory(values: InternalRow, files: Seq[FileStatus])
+case class PartitionDirectory(values: InternalRow, files: Seq[FileStatusWithMetadata])
+
+object PartitionDirectory {
+ // For backward compat with code that does not know about extra file metadata
+ def apply(values: InternalRow, files: Array[FileStatus]): PartitionDirectory =
+ PartitionDirectory(values, files.map(FileStatusWithMetadata(_)))
+}
/**
* An interface for objects capable of enumerating the root paths of a relation as well as the
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala
index 7fb2d9c8ac7..706228e656b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala
@@ -30,12 +30,12 @@ import org.apache.spark.paths.SparkPath
import org.apache.spark.rdd.{InputFileBlockHolder, RDD}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericInternalRow, JoinedRow, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericInternalRow, JoinedRow, Literal, UnsafeProjection, UnsafeRow}
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources.FileFormat._
-import org.apache.spark.sql.execution.vectorized.ConstantColumnVector
-import org.apache.spark.sql.types.{LongType, StringType, StructType}
+import org.apache.spark.sql.execution.vectorized.{ColumnVectorUtils, ConstantColumnVector}
+import org.apache.spark.sql.types._
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.NextIterator
@@ -50,6 +50,7 @@ import org.apache.spark.util.NextIterator
* @param length number of bytes to read.
* @param modificationTime The modification time of the input file, in milliseconds.
* @param fileSize The length of the input file (not the block), in bytes.
+ * @param otherConstantMetadataColumnValues The values of any additional constant metadata columns.
*/
case class PartitionedFile(
partitionValues: InternalRow,
@@ -58,7 +59,8 @@ case class PartitionedFile(
length: Long,
@transient locations: Array[String] = Array.empty,
modificationTime: Long = 0L,
- fileSize: Long = 0L) {
+ fileSize: Long = 0L,
+ otherConstantMetadataColumnValues: Map[String, Any] = Map.empty) {
def pathUri: URI = filePath.toUri
def toPath: Path = filePath.toPath
@@ -71,6 +73,7 @@ case class PartitionedFile(
/**
* An RDD that scans a list of file partitions.
+ * @param metadataColumns File-constant metadata columns to append to end of schema.
*/
class FileScanRDD(
@transient private val sparkSession: SparkSession,
@@ -148,7 +151,7 @@ class FileScanRDD(
if (metadataColumns.nonEmpty && currentFile != null) {
updateMetadataInternalRow(metadataRow, metadataColumns.map(_.name),
currentFile.toPath, currentFile.fileSize, currentFile.start, currentFile.length,
- currentFile.modificationTime)
+ currentFile.modificationTime, currentFile.otherConstantMetadataColumnValues)
}
/**
@@ -156,33 +159,55 @@ class FileScanRDD(
*/
private def createMetadataColumnVector(c: ColumnarBatch): Array[ColumnVector] = {
val path = currentFile.toPath
- metadataColumns.map(_.name).map {
- case FILE_PATH =>
+ lazy val tmpRow = new GenericInternalRow(1) // for populating custom metadata fields
+ metadataColumns.map(a => (a.name, a.dataType)).map {
+ case (FILE_PATH, dataType) =>
+ require(dataType == StringType)
val columnVector = new ConstantColumnVector(c.numRows(), StringType)
columnVector.setUtf8String(UTF8String.fromString(path.toString))
columnVector
- case FILE_NAME =>
+ case (FILE_NAME, dataType) =>
+ require(dataType == StringType)
val columnVector = new ConstantColumnVector(c.numRows(), StringType)
columnVector.setUtf8String(UTF8String.fromString(path.getName))
columnVector
- case FILE_SIZE =>
+ case (FILE_SIZE, dataType) =>
+ require(dataType == LongType)
val columnVector = new ConstantColumnVector(c.numRows(), LongType)
columnVector.setLong(currentFile.fileSize)
columnVector
- case FILE_BLOCK_START =>
+ case (FILE_BLOCK_START, dataType) =>
+ require(dataType == LongType)
val columnVector = new ConstantColumnVector(c.numRows(), LongType)
columnVector.setLong(currentFile.start)
columnVector
- case FILE_BLOCK_LENGTH =>
+ case (FILE_BLOCK_LENGTH, dataType) =>
+ require(dataType == LongType)
val columnVector = new ConstantColumnVector(c.numRows(), LongType)
columnVector.setLong(currentFile.length)
columnVector
- case FILE_MODIFICATION_TIME =>
+ case (FILE_MODIFICATION_TIME, dataType) =>
+ require(dataType == TimestampType)
val columnVector = new ConstantColumnVector(c.numRows(), LongType)
// the modificationTime from the file is in millisecond,
// while internally, the TimestampType is stored in microsecond
columnVector.setLong(currentFile.modificationTime * 1000L)
columnVector
+ case (other, dataType: DataType) =>
+ // Other metadata columns use the file-provided value, if one exists. Automatically
+ // convert raw values (including nulls) to literals as a courtesy, then populate the
+ // column by passing the resulting value through the `tmpRow` we allocated above.
+ Literal(currentFile.otherConstantMetadataColumnValues.get(other).orNull) match {
+ case Literal(null, _) =>
+ tmpRow.setNullAt(0)
+ case literal =>
+ require(dataType == literal.dataType)
+ tmpRow.update(0, literal.value)
+ }
+
+ val columnVector = new ConstantColumnVector(c.numRows(), dataType)
+ ColumnVectorUtils.populate(columnVector, tmpRow, 0)
+ columnVector
}.toArray
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
index 0263af16470..1819e1b1722 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala
@@ -216,19 +216,17 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
// Metadata attributes are part of a column of type struct up to this point. Here we extract
// this column from the schema and specify a matcher for that.
object MetadataStructColumn {
+ // The column returned by [[FileFormat.createFileMetadataCol]] is sanitized and lacks
+ // the internal metadata we rely on here. Map back to the real fields by field name.
+ lazy val availableMetadataFields = fsRelation.fileFormat.metadataSchemaFields
+ .map(field => field.name.toLowerCase(Locale.ROOT) -> field).toMap
+
def unapply(attributeReference: AttributeReference): Option[AttributeReference] = {
attributeReference match {
case attr @ FileSourceMetadataAttribute(
MetadataAttributeWithLogicalName(
AttributeReference(_, schema: StructType, _, _),
FileFormat.METADATA_NAME)) =>
- // The column returned by [[FileFormat.createFileMetadataCol]] is sanitized and lacks
- // the internal metadata we rely on here. Map back to the real fields by field name,
- // and restore their missing metadata.
- val availableMetadataFields = FileFormat.metadataSchemaFields(fsRelation.fileFormat)
- .map(field => field.name.toLowerCase(Locale.ROOT) -> field)
- .toMap
-
val adjustedFields = schema.fields.map { field =>
val metadata = availableMetadataFields(field.name.toLowerCase(Locale.ROOT)).metadata
field.copy(metadata = metadata)
@@ -243,14 +241,8 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
case MetadataStructColumn(attr) => attr
}
- // We divide metadata columns into two categories: constant and generated.
- // For constant metadata columns, we create these attributes as non-nullable
- // when passing to readers, since the values are always available.
- // For generated metadata columns, they are set as nullable when passed to readers,
- // as the values will be null when trying to read the missing column from the file.
- // They are then replaced by the actual values later in the process.
- // We then restore the specified nullability in the metadata projection node below.
- // Also remember the attribute for each column name, so we can easily map back to it.
+ // Track constant and generated columns separately, because they are used differently in code
+ // below. Also remember the attribute for each logical column name, so we can map back to it.
val constantMetadataColumns = mutable.Buffer.empty[Attribute]
val generatedMetadataColumns = mutable.Buffer.empty[Attribute]
val metadataColumnsByName = mutable.Map.empty[String, Attribute]
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala
index 4176389c58f..bff054fe783 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala
@@ -70,8 +70,7 @@ case class LogicalRelation(
override lazy val metadataOutput: Seq[AttributeReference] = relation match {
case relation: HadoopFsRelation =>
- metadataOutputWithOutConflicts(
- Seq(FileFormat.createFileMetadataCol(relation.fileFormat)))
+ metadataOutputWithOutConflicts(Seq(relation.fileFormat.createFileMetadataCol))
case _ => Nil
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
index e81882ca6c2..2aa5d30946c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
@@ -113,7 +113,7 @@ abstract class PartitioningAwareFileIndex(
}
val selectedPartitions = if (partitionSpec().partitionColumns.isEmpty) {
- PartitionDirectory(InternalRow.empty, allFiles()
+ PartitionDirectory(InternalRow.empty, allFiles().toArray
.filter(f => isNonEmptyFile(f) && matchFileMetadataPredicate(f))) :: Nil
} else {
if (recursiveFileLookup) {
@@ -132,7 +132,7 @@ abstract class PartitioningAwareFileIndex(
// Directory does not exist, or has no children files
Nil
}
- PartitionDirectory(values, files)
+ PartitionDirectory(values, files.toArray)
}
}
logTrace("Selected files after partition pruning:\n\t" + selectedPartitions.mkString("\n\t"))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/RowIndexUtil.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/RowIndexUtil.scala
deleted file mode 100644
index 1512b6da1e8..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/RowIndexUtil.scala
+++ /dev/null
@@ -1,40 +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.execution.datasources
-
-import org.apache.spark.sql.types.{LongType, StructField, StructType}
-
-
-object RowIndexUtil {
- def findRowIndexColumnIndexInSchema(sparkSchema: StructType): Int = {
- sparkSchema.fields.zipWithIndex.find { case (field: StructField, _: Int) =>
- field.name == FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
- } match {
- case Some((field: StructField, idx: Int)) =>
- if (field.dataType != LongType) {
- throw new RuntimeException(s"${FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME} must be of " +
- "LongType")
- }
- idx
- case _ => -1
- }
- }
-
- def isNeededForSchema(sparkSchema: StructType): Boolean = {
- findRowIndexColumnIndexInSchema(sparkSchema) >= 0
- }
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index dc1c9b27b9e..44c334c5452 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -350,9 +350,23 @@ class ParquetFileFormat
case _ => false
}
+
+ override def metadataSchemaFields: Seq[StructField] = {
+ super.metadataSchemaFields :+ ParquetFileFormat.ROW_INDEX_FIELD
+ }
}
object ParquetFileFormat extends Logging {
+ val ROW_INDEX = "row_index"
+
+ // A name for a temporary column that holds row indexes computed by the file format reader
+ // until they can be placed in the _metadata struct.
+ val ROW_INDEX_TEMPORARY_COLUMN_NAME = s"_tmp_metadata_$ROW_INDEX"
+
+ // The field readers can use to access the generated row index column.
+ val ROW_INDEX_FIELD = FileSourceGeneratedMetadataStructField(
+ ROW_INDEX, ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = false)
+
private[parquet] def readSchema(
footers: Seq[Footer], sparkSession: SparkSession): Option[StructType] = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexUtil.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexUtil.scala
index fb1f6a417f4..b1cd2ebee42 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexUtil.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexUtil.scala
@@ -25,10 +25,8 @@ import org.apache.parquet.column.page.PageReadStore
import org.apache.parquet.hadoop.ParquetRecordReader
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.execution.datasources.{FileFormat, RowIndexUtil}
-import org.apache.spark.sql.execution.datasources.RowIndexUtil.findRowIndexColumnIndexInSchema
import org.apache.spark.sql.execution.vectorized.WritableColumnVector
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{LongType, StructField, StructType}
object ParquetRowIndexUtil {
@@ -106,7 +104,7 @@ object ParquetRowIndexUtil {
def addRowIndexToRecordReaderIfNeeded(
reader: ParquetRecordReader[InternalRow],
sparkSchema: StructType): RecordReader[Void, InternalRow] = {
- val rowIndexColumnIdx = RowIndexUtil.findRowIndexColumnIndexInSchema(sparkSchema)
+ val rowIndexColumnIdx = findRowIndexColumnIndexInSchema(sparkSchema)
if (rowIndexColumnIdx >= 0) {
new RecordReaderWithRowIndexes(reader, rowIndexColumnIdx)
} else {
@@ -114,7 +112,25 @@ object ParquetRowIndexUtil {
}
}
+ def findRowIndexColumnIndexInSchema(sparkSchema: StructType): Int = {
+ sparkSchema.fields.zipWithIndex.find { case (field: StructField, _: Int) =>
+ field.name == ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
+ } match {
+ case Some((field: StructField, idx: Int)) =>
+ if (field.dataType != LongType) {
+ throw new RuntimeException(s"${ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME} " +
+ "must be of LongType")
+ }
+ idx
+ case _ => -1
+ }
+ }
+
+ def isNeededForSchema(sparkSchema: StructType): Boolean = {
+ findRowIndexColumnIndexInSchema(sparkSchema) >= 0
+ }
+
def isRowIndexColumn(column: ParquetColumn): Boolean = {
- column.path.length == 1 && column.path.last == FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
+ column.path.length == 1 && column.path.last == ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
index feca878498d..c7819d39cbc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
@@ -26,8 +26,8 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.connector.expressions.aggregate.Aggregation
import org.apache.spark.sql.connector.read.PartitionReaderFactory
-import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex, RowIndexUtil}
-import org.apache.spark.sql.execution.datasources.parquet.{ParquetOptions, ParquetReadSupport, ParquetWriteSupport}
+import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils, PartitioningAwareFileIndex}
+import org.apache.spark.sql.execution.datasources.parquet.{ParquetOptions, ParquetReadSupport, ParquetRowIndexUtil, ParquetWriteSupport}
import org.apache.spark.sql.execution.datasources.v2.FileScan
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.Filter
@@ -53,7 +53,7 @@ case class ParquetScan(
pushedAggregate.isEmpty &&
// SPARK-39634: Allow file splitting in combination with row index generation once
// the fix for PARQUET-2161 is available.
- !RowIndexUtil.isNeededForSchema(readSchema)
+ !ParquetRowIndexUtil.isNeededForSchema(readSchema)
}
override def readSchema(): StructType = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingRelation.scala
index 50da63489d7..977dbe3f4ef 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingRelation.scala
@@ -63,8 +63,7 @@ case class StreamingRelation(dataSource: DataSource, sourceName: String, output:
// If the dataSource provided class is a same or subclass of FileFormat class
case f if classOf[FileFormat].isAssignableFrom(f) =>
metadataOutputWithOutConflicts(
- Seq(FileFormat.createFileMetadataCol(
- dataSource.providingInstance().asInstanceOf[FileFormat])))
+ Seq(dataSource.providingInstance().asInstanceOf[FileFormat].createFileMetadataCol))
case _ => Nil
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala
new file mode 100644
index 00000000000..210535ba62b
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala
@@ -0,0 +1,354 @@
+/*
+ * 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.execution.datasources
+
+import java.io.File
+
+import org.apache.hadoop.fs.{FileStatus, Path}
+
+import org.apache.spark.sql.{DataFrame, Dataset, QueryTest, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Expression, FileSourceConstantMetadataStructField, FileSourceGeneratedMetadataStructField, Literal}
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.functions.{col, lit, when}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+/** Verifies the ability for a FileFormat to define custom metadata types */
+class FileSourceCustomMetadataStructSuite extends QueryTest with SharedSparkSession {
+ import FileSourceCustomMetadataStructSuite._
+
+ val extraConstantMetadataFields = Seq(
+ FileSourceConstantMetadataStructField("foo", IntegerType),
+ FileSourceConstantMetadataStructField("bar", StringType))
+
+ // Shamelessly recycle the parquet row index generated field.
+ val extraGeneratedMetadataFields = Seq(ParquetFileFormat.ROW_INDEX_FIELD.copy(name = "baz"))
+
+ val FILE_SCHEMA = new StructType()
+ .add("fileNum", IntegerType)
+ .add("x", LongType)
+
+ def withTempData(
+ testFileFormat: String, fileSchema: StructType)(
+ f: (DataFrame, FileStatus, FileStatus) => Unit): Unit = {
+
+ withTempDir { dir =>
+ val f0 = new File(dir, "data/f0").getCanonicalPath
+ spark.range(101, 103)
+ .select(lit(0).as("fileNum"), col("id").as("x"), when(col("id") % 2 === 0, lit(9)).as("y"))
+ .coalesce(1)
+ .write.format(testFileFormat).save(f0)
+
+ val f1 = new File(dir, "data/f1").getCanonicalPath
+ spark.range(111, 113)
+ .select(lit(1).as("fileNum"), col("id").as("x"), when(col("id") % 2 === 0, lit(7)).as("y"))
+ .coalesce(1)
+ .write.format(testFileFormat).save(f1)
+
+ // 2. read both f0 and f1
+ val df = spark.read.format(testFileFormat).schema(fileSchema)
+ .load(new File(dir, "data").getCanonicalPath + "/*")
+
+ val hadoopConf = spark.sessionState.newHadoopConfWithOptions(Map.empty)
+ def getFileStatus(subdir: String): FileStatus = {
+ val file = new File(dir, subdir)
+ .listFiles()
+ .filter(_.getName.endsWith(s".$testFileFormat"))
+ .head
+ val path = new Path(file.toURI)
+ val fs = path.getFileSystem(hadoopConf)
+ fs.getFileStatus(path)
+ }
+
+ val realF0 = getFileStatus("data/f0")
+ val realF1 = getFileStatus("data/f1")
+ assert(
+ df.inputFiles.map(new Path(_).getName).toSet ==
+ Seq(realF0, realF1).map(_.getPath.getName).toSet)
+ f(df, realF0, realF1)
+ }
+ }
+
+ def createDF(format: FileFormat, files: Seq[FileStatusWithMetadata]): DataFrame = {
+ val index = TestFileIndex(files)
+ val fsRelation = HadoopFsRelation(
+ index, index.partitionSchema, FILE_SCHEMA, None, format, Map.empty)(spark)
+ Dataset.ofRows(spark, LogicalRelation(fsRelation))
+ }
+
+ test("extra constant metadata fields") {
+ withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+ val format = new TestFileFormat(extraConstantMetadataFields)
+ val files = Seq(
+ // NOTE: The implementation accepts both raw values and literals for any field, and
+ // StringType fields can be String or UTF8String.
+ FileStatusWithMetadata(f0, Map("foo" -> 0, "bar" -> UTF8String.fromString("000"))),
+ FileStatusWithMetadata(f1, Map("foo" -> Literal(1), "bar" -> "111")))
+ val df = createDF(format, files)
+
+ // Query in declared order
+ checkAnswer(
+ df.select("fileNum", "x", "_metadata.row_index", "_metadata.foo", "_metadata.bar"),
+ Seq(
+ Row(0, 101L, 0L, 0, "000"),
+ Row(0, 102L, 1L, 0, "000"),
+ Row(1, 111L, 0L, 1, "111"),
+ Row(1, 112L, 1L, 1, "111")))
+
+ // Query in permuted order
+ checkAnswer(
+ df.select("_metadata.bar", "fileNum", "_metadata.row_index", "_metadata.foo", "x"),
+ Seq(
+ Row("000", 0, 0L, 0, 101L),
+ Row("000", 0, 1L, 0, 102L),
+ Row("111", 1, 0L, 1, 111L),
+ Row("111", 1, 1L, 1, 112L)))
+
+ // Query with repeated metadata fields
+ checkAnswer(
+ df.select(
+ "_metadata.bar", "_metadata.foo", "fileNum", "x", "_metadata.row_index",
+ "_metadata.foo", "_metadata.bar"),
+ Seq(
+ Row("000", 0, 0, 101L, 0L, 0, "000"),
+ Row("000", 0, 0, 102L, 1L, 0, "000"),
+ Row("111", 1, 1, 111L, 0L, 1, "111"),
+ Row("111", 1, 1, 112L, 1L, 1, "111")))
+ }
+ }
+
+ test("filters and projections on extra constant metadata fields") {
+ withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+ val format = new TestFileFormat(extraConstantMetadataFields)
+ val files = Seq(
+ FileStatusWithMetadata(f0, Map("foo" -> 0, "bar" -> "000")),
+ FileStatusWithMetadata(f1, Map("foo" -> 1, "bar" -> "111")))
+ val df = createDF(format, files)
+
+ checkAnswer(
+ df.select(
+ col("fileNum"), col("x"), col("_metadata.row_index"),
+ col("_metadata.foo").as("foofoo"), col("_metadata.bar").as("barbar"))
+ .where("foofoo != 10 and barbar != '999'"),
+ Seq(
+ Row(0, 101L, 0L, 0, "000"),
+ Row(0, 102L, 1L, 0, "000"),
+ Row(1, 111L, 0L, 1, "111"),
+ Row(1, 112L, 1L, 1, "111")))
+
+ checkAnswer(
+ df.select(
+ col("fileNum"), col("x"), col("_metadata.row_index"),
+ col("_metadata.foo").as("foofoo"), col("_metadata.bar").as("barbar"))
+ .where("foofoo != 0"),
+ Seq(
+ Row(1, 111L, 0L, 1, "111"),
+ Row(1, 112L, 1L, 1, "111")))
+
+ checkAnswer(
+ df.select(
+ col("fileNum"), col("x"), col("_metadata.row_index"),
+ col("_metadata.foo").as("foofoo"), col("_metadata.bar").as("barbar"))
+ .where("barbar != '111'"),
+ Seq(
+ Row(0, 101L, 0L, 0, "000"),
+ Row(0, 102L, 1L, 0, "000")))
+ }
+ }
+
+ test("nullable extra constant metadata fields") {
+ withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+ val format = new TestFileFormat(extraConstantMetadataFields)
+ val files = Seq(
+ FileStatusWithMetadata(f0, Map("foo" -> 0)), // no entry for bar
+ FileStatusWithMetadata(f1, Map("foo" -> null, "bar" -> "111"))) // set foo null
+ val df = createDF(format, files)
+
+ // Query in declared order
+ checkAnswer(
+ df.select("fileNum", "x", "_metadata.row_index", "_metadata.foo", "_metadata.bar"),
+ Seq(
+ Row(0, 101L, 0L, 0, null),
+ Row(0, 102L, 1L, 0, null),
+ Row(1, 111L, 0L, null, "111"),
+ Row(1, 112L, 1L, null, "111")))
+
+ // Query in permuted order
+ checkAnswer(
+ df.select("_metadata.bar", "fileNum", "_metadata.row_index", "_metadata.foo", "x"),
+ Seq(
+ Row(null, 0, 0L, 0, 101L),
+ Row(null, 0, 1L, 0, 102L),
+ Row("111", 1, 0L, null, 111L),
+ Row("111", 1, 1L, null, 112L)))
+
+ // Query with repeated metadata fields
+ checkAnswer(
+ df.select(
+ "_metadata.bar", "_metadata.foo", "fileNum", "x", "_metadata.row_index",
+ "_metadata.foo", "_metadata.bar"),
+ Seq(
+ Row(null, 0, 0, 101L, 0L, 0, null),
+ Row(null, 0, 0, 102L, 1L, 0, null),
+ Row("111", null, 1, 111L, 0L, null, "111"),
+ Row("111", null, 1, 112L, 1L, null, "111")))
+ }
+ }
+
+ test("extra generated metadata field") {
+ withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+ val format = new TestFileFormat(extraGeneratedMetadataFields)
+ val files = Seq(
+ // NOTE: The generated column should ignore this mapping
+ FileStatusWithMetadata(f0, Map("baz" -> 1001L)),
+ FileStatusWithMetadata(f1, Map("baz" -> 1002L)))
+ val df = createDF(format, files)
+
+ checkAnswer(
+ df.select("fileNum", "x", "_metadata.baz"),
+ Seq(
+ Row(0, 101L, 0L),
+ Row(0, 102L, 1L),
+ Row(1, 111L, 0L),
+ Row(1, 112L, 1L)))
+ }
+ }
+
+ test("nullable extra generated metadata fields") {
+ val schema = FILE_SCHEMA.add("y", IntegerType)
+ withTempData("parquet", schema) { (_, f0, f1) =>
+ // The column "y" contains some nulls. If we use it as the internal column that backs
+ // _metadata.baz, the latter should still be queryable, and should also return nulls.
+ val extraGeneratedMetadataFields =
+ Seq(FileSourceGeneratedMetadataStructField("baz", "y", IntegerType))
+
+ val format = new TestFileFormat(extraGeneratedMetadataFields)
+ val files = Seq(
+ // NOTE: The generated column should ignore this mapping
+ FileStatusWithMetadata(f0, Map("baz" -> 0)),
+ FileStatusWithMetadata(f1, Map("baz" -> "111")))
+ val df = createDF(format, files)
+
+ // Query in declared order
+ checkAnswer(
+ df.select("fileNum", "x", "_metadata.baz"),
+ Seq(
+ Row(0, 101L, null),
+ Row(0, 102L, 9),
+ Row(1, 111L, null),
+ Row(1, 112L, 7)))
+ }
+ }
+
+ test("filter and projection on extra generated metadata field") {
+ withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+ val format = new TestFileFormat(extraGeneratedMetadataFields)
+ val files = Seq(
+ // NOTE: The generated column should ignore this mapping
+ FileStatusWithMetadata(f0, Map("baz" -> 1001L)),
+ FileStatusWithMetadata(f1, Map("baz" -> 1002L)))
+ val df = createDF(format, files)
+
+ checkAnswer(
+ df.select(col("fileNum"), col("x"), col("_metadata.baz").as("bazbaz"))
+ .where("bazbaz != 999"),
+ Seq(
+ Row(0, 101L, 0L),
+ Row(0, 102L, 1L),
+ Row(1, 111L, 0L),
+ Row(1, 112L, 1L)))
+
+ checkAnswer(
+ df.select(col("fileNum"), col("x"), col("_metadata.baz").as("bazbaz"))
+ .where("bazbaz != 0"),
+ Seq(
+ Row(0, 102L, 1L),
+ Row(1, 112L, 1L)))
+ }
+ }
+
+ test("mix of extra constant and generated metadata fields") {
+ withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+ val format = new TestFileFormat(extraGeneratedMetadataFields ++ extraConstantMetadataFields)
+ val files = Seq(
+ FileStatusWithMetadata(f0, Map("foo" -> 0, "bar" -> "000")),
+ FileStatusWithMetadata(f1, Map("foo" -> 1, "bar" -> "111")))
+ val df = createDF(format, files)
+
+ checkAnswer(
+ df.select("fileNum", "x", "_metadata.foo", "_metadata.bar", "_metadata.baz"),
+ Seq(
+ Row(0, 101L, 0, "000", 0L),
+ Row(0, 102L, 0, "000", 1L),
+ Row(1, 111L, 1, "111", 0L),
+ Row(1, 112L, 1, "111", 1L)))
+ }
+ }
+
+ test("cannot override base metadata fields") {
+ withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+ import FileFormat.{FILE_NAME, FILE_SIZE}
+ import ParquetFileFormat.ROW_INDEX
+
+ val format = new TestFileFormat(extraConstantMetadataFields)
+ val files = Seq(
+ FileStatusWithMetadata(f0, Map(FILE_SIZE -> 0L, FILE_NAME -> "000")),
+ FileStatusWithMetadata(f1, Map(ROW_INDEX -> 17L, FILE_NAME -> "111")))
+ val df = createDF(format, files)
+
+ checkAnswer(
+ df.select(
+ "fileNum", "x", s"_metadata.${ROW_INDEX}", s"_metadata.${FILE_SIZE}",
+ s"_metadata.${FILE_NAME}"),
+ Seq(
+ Row(0, 101L, 0L, f0.getLen, f0.getPath.getName),
+ Row(0, 102L, 1L, f0.getLen, f0.getPath.getName),
+ Row(1, 111L, 0L, f1.getLen, f1.getPath.getName),
+ Row(1, 112L, 1L, f1.getLen, f1.getPath.getName)))
+ }
+ }
+}
+
+object FileSourceCustomMetadataStructSuite {
+ case class TestFileIndex(files: Seq[FileStatusWithMetadata]) extends FileIndex {
+ // The main override of interest for our testing, but we only care about the listing
+ // itself -- partition and data filters are intentionally ignored.
+ override def listFiles(partitionFilters: Seq[Expression], dataFilters: Seq[Expression])
+ : Seq[PartitionDirectory] = {
+ Seq(PartitionDirectory(InternalRow.empty, files))
+ }
+
+ // mildly interesting overrides
+ override def sizeInBytes: Long = files.map(_.getLen).sum
+ override val rootPaths: Seq[Path] = files.map(_.getPath)
+ override def inputFiles: Array[String] = rootPaths.map(_.toString).toArray
+
+ // uninteresting overrides
+ override def refresh(): Unit = {}
+ override def partitionSchema: StructType = new StructType()
+ }
+
+ class TestFileFormat(extraMetadataFields: Seq[StructField]) extends ParquetFileFormat {
+ override def metadataSchemaFields: Seq[StructField] = {
+ super.metadataSchemaFields ++ extraMetadataFields
+ }
+ }
+
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructRowIndexSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileMetadataStructRowIndexSuite.scala
similarity index 78%
rename from sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructRowIndexSuite.scala
rename to sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileMetadataStructRowIndexSuite.scala
index 27a71791669..c10e1799702 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileMetadataStructRowIndexSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileMetadataStructRowIndexSuite.scala
@@ -14,16 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.execution.datasources
+package org.apache.spark.sql.execution.datasources.parquet
import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest}
+import org.apache.spark.sql.execution.datasources.FileFormat
import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{LongType, StructField, StructType}
-class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession {
+class ParquetFileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession {
import testImplicits._
+ import ParquetFileFormat.{ROW_INDEX, ROW_INDEX_TEMPORARY_COLUMN_NAME}
val EXPECTED_ROW_ID_COL = "expected_row_idx"
val EXPECTED_EXTRA_COL = "expected_extra_col"
@@ -61,9 +63,10 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
private val allMetadataCols = Seq(
FileFormat.FILE_PATH,
+ FileFormat.FILE_NAME,
FileFormat.FILE_SIZE,
FileFormat.FILE_MODIFICATION_TIME,
- FileFormat.ROW_INDEX
+ ParquetFileFormat.ROW_INDEX
)
/** Identifies the names of all the metadata columns present in the schema. */
@@ -88,8 +91,8 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> useVectorizedReader.toString,
SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> useOffHeapMemory.toString) {
withReadDataFrame("parquet", partitionCol = "pb") { df =>
- val res = df.select("*", s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX}")
- .where(s"$EXPECTED_ROW_ID_COL != ${FileFormat.ROW_INDEX}")
+ val res = df.select("*", s"${FileFormat.METADATA_NAME}.${ROW_INDEX}")
+ .where(s"$EXPECTED_ROW_ID_COL != ${ROW_INDEX}")
assert(res.count() == 0)
}
}
@@ -102,7 +105,7 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
// `_metadata.row_index` column is present when selecting `_metadata` as a whole.
val metadataCols = collectMetadataCols(withMetadataStruct.schema)
- assert(metadataCols.contains(FileFormat.ROW_INDEX))
+ assert(metadataCols.contains(ROW_INDEX))
}
}
@@ -115,7 +118,7 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
// Schema does not contain row index column, but contains all the remaining metadata columns.
val metadataCols = collectMetadataCols(withMetadataStruct.schema)
- assert(!metadataCols.contains(FileFormat.ROW_INDEX))
+ assert(!metadataCols.contains(ROW_INDEX))
assert(allMetadataCols.intersect(metadataCols).size == allMetadataCols.size - 1)
}
}
@@ -124,7 +127,7 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
withReadDataFrame("orc") { df =>
checkError(
exception = intercept[AnalysisException] {
- df.select("*", s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX}")
+ df.select("*", s"${FileFormat.METADATA_NAME}.${ROW_INDEX}")
},
errorClass = "FIELD_NOT_FOUND",
parameters = Map(
@@ -141,7 +144,7 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> useVectorizedReader.toString) {
withReadDataFrame("parquet") { df =>
val mixedCaseRowIndex = "RoW_InDeX"
- assert(mixedCaseRowIndex.toLowerCase() == FileFormat.ROW_INDEX)
+ assert(mixedCaseRowIndex.toLowerCase() == ROW_INDEX)
assert(df.select("*", s"${FileFormat.METADATA_NAME}.$mixedCaseRowIndex")
.where(s"$EXPECTED_ROW_ID_COL != $mixedCaseRowIndex")
@@ -151,56 +154,56 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
}
}
- test(s"reading ${FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME} - not present in a table") {
+ test(s"reading ${ROW_INDEX_TEMPORARY_COLUMN_NAME} - not present in a table") {
// File format supporting row index generation populates the column with row indexes.
withReadDataFrame("parquet", extraSchemaFields =
- Seq(StructField(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType))) { df =>
+ Seq(StructField(ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType))) { df =>
assert(df
- .where(col(EXPECTED_ROW_ID_COL) === col(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ .where(col(EXPECTED_ROW_ID_COL) === col(ROW_INDEX_TEMPORARY_COLUMN_NAME))
.count == NUM_ROWS)
}
// File format not supporting row index generation populates missing column with nulls.
withReadDataFrame("json", extraSchemaFields =
- Seq(StructField(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType))) { df =>
+ Seq(StructField(ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType))) { df =>
assert(df
- .where(col(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME).isNull)
+ .where(col(ROW_INDEX_TEMPORARY_COLUMN_NAME).isNull)
.count == NUM_ROWS)
}
}
- test(s"reading ${FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME} - present in a table") {
- withReadDataFrame("parquet", extraCol = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME) { df =>
- // Values of FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME column are always populated with
+ test(s"reading ${ROW_INDEX_TEMPORARY_COLUMN_NAME} - present in a table") {
+ withReadDataFrame("parquet", extraCol = ROW_INDEX_TEMPORARY_COLUMN_NAME) { df =>
+ // Values of ROW_INDEX_TEMPORARY_COLUMN_NAME column are always populated with
// generated row indexes, rather than read from the file.
// TODO(SPARK-40059): Allow users to include columns named
- // FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME in their schemas.
+ // ROW_INDEX_TEMPORARY_COLUMN_NAME in their schemas.
assert(df
- .where(col(EXPECTED_ROW_ID_COL) === col(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ .where(col(EXPECTED_ROW_ID_COL) === col(ROW_INDEX_TEMPORARY_COLUMN_NAME))
.count == NUM_ROWS)
// Column cannot be read in combination with _metadata.row_index.
intercept[AnalysisException](df.select("*", FileFormat.METADATA_NAME).collect())
intercept[AnalysisException](df
- .select("*", s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX}").collect())
+ .select("*", s"${FileFormat.METADATA_NAME}.${ROW_INDEX}").collect())
}
}
- test(s"reading ${FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME} - as partition col") {
- withReadDataFrame("parquet", partitionCol = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME) { df =>
+ test(s"reading ${ROW_INDEX_TEMPORARY_COLUMN_NAME} - as partition col") {
+ withReadDataFrame("parquet", partitionCol = ROW_INDEX_TEMPORARY_COLUMN_NAME) { df =>
// Column values are set for each partition, rather than populated with generated row indexes.
assert(df
- .where(col(EXPECTED_PARTITION_COL) === col(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ .where(col(EXPECTED_PARTITION_COL) === col(ROW_INDEX_TEMPORARY_COLUMN_NAME))
.count == NUM_ROWS)
// Column cannot be read in combination with _metadata.row_index.
intercept[AnalysisException](df.select("*", FileFormat.METADATA_NAME).collect())
intercept[AnalysisException](df
- .select("*", s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX}").collect())
+ .select("*", s"${FileFormat.METADATA_NAME}.${ROW_INDEX}").collect())
}
}
- test(s"cannot make ${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} a partition column") {
+ test(s"cannot make ${FileFormat.METADATA_NAME}.${ROW_INDEX} a partition column") {
withTempPath { srcPath =>
spark.range(0, 10, 1, 1).toDF("id").write.parquet(srcPath.getAbsolutePath)
@@ -209,14 +212,14 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
spark.read.parquet(srcPath.getAbsolutePath)
.select("*", FileFormat.METADATA_NAME)
.write
- .partitionBy(s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX}")
+ .partitionBy(s"${FileFormat.METADATA_NAME}.${ROW_INDEX}")
.save(dstPath.getAbsolutePath)
}
}
}
}
- test(s"read user created ${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX} column") {
+ test(s"read user created ${FileFormat.METADATA_NAME}.${ROW_INDEX} column") {
withReadDataFrame("parquet", partitionCol = "pb") { df =>
withTempPath { dir =>
// The `df` has 10 input files with 10 rows each. Therefore the `_metadata.row_index` values
@@ -232,7 +235,7 @@ class FileMetadataStructRowIndexSuite extends QueryTest with SharedSparkSession
// The _metadata.row_index is returning data from the file, not generated metadata.
assert(spark
.read.parquet(dir.getAbsolutePath)
- .select(s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX}")
+ .select(s"${FileFormat.METADATA_NAME}.${ROW_INDEX}")
.distinct.count == NUM_ROWS / 10)
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala
index a12d47a3279..d9ba3456825 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala
@@ -178,9 +178,9 @@ class ParquetRowIndexSuite extends QueryTest with SharedSparkSession {
case _ => false
}
val rowIndexColName = if (rowIndexMetadataColumnSupported) {
- s"${FileFormat.METADATA_NAME}.${FileFormat.ROW_INDEX}"
+ s"${FileFormat.METADATA_NAME}.${ParquetFileFormat.ROW_INDEX}"
} else {
- FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
+ ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
}
val numRecordsPerFile = conf.numRows / conf.numFiles
val (skipCentileFirst, skipCentileMidLeft, skipCentileMidRight, skipCentileLast) =
@@ -306,7 +306,7 @@ class ParquetRowIndexSuite extends QueryTest with SharedSparkSession {
withTempPath{ path =>
val df = spark.range(0, 10, 1, 1).toDF("id")
val schemaWithRowIdx = df.schema
- .add(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, StringType)
+ .add(ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME, StringType)
df.write
.format(conf.writeFormat)
@@ -318,7 +318,7 @@ class ParquetRowIndexSuite extends QueryTest with SharedSparkSession {
.load(path.getAbsolutePath)
val exception = intercept[Exception](dfRead.collect())
- assert(exception.getMessage.contains(FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(exception.getMessage.contains(ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME))
}
}
}
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 4d13c02c503..dbfb4d65bd1 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
@@ -339,7 +339,7 @@ private[hive] class HiveMetastoreCatalog(sparkSession: SparkSession) extends Log
.inferSchema(
sparkSession,
options,
- fileIndex.listFiles(Nil, Nil).flatMap(_.files))
+ fileIndex.listFiles(Nil, Nil).flatMap(_.files).map(_.fileStatus))
.map(mergeWithMetastoreSchema(relation.tableMeta.dataSchema, _))
inferredSchema match {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org