You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/06/23 04:59:05 UTC
[GitHub] [spark] HeartSaVioR opened a new pull request, #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
HeartSaVioR opened a new pull request, #36963:
URL: https://github.com/apache/spark/pull/36963
### What changes were proposed in this pull request?
This PR proposes to expose the information of catalog table (V1/V2) to the logical plan in streaming query, specifically, parsed plan and analyzed plan. (We may discard some information in optimized plan.)
The major change is to propagate the information of catalog table from the place we resolve the table to the place we execute the query. In MicroBatch execution, we have several transformations on the logical plan which replace the node with another node, hence this PR touches multiple logical nodes which the code path passes through.
Specifically for DSv1 sink, we don't have a specific write logical node, hence it's not feasible to expose the information for the destination. This PR introduces `WriteToMicroBatchDataSourceV1` which is DSv1 version of `WriteToMicroBatchDataSource` as a logical node for DSv1 sink. Worth noting that `WriteToMicroBatchDataSourceV1` plays as a marker - we eliminate this node in streaming specific optimization phase.
### Why are the changes needed?
This PR give a better UX to end users who use table API for streaming query. Previously it's not easy or even not feasible to check which tables are being read and written from the streaming query. Most likely it requires end users to look into their code/query.
### Does this PR introduce _any_ user-facing change?
Yes, in parsed/analyzed plan, we now expose the table information into the read/write logical node. Specifically for DSv1, we introduce a marker write node to expose the information for destination without majorly changing existing logic.
> DSv1 read and write
>> Before the patch
<img width="1635" alt="original-read-from-dsv1-write-to-dsv1" src="https://user-images.githubusercontent.com/1317309/175210731-dcc4cc4d-a70b-467d-b577-79c20600db32.png">
>> After the patch
<img width="1716" alt="proposal-read-from-dsv1-write-to-dsv1" src="https://user-images.githubusercontent.com/1317309/175210753-8f3c0b81-6ec5-45df-a2f8-37589960bba2.png">
> DSv2 read and write
>> Before the patch
<img width="1684" alt="original-read-from-dsv2-write-to-dsv2" src="https://user-images.githubusercontent.com/1317309/175210780-4a99c670-8a42-4511-959c-cafe0c24bc00.png">
>> After the patch
<img width="1755" alt="proposal-read-from-dsv2-write-to-dsv2" src="https://user-images.githubusercontent.com/1317309/175210807-2b0b6935-e696-4051-b1a2-725d784d9d5f.png">
### How was this patch tested?
New test cases. Also manually tested via running following query and checked the UI page:
> DSv1 read and write
```
/*
./bin/spark-shell --conf "spark.sql.ui.explainMode=extended"
*/
spark.sql("drop table if exists stream_source")
spark.sql("drop table if exists stream_target")
spark.sql("create table stream_source (col1 string, col2 int) using parquet")
spark.sql("create table stream_target (col1 string, col2 int) using parquet")
val checkpointDir = java.nio.file.Files.createTempDirectory("checkpoint-")
val q = spark.readStream.table("stream_source").writeStream.format("parquet").option("checkpointLocation", checkpointDir.toString).toTable("stream_target")
Thread.sleep(10000)
spark.sql("insert into stream_source values ('a', 1)")
spark.sql("insert into stream_source values ('a', 2)")
spark.sql("insert into stream_source values ('a', 3)")
q.processAllAvailable()
spark.sql("insert into stream_source values ('b', 1)")
spark.sql("insert into stream_source values ('b', 2)")
spark.sql("insert into stream_source values ('b', 3)")
q.processAllAvailable()
spark.sql("insert into stream_source values ('c', 1)")
spark.sql("insert into stream_source values ('c', 2)")
spark.sql("insert into stream_source values ('c', 3)")
q.processAllAvailable()
q.stop()
```
> DSv2 read and write
```
/*
./bin/spark-shell --packages org.apache.iceberg:iceberg-spark-runtime-3.2_2.12:0.13.1\
--conf spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.local.type=hadoop \
--conf spark.sql.catalog.local.warehouse=$PWD/warehouse \
--conf spark.sql.ui.explainMode=extended
*/
spark.sql("drop table if exists local.db.stream_target")
spark.sql("create table local.db.stream_source (col1 string, col2 int) using iceberg")
spark.sql("create table local.db.stream_target (col1 string, col2 int) using iceberg")
val checkpointDir = java.nio.file.Files.createTempDirectory("checkpoint-")
val q = spark.readStream.table("local.db.stream_source").writeStream.format("iceberg").option("checkpointLocation", checkpointDir.toString).toTable("local.db.stream_target")
Thread.sleep(10000)
spark.sql("insert into local.db.stream_source values ('a', 1)")
spark.sql("insert into local.db.stream_source values ('a', 2)")
spark.sql("insert into local.db.stream_source values ('a', 3)")
q.processAllAvailable()
spark.sql("insert into local.db.stream_source values ('b', 1)")
spark.sql("insert into local.db.stream_source values ('b', 2)")
spark.sql("insert into local.db.stream_source values ('b', 3)")
q.processAllAvailable()
spark.sql("insert into local.db.stream_source values ('c', 1)")
spark.sql("insert into local.db.stream_source values ('c', 2)")
spark.sql("insert into local.db.stream_source values ('c', 3)")
q.processAllAvailable()
q.stop()
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1167874348
Let me merge this as the only failure is not relevant to this PR.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1164155040
@viirya Great point. I just marked the outstanding difference on the screenshot of the proposal (both DSv1 and DSv2). Thanks!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1167799733
@jerrypeng Please see the screenshot of "after the change". I described where they come and why they are not something we can rely on.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r907049142
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala:
##########
@@ -877,13 +879,18 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product with Tre
t.copy(properties = Utils.redact(t.properties).toMap,
options = Utils.redact(t.options).toMap) :: Nil
case table: CatalogTable =>
- table.storage.serde match {
- case Some(serde) => table.identifier :: serde :: Nil
- case _ => table.identifier :: Nil
- }
+ stringArgsForCatalogTable(table)
+
case other => other :: Nil
}.mkString(", ")
+ private def stringArgsForCatalogTable(table: CatalogTable): Seq[Any] = {
+ table.storage.serde match {
+ case Some(serde) => table.identifier :: serde :: Nil
Review Comment:
I think quoted string is better, let's keep it as it was first.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1165027544
> That's said previously it also shows some table info, but it mostly depends on data source implementation. Is it correct?
Correct. The information it prints out is mostly just a collection of toString in fields in case class. Since we don't have any contract with how the information data source provides reflects to the logical plan, it's totally up to the data source implementation.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904747969
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -650,11 +650,17 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
object StreamingRelationStrategy extends Strategy {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case s: StreamingRelation =>
- StreamingRelationExec(s.sourceName, s.output) :: Nil
+ val tableQualifier = s.dataSource.catalogTable.map(_.identifier.unquotedString)
Review Comment:
Yeah the name is much clearer. Thanks!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1166688561
@cloud-fan Friendly reminder.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r907053973
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/WriteToMicroBatchDataSourceV1.scala:
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.streaming.sources
+
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
+import org.apache.spark.sql.execution.streaming.Sink
+import org.apache.spark.sql.streaming.OutputMode
+
+/**
+ * Marker node to represent a DSv1 sink on streaming query.
+ *
+ * Despite this is expected to be the top node, this node should behave like "pass-through"
+ * since the DSv1 codepath on microbatch execution handles sink operation separately.
+ *
+ * This node is eliminated in streaming specific optimization phase, which means there is no
+ * matching physical node.
+ */
+case class WriteToMicroBatchDataSourceV1(
+ catalogTable: Option[CatalogTable],
+ sink: Sink,
+ query: LogicalPlan,
+ queryId: String,
+ writeOptions: Map[String, String],
+ outputMode: OutputMode,
+ batchId: Option[Long] = None)
Review Comment:
where do we use this `batchId`?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904804693
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -576,14 +589,22 @@ class MicroBatchExecution(
// Replace sources in the logical plan with data that has arrived since the last batch.
val newBatchesPlan = logicalPlan transform {
// For v1 sources.
- case StreamingExecutionRelation(source, output) =>
+ case StreamingExecutionRelation(source, output, catalogTable) =>
newData.get(source).map { dataPlan =>
val hasFileMetadata = output.exists {
case FileSourceMetadataAttribute(_) => true
case _ => false
}
val finalDataPlan = dataPlan transformUp {
- case l: LogicalRelation if hasFileMetadata => l.withMetadataColumns()
+ case l: LogicalRelation =>
+ var newRelation = l
+ if (hasFileMetadata) {
+ newRelation = newRelation.withMetadataColumns()
+ }
+ catalogTable.foreach { table =>
+ newRelation = newRelation.copy(catalogTable = Some(table))
Review Comment:
yea we should have an assertion. Blindly overwriting the catalogTable in `LogicalRelation` looks a bit risky.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r907059119
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala:
##########
@@ -877,13 +879,18 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product with Tre
t.copy(properties = Utils.redact(t.properties).toMap,
options = Utils.redact(t.options).toMap) :: Nil
case table: CatalogTable =>
- table.storage.serde match {
- case Some(serde) => table.identifier :: serde :: Nil
- case _ => table.identifier :: Nil
- }
+ stringArgsForCatalogTable(table)
+
case other => other :: Nil
}.mkString(", ")
+ private def stringArgsForCatalogTable(table: CatalogTable): Seq[Any] = {
+ table.storage.serde match {
+ case Some(serde) => table.identifier :: serde :: Nil
Review Comment:
OK. Let me leave this as it is, and see whether I can make it consistent for other places in the following work (out of this PR).
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
viirya commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1165235522
Please let @cloud-fan take another look.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1165237196
Sure, thanks for reviewing!
Btw, I missed to mention that the screenshot is taken with the change of configuration (you know, we only show the physical plan by default). I've just added the fact to the PR description.
> Worth noting that the screenshot is taken with the config "spark.sql.ui.explainMode=extended". By default, we only show physical plan as formatted one, which hides the improvement being done here. Still, end users can run query.explain(extended=true) to print out plan"s" which contains parsed/analyzed plans.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904649705
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala:
##########
@@ -868,6 +868,8 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product with Tre
case null => Nil
case None => Nil
case Some(null) => Nil
+ case Some(table: CatalogTable) =>
Review Comment:
It's already the case for batch queries and I'd suggest to do it in a separated PR if we want to
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904529825
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -113,7 +114,9 @@ class MicroBatchExecution(
v1.get.asInstanceOf[StreamingRelation].dataSource.createSource(metadataPath)
nextSourceId += 1
logInfo(s"Using Source [$source] from DataSourceV2 named '$srcName' $dsStr")
- StreamingExecutionRelation(source, output)(sparkSession)
+ // We don't have a catalog table but may have a table identifier. Given this is about
Review Comment:
Actually this else statement is to handle edge case where the data source is based on DSv2 and implements continuous read but does not implement microbatch read. Really a rare case.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1167878489
Thanks all for reviewing! I merged this to master.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904747378
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -576,14 +589,22 @@ class MicroBatchExecution(
// Replace sources in the logical plan with data that has arrived since the last batch.
val newBatchesPlan = logicalPlan transform {
// For v1 sources.
- case StreamingExecutionRelation(source, output) =>
+ case StreamingExecutionRelation(source, output, catalogTable) =>
newData.get(source).map { dataPlan =>
val hasFileMetadata = output.exists {
case FileSourceMetadataAttribute(_) => true
case _ => false
}
val finalDataPlan = dataPlan transformUp {
- case l: LogicalRelation if hasFileMetadata => l.withMetadataColumns()
+ case l: LogicalRelation =>
+ var newRelation = l
+ if (hasFileMetadata) {
+ newRelation = newRelation.withMetadataColumns()
+ }
+ catalogTable.foreach { table =>
+ newRelation = newRelation.copy(catalogTable = Some(table))
Review Comment:
Just to clarify, did you mean having an assertion, or don't overwrite catalogTable if data source produces one in the relation? I can't imagine the codepath latter becomes true, but just wanted to clarify so that I can reflect the review properly.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -576,14 +589,22 @@ class MicroBatchExecution(
// Replace sources in the logical plan with data that has arrived since the last batch.
val newBatchesPlan = logicalPlan transform {
// For v1 sources.
- case StreamingExecutionRelation(source, output) =>
+ case StreamingExecutionRelation(source, output, catalogTable) =>
newData.get(source).map { dataPlan =>
val hasFileMetadata = output.exists {
case FileSourceMetadataAttribute(_) => true
case _ => false
}
val finalDataPlan = dataPlan transformUp {
- case l: LogicalRelation if hasFileMetadata => l.withMetadataColumns()
+ case l: LogicalRelation =>
+ var newRelation = l
+ if (hasFileMetadata) {
+ newRelation = newRelation.withMetadataColumns()
+ }
+ catalogTable.foreach { table =>
+ newRelation = newRelation.copy(catalogTable = Some(table))
Review Comment:
Just to clarify, did you mean having an assertion, or don't overwrite catalogTable if data source produces one in the relation? I can't imagine the codepath latter can be true, but just wanted to clarify so that I can reflect the review properly.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] feng-tao commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
feng-tao commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1165215365
exciting!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r907049815
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -68,7 +68,11 @@ case class DataSourceV2Relation(
override def skipSchemaResolution: Boolean = table.supports(TableCapability.ACCEPT_ANY_SCHEMA)
override def simpleString(maxFields: Int): String = {
- s"RelationV2${truncatedString(output, "[", ", ", "]", maxFields)} $name"
+ val tableQualifier = (catalog, identifier) match {
Review Comment:
nit: qualifiedTableName
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r907061291
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/WriteToMicroBatchDataSourceV1.scala:
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.streaming.sources
+
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
+import org.apache.spark.sql.execution.streaming.Sink
+import org.apache.spark.sql.streaming.OutputMode
+
+/**
+ * Marker node to represent a DSv1 sink on streaming query.
+ *
+ * Despite this is expected to be the top node, this node should behave like "pass-through"
+ * since the DSv1 codepath on microbatch execution handles sink operation separately.
+ *
+ * This node is eliminated in streaming specific optimization phase, which means there is no
+ * matching physical node.
+ */
+case class WriteToMicroBatchDataSourceV1(
+ catalogTable: Option[CatalogTable],
+ sink: Sink,
+ query: LogicalPlan,
+ queryId: String,
+ writeOptions: Map[String, String],
+ outputMode: OutputMode,
+ batchId: Option[Long] = None)
Review Comment:
This is to make this class be symmetric with WriteToMicroBatchDataSource. Many of parameters are not actually used.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904654762
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala:
##########
@@ -868,6 +868,8 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product with Tre
case null => Nil
case None => Nil
case Some(null) => Nil
+ case Some(table: CatalogTable) =>
Review Comment:
Ah OK. Didn't realize logical nodes for batch also have such case. I agree it's better to deal with separate PR.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904650153
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -650,11 +650,17 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
object StreamingRelationStrategy extends Strategy {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case s: StreamingRelation =>
- StreamingRelationExec(s.sourceName, s.output) :: Nil
+ val tableQualifier = s.dataSource.catalogTable.map(_.identifier.unquotedString)
Review Comment:
do you mean `qualifiedTablrName`?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1167689731
@HeartSaVioR in regard to the UI screen shots you provided for DSv2 read and write, the logical plan before the patch already contained the src and target catalog tables. What is the difference to the logical after the this patch is applied for DSv2 connectors?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1167873442
I see only one test failure from ParquetIOSuite, and I figured out it is failing from master branch as well.
I just filed https://issues.apache.org/jira/browse/SPARK-39622, and will notice this to dev@ mailing list.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
viirya commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1164737634
> @viirya Great point. I just marked the outstanding difference on the screenshot of the proposal (both DSv1 and DSv2). Thanks!
Thanks! So as I look the difference of DSv2, what this does is more to guarantee table info in the logical plan? That's said previously it also shows some table info, but it mostly depends on data source implementation. Is it correct?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r905655365
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala:
##########
@@ -868,6 +868,8 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product with Tre
case null => Nil
case None => Nil
case Some(null) => Nil
+ case Some(table: CatalogTable) =>
Review Comment:
This triggers test failures on SQLQueryTestSuite / ThriftServerQueryTestSuite due to the change of representation of CatalogTable wrapped with Some. While I think it "was" a bug or at least not intended (because we care for CatalogTable not wrapped with Some), please let me know if we want to retain the current.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1167101071
Just rebased to pick up the fixes on GA. No changes during rebase.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904652126
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -576,14 +589,22 @@ class MicroBatchExecution(
// Replace sources in the logical plan with data that has arrived since the last batch.
val newBatchesPlan = logicalPlan transform {
// For v1 sources.
- case StreamingExecutionRelation(source, output) =>
+ case StreamingExecutionRelation(source, output, catalogTable) =>
newData.get(source).map { dataPlan =>
val hasFileMetadata = output.exists {
case FileSourceMetadataAttribute(_) => true
case _ => false
}
val finalDataPlan = dataPlan transformUp {
- case l: LogicalRelation if hasFileMetadata => l.withMetadataColumns()
+ case l: LogicalRelation =>
+ var newRelation = l
+ if (hasFileMetadata) {
+ newRelation = newRelation.withMetadataColumns()
+ }
+ catalogTable.foreach { table =>
+ newRelation = newRelation.copy(catalogTable = Some(table))
Review Comment:
shall we make sure `newRelation.catalogTable` is empty?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904526435
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/IncrementalExecution.scala:
##########
@@ -77,7 +78,11 @@ class IncrementalExecution(
*/
override
lazy val optimizedPlan: LogicalPlan = executePhase(QueryPlanningTracker.OPTIMIZATION) {
- sparkSession.sessionState.optimizer.executeAndTrack(withCachedData,
+ // Performing pre-optimization for streaming specific
+ val preOptimized = withCachedData.transform {
Review Comment:
Self-comment: additional comment to explain what it does.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala:
##########
@@ -877,13 +879,18 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product with Tre
t.copy(properties = Utils.redact(t.properties).toMap,
options = Utils.redact(t.options).toMap) :: Nil
case table: CatalogTable =>
- table.storage.serde match {
- case Some(serde) => table.identifier :: serde :: Nil
- case _ => table.identifier :: Nil
- }
+ stringArgsForCatalogTable(table)
+
case other => other :: Nil
}.mkString(", ")
+ private def stringArgsForCatalogTable(table: CatalogTable): Seq[Any] = {
+ table.storage.serde match {
+ case Some(serde) => table.identifier :: serde :: Nil
Review Comment:
I see inconsistence on representing table identifier on the plan - while this produces "quoted" string, I also see existing codebase to produce "unquoted" string. e.g. LogicalRelation.
I'd like to hear the voice which is our preference. For now I used "unquoted" string for other places and leave this as it is (hence "quoted" string).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -167,6 +173,17 @@ case class StreamingDataSourceV2Relation(
case _ =>
Statistics(sizeInBytes = conf.defaultSizeInBytes)
}
+
+ private val stringArgsVal: Seq[Any] = {
+ val tableQualifier = (catalog, identifier) match {
Review Comment:
Printing out `catalog` and `identifier` separately does not seem to give a good output, hence I changed to produce a single output for composite fields `catalog` and `identifier` whenever they are provided altogether in the logical node (only on streaming code path).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala:
##########
@@ -868,6 +868,8 @@ abstract class TreeNode[BaseType <: TreeNode[BaseType]] extends Product with Tre
case null => Nil
case None => Nil
case Some(null) => Nil
+ case Some(table: CatalogTable) =>
Review Comment:
This is needed to represent the CatalogTable properly if it is presented. (Before the fix it calls CatalogTable.toString which produces a bunch of information with multiple lines.)
I feel it could be something be generalized, but I don't know we have other existing cases to handle.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -576,14 +589,22 @@ class MicroBatchExecution(
// Replace sources in the logical plan with data that has arrived since the last batch.
val newBatchesPlan = logicalPlan transform {
// For v1 sources.
- case StreamingExecutionRelation(source, output) =>
+ case StreamingExecutionRelation(source, output, catalogTable) =>
newData.get(source).map { dataPlan =>
val hasFileMetadata = output.exists {
case FileSourceMetadataAttribute(_) => true
case _ => false
}
val finalDataPlan = dataPlan transformUp {
- case l: LogicalRelation if hasFileMetadata => l.withMetadataColumns()
+ case l: LogicalRelation =>
Review Comment:
Many DSv1 data sources produce LogicalRelation as leaf node, which contains Relation having a field of catalogTable. We fill out the information of the catalogTable in case when the source is not able to provide the information.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingRelation.scala:
##########
@@ -42,7 +48,11 @@ object StreamingRelation {
* It should be used to create [[Source]] and converted to [[StreamingExecutionRelation]] when
* passing to [[StreamExecution]] to run a query.
*/
-case class StreamingRelation(dataSource: DataSource, sourceName: String, output: Seq[Attribute])
+case class StreamingRelation(
+ dataSource: DataSource,
+ sourceName: String,
+ output: Seq[Attribute],
+ catalogTable: Option[CatalogTable])
Review Comment:
Self-comment: DataSource has a field `catalogTable`. Leverage it instead of adding a new column.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/WriteToMicroBatchDataSourceV1.scala:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.streaming.sources
+
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
+import org.apache.spark.sql.execution.streaming.Sink
+import org.apache.spark.sql.streaming.OutputMode
+
+case class WriteToMicroBatchDataSourceV1(
+ catalogTable: Option[CatalogTable],
+ sink: Sink,
+ query: LogicalPlan,
+ queryId: String,
+ writeOptions: Map[String, String],
+ outputMode: OutputMode,
+ batchId: Option[Long] = None)
+ extends UnaryNode {
+
+ override def child: LogicalPlan = query
+
+ // Despite this is logically the top node, this node should behave like "pass-through"
+ // since the DSv1 codepath on microbatch execution handles sink operation separately.
+ // We will eliminate this node in physical planning, which shouldn't make difference as
Review Comment:
Self-review: Outdated. Now we eliminate this in optimization. Update the comment.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -113,7 +114,9 @@ class MicroBatchExecution(
v1.get.asInstanceOf[StreamingRelation].dataSource.createSource(metadataPath)
nextSourceId += 1
logInfo(s"Using Source [$source] from DataSourceV2 named '$srcName' $dsStr")
- StreamingExecutionRelation(source, output)(sparkSession)
+ // We don't have a catalog table but may have a table identifier. Given this is about
Review Comment:
Actually this else statement is to handle edge case where the data source is based on DSv2 and implements continuous read but does not implement microbatch read.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -168,7 +171,17 @@ class MicroBatchExecution(
extraOptions,
outputMode)
- case _ => _logicalPlan
+ case s: Sink =>
+ WriteToMicroBatchDataSourceV1(
Review Comment:
It'd be nice if we can deal with SPARK-27484, but let's defer it as of now as it may bring additional works/concerns.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -113,7 +114,9 @@ class MicroBatchExecution(
v1.get.asInstanceOf[StreamingRelation].dataSource.createSource(metadataPath)
nextSourceId += 1
logInfo(s"Using Source [$source] from DataSourceV2 named '$srcName' $dsStr")
- StreamingExecutionRelation(source, output)(sparkSession)
+ // We don't have a catalog table but may have a table identifier. Given this is about
Review Comment:
This is based on the fact we are encouraging 3rd party to use DSv2 for streaming data source whenever possible. (We don't make any improvements on DSv1 now.) When the data source implements DSv2, it is advised to implement the functionality of microbatch read.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingRelation.scala:
##########
@@ -111,7 +122,11 @@ case class StreamingExecutionRelation(
* A dummy physical plan for [[StreamingRelation]] to support
* [[org.apache.spark.sql.Dataset.explain]]
*/
-case class StreamingRelationExec(sourceName: String, output: Seq[Attribute]) extends LeafExecNode {
+case class StreamingRelationExec(
+ sourceName: String,
+ output: Seq[Attribute],
+ tableIdentifier: Option[String]) extends LeafExecNode {
+ // FIXME: check the representation of this node and come up with good format to show table name
Review Comment:
Self-review: remove it as it's already addressed.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR closed pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
URL: https://github.com/apache/spark/pull/36963
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r907066953
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -68,7 +68,11 @@ case class DataSourceV2Relation(
override def skipSchemaResolution: Boolean = table.supports(TableCapability.ACCEPT_ANY_SCHEMA)
override def simpleString(maxFields: Int): String = {
- s"RelationV2${truncatedString(output, "[", ", ", "]", maxFields)} $name"
+ val tableQualifier = (catalog, identifier) match {
Review Comment:
[4ae4b81](https://github.com/apache/spark/pull/36963/commits/4ae4b813e295fda8f6c0dab4ffee43fef0761496)
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r907063933
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/WriteToMicroBatchDataSourceV1.scala:
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.streaming.sources
+
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
+import org.apache.spark.sql.execution.streaming.Sink
+import org.apache.spark.sql.streaming.OutputMode
+
+/**
+ * Marker node to represent a DSv1 sink on streaming query.
+ *
+ * Despite this is expected to be the top node, this node should behave like "pass-through"
+ * since the DSv1 codepath on microbatch execution handles sink operation separately.
+ *
+ * This node is eliminated in streaming specific optimization phase, which means there is no
+ * matching physical node.
+ */
+case class WriteToMicroBatchDataSourceV1(
+ catalogTable: Option[CatalogTable],
+ sink: Sink,
+ query: LogicalPlan,
+ queryId: String,
+ writeOptions: Map[String, String],
+ outputMode: OutputMode,
+ batchId: Option[Long] = None)
Review Comment:
Previous self-comment:
> It'd be nice if we can deal with [SPARK-27484](https://issues.apache.org/jira/browse/SPARK-27484), but let's defer it as of now as it may bring additional works/concerns.
These parameters could probably help us to go with SPARK-27484.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on PR #36963:
URL: https://github.com/apache/spark/pull/36963#issuecomment-1163995769
cc. @cloud-fan since it touches several SQL codepath.
cc. @viirya @xuanyuanking @zsxwing for reviewing the SS area.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a diff in pull request #36963: [SPARK-39564][SS] Expose the information of catalog table to the logical plan in streaming query
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #36963:
URL: https://github.com/apache/spark/pull/36963#discussion_r904533116
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -576,14 +589,22 @@ class MicroBatchExecution(
// Replace sources in the logical plan with data that has arrived since the last batch.
val newBatchesPlan = logicalPlan transform {
// For v1 sources.
- case StreamingExecutionRelation(source, output) =>
+ case StreamingExecutionRelation(source, output, catalogTable) =>
newData.get(source).map { dataPlan =>
val hasFileMetadata = output.exists {
case FileSourceMetadataAttribute(_) => true
case _ => false
}
val finalDataPlan = dataPlan transformUp {
- case l: LogicalRelation if hasFileMetadata => l.withMetadataColumns()
+ case l: LogicalRelation =>
Review Comment:
Many DSv1 data sources produce LogicalRelation as leaf node when calling `source.getBatch()`, which contains Relation having a field of catalogTable. We fill out the information of the catalogTable in case when the source is not able to provide the information.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org