HeartSaVioR commented on a change in pull request #29767:
URL: https://github.com/apache/spark/pull/29767#discussion_r489137859
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala
##########
@@ -300,97 +301,108 @@ final class DataStreamWriter[T] private[sql](ds:
Dataset[T]) {
"write files of Hive data source directly.")
}
- if (source == "memory") {
- assertNotPartitioned("memory")
- if (extraOptions.get("queryName").isEmpty) {
- throw new AnalysisException("queryName must be specified for memory
sink")
- }
- val sink = new MemorySink()
- val resultDf = Dataset.ofRows(df.sparkSession, new MemoryPlan(sink,
df.schema.toAttributes))
- val chkpointLoc = extraOptions.get("checkpointLocation")
- val recoverFromChkpoint = outputMode == OutputMode.Complete()
- val query =
df.sparkSession.sessionState.streamingQueryManager.startQuery(
- extraOptions.get("queryName"),
- chkpointLoc,
- df,
- extraOptions.toMap,
- sink,
- outputMode,
- useTempCheckpointLocation = true,
- recoverFromCheckpointLocation = recoverFromChkpoint,
- trigger = trigger)
- resultDf.createOrReplaceTempView(query.name)
- query
- } else if (source == "foreach") {
- assertNotPartitioned("foreach")
- val sink = ForeachWriterTable[T](foreachWriter, ds.exprEnc)
- df.sparkSession.sessionState.streamingQueryManager.startQuery(
- extraOptions.get("queryName"),
- extraOptions.get("checkpointLocation"),
- df,
- extraOptions.toMap,
- sink,
- outputMode,
- useTempCheckpointLocation = true,
- trigger = trigger)
- } else if (source == "foreachBatch") {
- assertNotPartitioned("foreachBatch")
- if (trigger.isInstanceOf[ContinuousTrigger]) {
- throw new AnalysisException("'foreachBatch' is not supported with
continuous trigger")
- }
- val sink = new ForeachBatchSink[T](foreachBatchWriter, ds.exprEnc)
- df.sparkSession.sessionState.streamingQueryManager.startQuery(
- extraOptions.get("queryName"),
- extraOptions.get("checkpointLocation"),
- df,
- extraOptions.toMap,
- sink,
- outputMode,
- useTempCheckpointLocation = true,
- trigger = trigger)
- } else {
- val cls = DataSource.lookupDataSource(source,
df.sparkSession.sessionState.conf)
- val disabledSources =
df.sparkSession.sqlContext.conf.disabledV2StreamingWriters.split(",")
- val useV1Source = disabledSources.contains(cls.getCanonicalName) ||
- // file source v2 does not support streaming yet.
- classOf[FileDataSourceV2].isAssignableFrom(cls)
-
- val optionsWithPath = if (path.isEmpty) {
- extraOptions
- } else {
- extraOptions + ("path" -> path.get)
- }
+ val queryName = extraOptions.get("queryName")
Review comment:
The huge diff comes from refactor - I have to go with refactoring as the
statement for StreamingQueryManager.startQuery() are all duplicated and I was
about to add one more duplication.
The actual change is only performed for `source == SOURCE_NAME_TABLE`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
