[SPARK-25348][SQL] Data source for binary files

sql/core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister

@@ -8,3 +8,4 @@ org.apache.spark.sql.execution.datasources.v2.text.TextDataSourceV2
 org.apache.spark.sql.execution.streaming.ConsoleSinkProvider
 org.apache.spark.sql.execution.streaming.sources.RateStreamProvider
 org.apache.spark.sql.execution.streaming.sources.TextSocketSourceProvider
+org.apache.spark.sql.execution.datasources.binaryfile.BinaryFileFormat

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileDataSource.scala

@@ -0,0 +1,48 @@
+/*
+ * 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.binaryfile
+
+import org.apache.spark.sql.types._
+
+/**
+ * `binaryfile` package implements Spark SQL data source API for loading binary file data
+ * as `DataFrame`.
+ *
+ * The loaded `DataFrame` has two columns, the schema is:
+ *  - status: `StructType` (the file status information)
+ *  - content: `BinaryType` (binary data of the file content)
+ *
+ * The schema of "status" column described above is:
+ *  - path: `StringType` (the file path)
+ *  - modification_time: `TimestampType` (last modification time of the file)
+ *  - length: `LongType` (the file length)
+ */
+class BinaryFileDataSource private() {}
+
+object BinaryFileDataSource {
+
+  val fileStatusSchema = StructType(
+    StructField("path", StringType, true) ::
+      StructField("modification_time", TimestampType, true) ::
+      StructField("length", LongType, true) :: Nil)
+
+  val binaryFileSchema = StructType(
+    StructField("status", fileStatusSchema, true) ::
+      StructField("content", BinaryType, true) :: Nil)
+
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

@@ -0,0 +1,120 @@
+/*
+ * 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.binaryfile
+
+import java.sql.Timestamp
+
+import com.google.common.io.{ByteStreams, Closeables}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, GlobFilter, Path}
+import org.apache.hadoop.mapreduce.Job
+
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.encoders.RowEncoder
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
+import org.apache.spark.sql.execution.datasources.{DataSource, FileFormat, OutputWriterFactory, PartitionedFile}
+import org.apache.spark.sql.sources.{DataSourceRegister, Filter}
+import org.apache.spark.sql.types._
+import org.apache.spark.util.SerializableConfiguration
+
+
+private[binaryfile] class BinaryFileFormat extends FileFormat with DataSourceRegister {
+
+  override def inferSchema(
+      sparkSession: SparkSession,
+      options: Map[String, String],
+      files: Seq[FileStatus]): Option[StructType] = Some(BinaryFileDataSource.binaryFileSchema)
+
+  override def prepareWrite(
+      sparkSession: SparkSession,
+      job: Job,
+      options: Map[String, String],
+      dataSchema: StructType): OutputWriterFactory = {
+    throw new UnsupportedOperationException("Write is not supported for binary file data source")
+  }
+
+  override def shortName(): String = "binaryFile"
+
+  override protected def buildReader(
+      sparkSession: SparkSession,
+      dataSchema: StructType,
+      partitionSchema: StructType,
+      requiredSchema: StructType,
+      filters: Seq[Filter],
+      options: Map[String, String],
+      hadoopConf: Configuration): (PartitionedFile) => Iterator[InternalRow] = {
+
+    val broadcastedHadoopConf =
+      sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
+
+    val binaryFileSourceOptions = new BinaryFileSourceOptions(options)
+
+    val pathFilterRegex = binaryFileSourceOptions.pathFilterRegex
+    val globFilter = if (pathFilterRegex.isEmpty) { null } else {
+      new GlobFilter(pathFilterRegex)
+    }
+
+    (file: PartitionedFile) => {
+      val path = file.filePath
+      val fsPath = new Path(path)
+
+      if (globFilter == null || globFilter.accept(fsPath)) {
+        val fs = fsPath.getFileSystem(broadcastedHadoopConf.value.value)
+        val fileStatus = fs.getFileStatus(fsPath)
+        val length = fileStatus.getLen()
+        val modificationTime = new Timestamp(fileStatus.getModificationTime())
+        val stream = fs.open(fsPath)
+        val content = try {
+          ByteStreams.toByteArray(stream)
+        } finally {
+          Closeables.close(stream, true)
+        }
+
+        val converter = RowEncoder(dataSchema)
+        val fullOutput = dataSchema.map { f =>
+          AttributeReference(f.name, f.dataType, f.nullable, f.metadata)()
+        }
+        val requiredOutput = fullOutput.filter { a =>
+          requiredSchema.fieldNames.contains(a.name)
+        }
+
+        val requiredColumns = GenerateUnsafeProjection.generate(requiredOutput, fullOutput)
+
+        val row = Row(Row(path, modificationTime, length), content)
+
+        Iterator(requiredColumns(converter.toRow(row)))
+      } else {
+        Iterator.empty
+      }
+    }
+  }
+}
+
+private[binaryfile] class BinaryFileSourceOptions(
+    @transient private val parameters: CaseInsensitiveMap[String]) extends Serializable {
+
+  def this(parameters: Map[String, String]) = this(CaseInsensitiveMap(parameters))
+
+  /**
+   * only include files with path matching the regex pattern.
+   */
+  val pathFilterRegex = parameters.getOrElse("pathFilterRegex", "").toString
+}

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileSuite.scala

@@ -0,0 +1,82 @@
+/*
+ * 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.binaryfile
+
+import java.sql.Timestamp
+
+import com.google.common.io.{ByteStreams, Closeables}
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.functions.{col, substring_index}
+import org.apache.spark.sql.test.{SharedSQLContext, SQLTestUtils}
+import org.apache.spark.sql.types.LongType
+
+class BinaryFileSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
+  import testImplicits._
+
+  private lazy val filePath = testFile("test-data/text-partitioned")
+
+  private lazy val fsFilePath = new Path(filePath)
+
+  private lazy val fs = fsFilePath.getFileSystem(sparkContext.hadoopConfiguration)
+
+  test("binary file test") {
+
+    val resultDF = spark.read.format("binaryFile")
+      .load(filePath)
+      .select(
+        substring_index(col("status.path"), "/", -1).as("path"),
+        col("status.modification_time"),
+        col("status.length"),
+        col("content"),
+        col("year")
+      )
+
+    val expectedRowSet = new collection.mutable.HashSet[Row]()
+
+    for (partitionDirStatus <- fs.listStatus(fsFilePath)) {
+      val dirPath = partitionDirStatus.getPath
+
+      for (fileStatus <- fs.listStatus(dirPath)) {
+        val fname = fileStatus.getPath.getName
+        val flen = fileStatus.getLen
+        val modificationTime = new Timestamp(fileStatus.getModificationTime)
+
+        val fcontent = {
+          val stream = fs.open(fileStatus.getPath)
+          val content = try {
+            ByteStreams.toByteArray(stream)
+          } finally {
+            Closeables.close(stream, true)
+          }
+          content
+        }
+
+        val partitionName = dirPath.getName.split("=")(1)
+        val year = partitionName.toInt
+        val row = Row(fname, modificationTime, flen, fcontent, year)
+        expectedRowSet.add(row)
+      }
+    }
+
+    val result = resultDF.collect()
+    assert(Set(result: _*) === expectedRowSet.toSet)
+  }
+
+}