[SPARK-19633][SS] FileSource read from FileSink

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

@@ -279,28 +279,6 @@ case class DataSource(
     }
   }
 
-  /**
-   * Returns true if there is a single path that has a metadata log indicating which files should
-   * be read.
-   */
-  def hasMetadata(path: Seq[String]): Boolean = {
-    path match {
-      case Seq(singlePath) =>
-        try {
-          val hdfsPath = new Path(singlePath)
-          val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf())
-          val metadataPath = new Path(hdfsPath, FileStreamSink.metadataDir)
-          val res = fs.exists(metadataPath)
-          res
-        } catch {
-          case NonFatal(e) =>
-            logWarning(s"Error while looking for metadata directory.")
-            false
-        }
-      case _ => false
-    }
-  }
-
   /**
    * Create a resolved [[BaseRelation]] that can be used to read data from or write data into this
    * [[DataSource]]
@@ -331,7 +309,9 @@ case class DataSource(
       // We are reading from the results of a streaming query. Load files from the metadata log
       // instead of listing them using HDFS APIs.
       case (format: FileFormat, _)
-          if hasMetadata(caseInsensitiveOptions.get("path").toSeq ++ paths) =>
+          if FileStreamSink.hasMetadata(
+            caseInsensitiveOptions.get("path").toSeq ++ paths,
+            sparkSession.sessionState.newHadoopConf()) =>
         val basePath = new Path((caseInsensitiveOptions.get("path").toSeq ++ paths).head)
         val fileCatalog = new MetadataLogFileIndex(sparkSession, basePath)
         val dataSchema = userSpecifiedSchema.orElse {

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala

@@ -17,6 +17,9 @@
 
 package org.apache.spark.sql.execution.streaming
 
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 
 import org.apache.spark.internal.Logging
@@ -25,9 +28,31 @@ import org.apache.spark.sql.{DataFrame, SparkSession}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.execution.datasources.{FileFormat, FileFormatWriter}
 
-object FileStreamSink {
+object FileStreamSink extends Logging {
   // The name of the subdirectory that is used to store metadata about which files are valid.
   val metadataDir = "_spark_metadata"
+
+  /**
+   * Returns true if there is a single path that has a metadata log indicating which files should
+   * be read.
+   */
+  def hasMetadata(path: Seq[String], hadoopConf: Configuration): Boolean = {
+    path match {
+      case Seq(singlePath) =>
+        try {
+          val hdfsPath = new Path(singlePath)
+          val fs = hdfsPath.getFileSystem(hadoopConf)
+          val metadataPath = new Path(hdfsPath, metadataDir)
+          val res = fs.exists(metadataPath)
+          res
+        } catch {
+          case NonFatal(e) =>
+            logWarning(s"Error while looking for metadata directory.")
+            false
+        }
+      case _ => false
+    }
+  }
 }
 
 /**

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala

@@ -157,13 +157,29 @@ class FileStreamSource(
       checkFilesExist = false)))
   }
 
+  /**
+   * If the source has a metadata log indicating which files should be read, then we should use it.
+   * We figure out whether there exists some metadata log only when user gives a non-glob path.
+   */
+  private val sourceHasMetadata: Boolean =
+    !SparkHadoopUtil.get.isGlobPath(new Path(path)) &&
+      FileStreamSink.hasMetadata(Seq(path), sparkSession.sessionState.newHadoopConf())
+
   /**
    * Returns a list of files found, sorted by their timestamp.
    */
   private def fetchAllFiles(): Seq[(String, Long)] = {
     val startTime = System.nanoTime
-    val globbedPaths = SparkHadoopUtil.get.globPathIfNecessary(qualifiedBasePath)
-    val catalog = new InMemoryFileIndex(sparkSession, globbedPaths, options, Some(new StructType))
+    val catalog =
+      if (sourceHasMetadata) {
+        // Note if `sourceHasMetadata` holds, then `qualifiedBasePath` is guaranteed to be a
+        // non-glob path
+        new MetadataLogFileIndex(sparkSession, qualifiedBasePath)
+      } else {
+        // `qualifiedBasePath` can contains glob patterns
+        val globbedPaths = SparkHadoopUtil.get.globPathIfNecessary(qualifiedBasePath)
+        new InMemoryFileIndex(sparkSession, globbedPaths, options, Some(new StructType))
+      }
     val files = catalog.allFiles().sortBy(_.getModificationTime)(fileSortOrder).map { status =>
       (status.getPath.toUri.toString, status.getModificationTime)
     }

sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala

@@ -662,6 +662,114 @@ class FileStreamSourceSuite extends FileStreamSourceTest {
     }
   }
 
+  test("read data from outputs of another streaming query") {
+    withSQLConf(SQLConf.FILE_SINK_LOG_COMPACT_INTERVAL.key -> "3") {
+      withTempDirs { case (dir, tmp) =>
+        // q1 is a streaming query that reads from memory and writes to text files
+        val q1_source = MemoryStream[String]
+        val q1_checkpointDir = new File(dir, "q1_checkpointDir").getCanonicalPath
+        val q1_outputDir = new File(dir, "q1_outputDir").getCanonicalPath
+        val q1 =
+          q1_source
+            .toDF()
+            .writeStream
+            .option("checkpointLocation", q1_checkpointDir)
+            .format("text")
+            .start(q1_outputDir)
+
+        // q2 is a streaming query that reads q1's text outputs
+        val q2 = createFileStream("text", q1_outputDir).filter($"value" contains "keep")
+
+        def q1AddData(data: String*): StreamAction =
+          Execute { _ =>
+            q1_source.addData(data)
+            q1.processAllAvailable()
+          }
+        def q2ProcessAllAvailable(): StreamAction = Execute { q2 => q2.processAllAvailable() }
+
+        testStream(q2)(
+          // batch 0
+          q1AddData("drop1", "keep2"),
+          q2ProcessAllAvailable(),
+          CheckAnswer("keep2"),
+
+          // batch 1
+          Assert {
+            // create a text file that won't be on q1's sink log
+            // thus even if its contents contains "keep", it should NOT appear in q2's answer
+            val shouldNotKeep = new File(q1_outputDir, "should_not_keep.txt")
+            stringToFile(shouldNotKeep, "should_not_keep!!!")
+            shouldNotKeep.exists()
+          },
+          q1AddData("keep3"),
+          q2ProcessAllAvailable(),
+          CheckAnswer("keep2", "keep3"),
+
+          // batch 2: check that things work well when the sink log gets compacted
+          q1AddData("keep4"),
+          Assert {
+            // compact interval is 3, so file "2.compact" should exist
+            new File(q1_outputDir, s"${FileStreamSink.metadataDir}/2.compact").exists()
+          },
+          q2ProcessAllAvailable(),
+          CheckAnswer("keep2", "keep3", "keep4"),
+
+          // stop q1 manually
+          Execute { _ => q1.stop() }
+        )
+      }
+    }
+  }
+
+  test("read partitioned data from outputs of another streaming query") {
+    withTempDirs { case (dir, tmp) =>
+      // q1 is a streaming query that reads from memory and writes to partitioned json files
+      val q1_source = MemoryStream[(String, String)]
+      val q1_checkpointDir = new File(dir, "q1_checkpointDir").getCanonicalPath
+      val q1_outputDir = new File(dir, "q1_outputDir").getCanonicalPath
+      val q1 =
+        q1_source
+          .toDF()
+          .select($"_1" as "partition", $"_2" as "value")
+          .writeStream
+          .option("checkpointLocation", q1_checkpointDir)
+          .partitionBy("partition")
+          .format("json")
+          .start(q1_outputDir)
+
+      // q2 is a streaming query that reads q1's partitioned json outputs
+      val schema = new StructType().add("value", StringType).add("partition", StringType)
+      val q2 = createFileStream("json", q1_outputDir, Some(schema)).filter($"value" contains "keep")
+
+      def q1AddData(data: (String, String)*): StreamAction =
+        Execute { _ =>
+          q1_source.addData(data)
+          q1.processAllAvailable()
+        }
+      def q2ProcessAllAvailable(): StreamAction = Execute { q2 => q2.processAllAvailable() }
+
+      testStream(q2)(
+        // batch 0: append to a new partition=foo, should read value and partition
+        q1AddData(("foo", "drop1"), ("foo", "keep2")),
+        q2ProcessAllAvailable(),
+        CheckAnswer(("keep2", "foo")),
+
+        // batch 1: append to same partition=foo, should read value and partition
+        q1AddData(("foo", "keep3")),
+        q2ProcessAllAvailable(),
+        CheckAnswer(("keep2", "foo"), ("keep3", "foo")),
+
+        // batch 2: append to a different partition=bar, should read value and partition
+        q1AddData(("bar", "keep4")),
+        q2ProcessAllAvailable(),
+        CheckAnswer(("keep2", "foo"), ("keep3", "foo"), ("keep4", "bar")),
+
+        // stop q1 manually
+        Execute { _ => q1.stop() }
+      )
+    }
+  }
+
   test("when schema inference is turned on, should read partition data") {
     def createFile(content: String, src: File, tmp: File): Unit = {
       val tempFile = Utils.tempFileWith(new File(tmp, "text"))

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala

@@ -208,6 +208,11 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
     }
   }
 
+  /** Execute arbitrary code */
+  case class Execute(val func: StreamExecution => Any) extends StreamAction {
+    override def toString: String = s"Execute(<func>)"
+  }
+
   class StreamManualClock(time: Long = 0L) extends ManualClock(time) with Serializable {
     private var waitStartTime: Option[Long] = None
 
@@ -472,10 +477,16 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
 
           case a: AssertOnQuery =>
             verify(currentStream != null || lastStream != null,
-              "cannot assert when not stream has been started")
+              "cannot assert when no stream has been started")
             val streamToAssert = Option(currentStream).getOrElse(lastStream)
             verify(a.condition(streamToAssert), s"Assert on query failed: ${a.message}")
 
+          case exe: Execute =>
+            verify(currentStream != null || lastStream != null,
+              "cannot execute when no stream has been started")
+            val streamToExecute = Option(currentStream).getOrElse(lastStream)
+            exe.func(streamToExecute)
+
           case a: Assert =>
             val streamToAssert = Option(currentStream).getOrElse(lastStream)
             verify({ a.run(); true }, s"Assert failed: ${a.message}")