Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-23092][SQL] Migrate MemoryStream to DataSourceV2 APIs #20445

Closed
wants to merge 20 commits into from
Closed

[SPARK-23092][SQL] Migrate MemoryStream to DataSourceV2 APIs #20445

Changes from 1 commit
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
7c09b37
save for so far
brkyvz Jan 11, 2018
78c50f8
Save so far
brkyvz Jan 12, 2018
2777b5b
save so far
brkyvz Jan 13, 2018
50a541b
Compiles and I think also runs correctly
brkyvz Jan 16, 2018
fd61724
save
brkyvz Jan 16, 2018
7a0b564
fix merge conflicts
brkyvz Jan 16, 2018
a81c2ec
fix hive
brkyvz Jan 16, 2018
1a4f410
Undo changes to HiveSessionStateBuilder.scala
tdas Jan 27, 2018
083e93c
Merge remote-tracking branch 'apache-github/master' into HEAD
tdas Jan 27, 2018
a817c8d
Fixed the setOffsetRange bug
tdas Jan 29, 2018
35b8854
Fixed DataSourceV2ScanExec canonicalization bug
tdas Jan 30, 2018
e66d809
Fixed metrics reported by MicroBatchExecution
tdas Jan 31, 2018
5adf1fe
Merge remote-tracking branch 'apache-github/master' into SPARK-23092
tdas Jan 31, 2018
478ad17
Reverted changes to DataSourceV2*
tdas Feb 1, 2018
6389d80
Merge remote-tracking branch 'apache-github/master' into SPARK-23092
tdas Feb 1, 2018
3f50f33
Updated package paths to fix compilation
tdas Feb 2, 2018
c713048
Store added data as rows not datasets
tdas Feb 2, 2018
1204755
Fixed ForeachSinkSuite
tdas Feb 3, 2018
f0ce5df
Fixed bug
tdas Feb 7, 2018
c3508e9
Addressed comments
tdas Feb 7, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 5 additions & 11 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ case class MemoryStream[A : Encoder](id: Int, sqlContext: SQLContext)
batches.slice(sliceStart, sliceEnd)
}

logDebug(generateDebugString(newBlocks, startOrdinal, endOrdinal))
logDebug(generateDebugString(newBlocks.flatten, startOrdinal, endOrdinal))

newBlocks.map { block =>
new MemoryStreamDataReaderFactory(block).asInstanceOf[DataReaderFactory[UnsafeRow]]
Expand All @@ -149,18 +149,12 @@ case class MemoryStream[A : Encoder](id: Int, sqlContext: SQLContext)
}

private def generateDebugString(
blocks: Iterable[Array[UnsafeRow]],
blocks: Seq[UnsafeRow],
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: it's probably more "rows" than "blocks" now

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

right! i thought of changing but forgot. my bad.

startOrdinal: Int,
endOrdinal: Int): String = {
val originalUnsupportedCheck =
sqlContext.getConf("spark.sql.streaming.unsupportedOperationCheck")
try {
sqlContext.setConf("spark.sql.streaming.unsupportedOperationCheck", "false")
s"MemoryBatch [$startOrdinal, $endOrdinal]: " +
s"${blocks.flatten.map(row => encoder.fromRow(row)).mkString(", ")}"
} finally {
sqlContext.setConf("spark.sql.streaming.unsupportedOperationCheck", originalUnsupportedCheck)
}
val fromRow = encoder.resolveAndBind().fromRow _
s"MemoryBatch [$startOrdinal, $endOrdinal]: " +
s"${blocks.map(row => fromRow(row)).mkString(", ")}"
}

override def commit(end: OffsetV2): Unit = synchronized {
Expand Down