[SPARK-14874][SQL][Streaming] Remove the obsolete Batch representation

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

@@ -47,7 +47,7 @@ class FileStreamSink(
   private val fileLog = new FileStreamSinkLog(sqlContext, logPath.toUri.toString)
   private val fs = basePath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
 
-  override def addBatch(batchId: Long, data: DataFrame): Unit = {
+  override def addData(batchId: Long, data: DataFrame): Unit = {
     if (batchId <= fileLog.getLatest().map(_._1).getOrElse(-1L)) {
       logInfo(s"Skipping already committed batch $batchId")
     } else {

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

@@ -88,9 +88,9 @@ class FileStreamSource(
   }
 
   /**
-   * Returns the next batch of data that is available after `start`, if any is available.
+   * Returns the data that is between the offsets (`start`, `end`].
    */
-  override def getBatch(start: Option[Offset], end: Offset): DataFrame = {
+  override def getData(start: Option[Offset], end: Offset): DataFrame = {
     val startId = start.map(_.asInstanceOf[LongOffset].offset).getOrElse(-1L)
     val endId = end.asInstanceOf[LongOffset].offset
 

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

@@ -27,9 +27,9 @@ import org.apache.spark.sql.DataFrame
 trait Sink {
 
   /**
-   * Adds a batch of data to this sink.  The data for a given `batchId` is deterministic and if
+   * Adds a batch of data to this sink. The data for a given `batchId` is deterministic and if
    * this method is called more than once with the same batchId (which will happen in the case of
    * failures), then `data` should only be added once.
    */
-  def addBatch(batchId: Long, data: DataFrame): Unit
+  def addData(batchId: Long, data: DataFrame): Unit
 }

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

@@ -34,9 +34,9 @@ trait Source  {
   def getOffset: Option[Offset]
 
   /**
-   * Returns the data that is is between the offsets (`start`, `end`].  When `start` is `None` then
-   * the batch should begin with the first available record.  This method must always return the
+   * Returns the data that is between the offsets (`start`, `end`]. When `start` is `None` then the
+   * returned data should begin with the first available record. This method must always return the
    * same data for a particular `start` and `end` pair.
    */
-  def getBatch(start: Option[Offset], end: Offset): DataFrame
+  def getData(start: Option[Offset], end: Offset): DataFrame
 }

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

@@ -287,7 +287,7 @@ class StreamExecution(
     val newData = availableOffsets.flatMap {
       case (source, available) if committedOffsets.get(source).map(_ < available).getOrElse(true) =>
         val current = committedOffsets.get(source)
-        val batch = source.getBatch(current, available)
+        val batch = source.getData(current, available)
         logDebug(s"Retrieving data from $source: $current -> $available")
         Some(source -> batch)
       case _ => None
@@ -329,7 +329,7 @@ class StreamExecution(
 
     val nextBatch =
       new Dataset(sqlContext, lastExecution, RowEncoder(lastExecution.analyzed.schema))
-    sink.addBatch(currentBatchId - 1, nextBatch)
+    sink.addData(currentBatchId - 1, nextBatch)
 
     awaitBatchLock.synchronized {
       // Wake up any threads that are waiting for the stream to progress.

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

@@ -91,9 +91,9 @@ case class MemoryStream[A : Encoder](id: Int, sqlContext: SQLContext)
   }
 
   /**
-   * Returns the next batch of data that is available after `start`, if any is available.
+   * Returns the data that is between the offsets (`start`, `end`].
    */
-  override def getBatch(start: Option[Offset], end: Offset): DataFrame = {
+  override def getData(start: Option[Offset], end: Offset): DataFrame = {
     val startOrdinal =
       start.map(_.asInstanceOf[LongOffset]).getOrElse(LongOffset(-1)).offset.toInt + 1
     val endOrdinal = end.asInstanceOf[LongOffset].offset.toInt + 1
@@ -135,7 +135,7 @@ class MemorySink(val schema: StructType) extends Sink with Logging {
     }.mkString("\n")
   }
 
-  override def addBatch(batchId: Long, data: DataFrame): Unit = synchronized {
+  override def addData(batchId: Long, data: DataFrame): Unit = synchronized {
     if (batchId == batches.size) {
       logDebug(s"Committing batch $batchId")
       batches.append(data.collect())

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

@@ -79,7 +79,7 @@ class DefaultSource extends StreamSourceProvider with StreamSinkProvider {
 
       override def getOffset: Option[Offset] = Some(new LongOffset(0))
 
-      override def getBatch(start: Option[Offset], end: Offset): DataFrame = {
+      override def getData(start: Option[Offset], end: Offset): DataFrame = {
         import sqlContext.implicits._
 
         Seq[Int]().toDS().toDF()
@@ -95,7 +95,7 @@ class DefaultSource extends StreamSourceProvider with StreamSinkProvider {
     LastOptions.partitionColumns = partitionColumns
     LastOptions.mockStreamSinkProvider.createSink(sqlContext, parameters, partitionColumns)
     new Sink {
-      override def addBatch(batchId: Long, data: DataFrame): Unit = {}
+      override def addData(batchId: Long, data: DataFrame): Unit = {}
     }
   }
 }

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

@@ -172,7 +172,7 @@ class FakeDefaultSource extends StreamSourceProvider {
         }
       }
 
-      override def getBatch(start: Option[Offset], end: Offset): DataFrame = {
+      override def getData(start: Option[Offset], end: Offset): DataFrame = {
         val startOffset = start.map(_.asInstanceOf[LongOffset].offset).getOrElse(-1L) + 1
         sqlContext.range(startOffset, end.asInstanceOf[LongOffset].offset + 1).toDF("a")
       }