You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "jerrypeng (via GitHub)" <gi...@apache.org> on 2025/10/15 06:45:07 UTC
[PR] [WIP][SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
jerrypeng opened a new pull request, #52620:
URL: https://github.com/apache/spark/pull/52620
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on PR #52620:
URL: https://github.com/apache/spark/pull/52620#issuecomment-3457585443
@viirya thank you for merging!
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya closed pull request #52620: [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches
URL: https://github.com/apache/spark/pull/52620
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2459019242
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -263,6 +290,12 @@ class MicroBatchExecution(
case s => s -> ReadLimit.allAvailable()
}.toMap
}
+ if (trigger.isInstanceOf[RealTimeTrigger] && uniqueSources.size != sources.size) {
+ throw new SparkIllegalStateException(
+ errorClass = s"STREAMING_REAL_TIME_MODE.IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED",
Review Comment:
Why this is specially for Union? Looks like it doesn't check if there is a Union.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryPlanTraverseHelper.scala:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.runtime
+
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
+
+/**
+ * This is a utility object placing methods to traverse the query plan for streaming query.
+ * This is used for patterns of traversal which are repeated in multiple places.
+ */
+object StreamingQueryPlanTraverseHelper {
+ def collectFromUnfoldedLatestExecutedPlan[B](
+ lastExecution: IncrementalExecution)(pf: PartialFunction[SparkPlan, B]): Seq[B] = {
+ val executedPlan = lastExecution.executedPlan
+
+ collectWithUnfolding(executedPlan)(pf)
+ }
+
+ private def collectWithUnfolding[B](executedPlan: SparkPlan)(
+ pf: PartialFunction[SparkPlan, B]): Seq[B] = {
+ executedPlan.flatMap {
+ // InMemoryTableScanExec is a node to represent a cached plan. The node has underlying
+ // actual executed plan, which we should traverse to collect the required information.
+ case s: InMemoryTableScanExec => collectWithUnfolding(s.relation.cachedPlan)(pf)
Review Comment:
Hm? Do you mean `RealTimeStreamScanExec` in a cache?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryPlanTraverseHelper.scala:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.runtime
+
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
+
+/**
+ * This is a utility object placing methods to traverse the query plan for streaming query.
+ * This is used for patterns of traversal which are repeated in multiple places.
+ */
+object StreamingQueryPlanTraverseHelper {
Review Comment:
Should we make it private to sql?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RealTimeStreamScanExec.scala:
##########
@@ -37,3 +49,121 @@ object LowLatencyClock {
clock = inputClock
}
}
+
+/**
+ * A wrap reader that turns a Partition Reader extending SupportsRealTimeRead to a
+ * normal PartitionReader and follow the task termination time `lowLatencyEndTime`, and
+ * report end offsets in the end to `endOffsets`.
+ */
+case class LowLatencyReaderWrap(
+ reader: SupportsRealTimeRead[InternalRow],
+ lowLatencyEndTime: Long,
+ endOffsets: CollectionAccumulator[PartitionOffsetWithIndex])
+ extends PartitionReader[InternalRow] {
+
+ override def next(): Boolean = {
+ val curTime = LowLatencyClock.getTimeMillis()
+ val ret = if (curTime >= lowLatencyEndTime) {
+ RecordStatus.newStatusWithoutArrivalTime(false)
+ } else {
+ reader.nextWithTimeout(lowLatencyEndTime - curTime)
+ }
+
+ if (!ret.hasRecord) {
+ // The way of using TaskContext.get().partitionId() to map to a partition
+ // may be fragile as it relies on thread locals.
+ endOffsets.add(
+ new PartitionOffsetWithIndex(TaskContext.get().partitionId(), reader.getOffset)
+ )
+ }
+ ret.hasRecord
+ }
+
+ override def get(): InternalRow = {
+ reader.get()
+ }
+
+ override def close(): Unit = {}
+}
+
+/**
+ * Wrapper factory that creates LowLatencyReaderWrap from reader as SupportsRealTimeRead
+ */
+case class LowLatencyReaderFactoryWrap(
+ partitionReaderFactory: PartitionReaderFactory,
+ lowLatencyEndTime: Long,
+ endOffsets: CollectionAccumulator[PartitionOffsetWithIndex])
+ extends PartitionReaderFactory
+ with Logging {
+ override def createReader(partition: InputPartition): PartitionReader[InternalRow] = {
+ val rowReader = partitionReaderFactory.createReader(partition)
+ assert(rowReader.isInstanceOf[SupportsRealTimeRead[InternalRow]])
+ logInfo(
+ log"Creating low latency PartitionReader, stopping at " +
+ log"${MDC(LogKeys.TO_TIME, lowLatencyEndTime)}"
+ )
+ LowLatencyReaderWrap(
+ rowReader.asInstanceOf[SupportsRealTimeRead[InternalRow]],
+ lowLatencyEndTime,
+ endOffsets
+ )
+ }
+}
+
+/**
+ * Physical plan node for scanning a micro-batch of data from a data source.
+ */
Review Comment:
Could we add more doc to clarify this is especially for RTM?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -1078,6 +1182,73 @@ class MicroBatchExecution(
sparkSessionForStream.sessionState.conf)) {
updateStateStoreCkptId(execCtx, latestExecPlan)
}
+
+ var needSignalProgressLock = false
+ // In real-time mode, we delay the offset logging until the end of the batch.
+ // We first gather the offsets processed up to from all RealTimeStreamScanExec,
+ // i.e. tasks that execute a source partition. We merge the offsets and
+ // write them to the offset log
+ if (trigger.isInstanceOf[RealTimeTrigger]) {
+ val actualLastExecution = execCtx.executionPlan
+
+ val execs = StreamingQueryPlanTraverseHelper
+ .collectFromUnfoldedLatestExecutedPlan(actualLastExecution) {
+ case e: RealTimeStreamScanExec => e
+ }
+
+ val endOffsetMap = MutableMap[SparkDataStream, OffsetV2]()
+ execs.foreach { e =>
+ val lowLatencyExec = e.asInstanceOf[RealTimeStreamScanExec]
+ val accus: Seq[PartitionOffsetWithIndex] =
+ lowLatencyExec.endOffsetsAccumulator.value.asScala.toSeq
+ val sortedPartitionOffsets = accus.sortBy(_.index).map(_.partitionOffset).toArray
+ val source = e.stream
+ val endOffset = source
+ .asInstanceOf[SupportsRealTimeMode]
+ .mergeOffsets(sortedPartitionOffsets)
+ endOffsetMap += (source -> endOffset)
+ }
+
+ assert(endOffsetMap.size == execs.size, "Identical sources exist in the physical nodes" +
+ " which is not supported.")
+
+ execCtx.endOffsets ++= endOffsetMap
+ execCtx.recordEndOffsets(execCtx.endOffsets)
+ execCtx.recordTriggerOffsets(
+ from = execCtx.startOffsets,
+ to = execCtx.endOffsets,
+ latest = execCtx.latestOffsets
+ )
+ execCtx.reportTimeTaken("walCommit") {
+ if (!offsetLog.add(
+ execCtx.batchId,
+ execCtx.endOffsets.toOffsetSeq(sources, execCtx.offsetSeqMetadata)
+ )) {
+ throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
+ }
+ }
+ logInfo(
+ log"Committed offsets for batch ${MDC(LogKeys.BATCH_ID, execCtx.batchId)}. Metadata " +
+ log"${MDC(LogKeys.OFFSET_SEQUENCE_METADATA, execCtx.offsetSeqMetadata)}"
+ )
+ var shouldUpdate = true
+ sources.foreach { s =>
+ execCtx.startOffsets.get(s).foreach { prevOffsets =>
+ if (!prevOffsets.equals(endOffsetMap(s))) {
+ shouldUpdate = false
+ }
+ }
+ }
+ if (shouldUpdate) {
+ // To trigger processAllAvailable() return.
+ noNewData = true
+ // We could signal here, but when the test thread sees ProcessAllAvailable(), but
+ // signaling after commit log will make it less likely that the caller of
+ // ProcessAllAvailable() sees offset log written but not commit log.
Review Comment:
I think I got what this tries to say, but seems a bit hard to read. Could we rephrase it?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RealTimeStreamScanExec.scala:
##########
@@ -37,3 +49,121 @@ object LowLatencyClock {
clock = inputClock
}
}
+
+/**
+ * A wrap reader that turns a Partition Reader extending SupportsRealTimeRead to a
+ * normal PartitionReader and follow the task termination time `lowLatencyEndTime`, and
+ * report end offsets in the end to `endOffsets`.
+ */
+case class LowLatencyReaderWrap(
+ reader: SupportsRealTimeRead[InternalRow],
+ lowLatencyEndTime: Long,
+ endOffsets: CollectionAccumulator[PartitionOffsetWithIndex])
+ extends PartitionReader[InternalRow] {
+
+ override def next(): Boolean = {
+ val curTime = LowLatencyClock.getTimeMillis()
+ val ret = if (curTime >= lowLatencyEndTime) {
+ RecordStatus.newStatusWithoutArrivalTime(false)
+ } else {
+ reader.nextWithTimeout(lowLatencyEndTime - curTime)
+ }
+
+ if (!ret.hasRecord) {
+ // The way of using TaskContext.get().partitionId() to map to a partition
+ // may be fragile as it relies on thread locals.
+ endOffsets.add(
+ new PartitionOffsetWithIndex(TaskContext.get().partitionId(), reader.getOffset)
+ )
+ }
+ ret.hasRecord
+ }
+
+ override def get(): InternalRow = {
+ reader.get()
+ }
+
+ override def close(): Unit = {}
+}
+
+/**
+ * Wrapper factory that creates LowLatencyReaderWrap from reader as SupportsRealTimeRead
+ */
+case class LowLatencyReaderFactoryWrap(
+ partitionReaderFactory: PartitionReaderFactory,
+ lowLatencyEndTime: Long,
+ endOffsets: CollectionAccumulator[PartitionOffsetWithIndex])
+ extends PartitionReaderFactory
+ with Logging {
+ override def createReader(partition: InputPartition): PartitionReader[InternalRow] = {
+ val rowReader = partitionReaderFactory.createReader(partition)
+ assert(rowReader.isInstanceOf[SupportsRealTimeRead[InternalRow]])
+ logInfo(
+ log"Creating low latency PartitionReader, stopping at " +
+ log"${MDC(LogKeys.TO_TIME, lowLatencyEndTime)}"
+ )
+ LowLatencyReaderWrap(
+ rowReader.asInstanceOf[SupportsRealTimeRead[InternalRow]],
+ lowLatencyEndTime,
+ endOffsets
+ )
+ }
+}
+
+/**
+ * Physical plan node for scanning a micro-batch of data from a data source.
+ */
+case class RealTimeStreamScanExec(
+ output: Seq[Attribute],
+ @transient scan: Scan,
+ @transient stream: MicroBatchStream,
+ @transient start: Offset,
+ batchDurationMs: Long)
+ extends DataSourceV2ScanExecBase {
+
+ override def keyGroupedPartitioning: Option[Seq[Expression]] = None
+
+ override def ordering: Option[Seq[SortOrder]] = None
+
+ val endOffsetsAccumulator: CollectionAccumulator[PartitionOffsetWithIndex] = {
+ assert(stream.isInstanceOf[SupportsRealTimeMode])
Review Comment:
Seems this assert can moved out from `endOffsetsAccumulator` initialization into `RealTimeStreamScanExec`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ProgressReporter.scala:
##########
@@ -244,6 +244,14 @@ abstract class ProgressContext(
currentTriggerLatestOffsets = latest.transform((_, v) => v.json)
}
+ /**
+ * Only used by Real-time Mode. For other cases, end offsets are determined
+ * in the batch planning phase so it is never need to be updated.
+ */
+ def recordEndOffsets(to: StreamProgress): Unit = {
+ currentTriggerEndOffsets = to.transform((_, v) => v.json)
+ }
Review Comment:
Can we move this to `MicroBatchExecutionContext`? I think it is particular instead of a general method for all `ProgressContext`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -333,6 +375,29 @@ class MicroBatchExecution(
private def initializeExecution(
sparkSessionForStream: SparkSession): MicroBatchExecutionContext = {
+ var latestStartedBatch = offsetLog.getLatest()
+ val latestCommittedBatch = commitLog.getLatest()
+
+ val lastCommittedBatchId = latestCommittedBatch match {
+ case Some((batchId, _)) => batchId
+ case _ => -1L
+ }
+
+ // For a query running in Real-time Mode that fails after
+ // writing to offset log but before writing to commit log, we delete the extra
+ // entries in offsetLog to sync up. Note that this also means async checkpoint rollback handling
+ // is not compatible with low latency mode at this stage.
Review Comment:
low latency mode -> Real-time Mode. It should be better to keep consistent term in the doc.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RealTimeStreamScanExec.scala:
##########
@@ -37,3 +49,121 @@ object LowLatencyClock {
clock = inputClock
}
}
+
+/**
+ * A wrap reader that turns a Partition Reader extending SupportsRealTimeRead to a
+ * normal PartitionReader and follow the task termination time `lowLatencyEndTime`, and
+ * report end offsets in the end to `endOffsets`.
+ */
+case class LowLatencyReaderWrap(
+ reader: SupportsRealTimeRead[InternalRow],
+ lowLatencyEndTime: Long,
+ endOffsets: CollectionAccumulator[PartitionOffsetWithIndex])
+ extends PartitionReader[InternalRow] {
+
+ override def next(): Boolean = {
+ val curTime = LowLatencyClock.getTimeMillis()
+ val ret = if (curTime >= lowLatencyEndTime) {
+ RecordStatus.newStatusWithoutArrivalTime(false)
+ } else {
+ reader.nextWithTimeout(lowLatencyEndTime - curTime)
+ }
+
+ if (!ret.hasRecord) {
+ // The way of using TaskContext.get().partitionId() to map to a partition
+ // may be fragile as it relies on thread locals.
Review Comment:
What does it mean? Is it a problem for RTM code?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionOffsetWithIndex.scala:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.internal.connector;
+
+import java.io.Serializable;
+
+import org.apache.spark.sql.connector.read.streaming.PartitionOffset;
+
+/**
+ * Internal class for real time mode to pass partition offset from executors to the driver.
+ */
+private[sql] case class PartitionOffsetWithIndex(index: Long, partitionOffset: PartitionOffset)
+ extends Serializable;
Review Comment:
I think Scala case class automatically extends `Serializable`.
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2461485055
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RealTimeStreamScanExec.scala:
##########
@@ -37,3 +49,121 @@ object LowLatencyClock {
clock = inputClock
}
}
+
+/**
+ * A wrap reader that turns a Partition Reader extending SupportsRealTimeRead to a
+ * normal PartitionReader and follow the task termination time `lowLatencyEndTime`, and
+ * report end offsets in the end to `endOffsets`.
+ */
+case class LowLatencyReaderWrap(
+ reader: SupportsRealTimeRead[InternalRow],
+ lowLatencyEndTime: Long,
+ endOffsets: CollectionAccumulator[PartitionOffsetWithIndex])
+ extends PartitionReader[InternalRow] {
+
+ override def next(): Boolean = {
+ val curTime = LowLatencyClock.getTimeMillis()
+ val ret = if (curTime >= lowLatencyEndTime) {
+ RecordStatus.newStatusWithoutArrivalTime(false)
+ } else {
+ reader.nextWithTimeout(lowLatencyEndTime - curTime)
+ }
+
+ if (!ret.hasRecord) {
+ // The way of using TaskContext.get().partitionId() to map to a partition
+ // may be fragile as it relies on thread locals.
Review Comment:
It is currently not a problem but the implementation may be fragile to changes, e.g. a scenario that when this method is called is not from same thread as the task thread.
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #52620:
URL: https://github.com/apache/spark/pull/52620#issuecomment-3454134562
All streaming and SQL tests are passing.
Only test failure is:
```
[info] OracleIntegrationSuite:
[info] org.apache.spark.sql.jdbc.OracleIntegrationSuite *** ABORTED *** (10 minutes, 16 seconds)
[info] The code passed to eventually never returned normally. Attempted 599 times over 10.016333858333333 minutes. Last failure message: ORA-12541: Cannot connect. No listener at host 10.1.1.134 port 42263. (CONNECTION_ID=pnBBEOlFSbqGfcL5vuHDIQ==)
```
It looks like unrelated.
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2459072601
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -3155,6 +3155,11 @@ object SparkContext extends Logging {
/** Separator of tags in SPARK_JOB_TAGS property */
private[spark] val SPARK_JOB_TAGS_SEP = ","
+ def isDriver(executorId: String): Boolean = {
+ // Check for null to match the behavior of executorId == DRIVER_IDENTIFIER
+ executorId != null && executorId.startsWith(DRIVER_IDENTIFIER)
+ }
Review Comment:
This looks like only used in test? Maybe we don't need to it to `SparkContext` but just in the test?
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #52620:
URL: https://github.com/apache/spark/pull/52620#issuecomment-3453498078
There is a test error in [Run Docker integration tests](https://github.com/jerrypeng/spark/actions/runs/18830494387/job/53723712763). But looks like unrelated. @jerrypeng Can you re-trigger it?
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2461960426
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ProgressReporter.scala:
##########
@@ -244,6 +244,14 @@ abstract class ProgressContext(
currentTriggerLatestOffsets = latest.transform((_, v) => v.json)
}
+ /**
+ * Only used by Real-time Mode. For other cases, end offsets are determined
+ * in the batch planning phase so it is never need to be updated.
+ */
+ def recordEndOffsets(to: StreamProgress): Unit = {
+ currentTriggerEndOffsets = to.transform((_, v) => v.json)
+ }
Review Comment:
The reason this method is in ProgressReporter is for consistency since the method just before this one is "recordTriggerOffsets" which offers similar functionality.
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on PR #52620:
URL: https://github.com/apache/spark/pull/52620#issuecomment-3445176830
@viirya thank you for the review! I have addressed your comments. PTAL.
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2459073303
##########
core/src/main/scala/org/apache/spark/util/Clock.scala:
##########
@@ -58,7 +58,7 @@ private[spark] trait Clock {
/**
* A clock backed by the actual time from the OS as reported by the `System` API.
*/
-private[spark] class SystemClock extends Clock {
+private[spark] class SystemClock extends Clock with Serializable {
Review Comment:
Why we need this?
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2461452571
##########
core/src/main/scala/org/apache/spark/util/Clock.scala:
##########
@@ -58,7 +58,7 @@ private[spark] trait Clock {
/**
* A clock backed by the actual time from the OS as reported by the `System` API.
*/
-private[spark] class SystemClock extends Clock {
+private[spark] class SystemClock extends Clock with Serializable {
Review Comment:
We are using it in "LowLatencyClock":
https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RealTimeStreamScanExec.scala#L24
This clock will be distributed to tasks so it needs to be serializable
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2461836339
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/MicroBatchExecution.scala:
##########
@@ -263,6 +290,12 @@ class MicroBatchExecution(
case s => s -> ReadLimit.allAvailable()
}.toMap
}
+ if (trigger.isInstanceOf[RealTimeTrigger] && uniqueSources.size != sources.size) {
+ throw new SparkIllegalStateException(
+ errorClass = s"STREAMING_REAL_TIME_MODE.IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED",
Review Comment:
There are only two types of operators that can have more than 1 streaming source for streaming queries. It is union and stream-stream join. Stream-stream join is not going to be supported for now because it is a stateful operator ( I will have a upcoming PR that checks for this). Thus, union is currently the only operator in which this condition can occur for.
--
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
Re: [PR] [SPARK-53915] Add RealTimeScanExec and ability to execute long running batches [spark]
Posted by "jerrypeng (via GitHub)" <gi...@apache.org>.
jerrypeng commented on code in PR #52620:
URL: https://github.com/apache/spark/pull/52620#discussion_r2462187188
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryPlanTraverseHelper.scala:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.runtime
+
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
+
+/**
+ * This is a utility object placing methods to traverse the query plan for streaming query.
+ * This is used for patterns of traversal which are repeated in multiple places.
+ */
+object StreamingQueryPlanTraverseHelper {
+ def collectFromUnfoldedLatestExecutedPlan[B](
+ lastExecution: IncrementalExecution)(pf: PartialFunction[SparkPlan, B]): Seq[B] = {
+ val executedPlan = lastExecution.executedPlan
+
+ collectWithUnfolding(executedPlan)(pf)
+ }
+
+ private def collectWithUnfolding[B](executedPlan: SparkPlan)(
+ pf: PartialFunction[SparkPlan, B]): Seq[B] = {
+ executedPlan.flatMap {
+ // InMemoryTableScanExec is a node to represent a cached plan. The node has underlying
+ // actual executed plan, which we should traverse to collect the required information.
+ case s: InMemoryTableScanExec => collectWithUnfolding(s.relation.cachedPlan)(pf)
Review Comment:
After some additional thinking, these conditions this code checks is not relevant to RTM. I will just remove this code.
--
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