HeartSaVioR commented on code in PR #40561:
URL: https://github.com/apache/spark/pull/40561#discussion_r2096700034
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala:
##########
@@ -980,3 +1023,67 @@ object StreamingDeduplicateExec {
private val EMPTY_ROW =
UnsafeProjection.create(Array[DataType](NullType)).apply(InternalRow.apply(null))
}
+
+case class StreamingDeduplicateWithinWatermarkExec(
+ keyExpressions: Seq[Attribute],
+ child: SparkPlan,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ eventTimeWatermarkForLateEvents: Option[Long] = None,
+ eventTimeWatermarkForEviction: Option[Long] = None)
+ extends BaseStreamingDeduplicateExec {
+
+ protected val schemaForValueRow: StructType = StructType(
+ Array(StructField("expiresAtMicros", LongType, nullable = false)))
+
+ protected val extraOptionOnStateStore: Map[String, String] = Map.empty
+
+ private val eventTimeCol: Attribute =
WatermarkSupport.findEventTimeColumn(child.output,
+ allowMultipleEventTimeColumns = false).get
+ private val delayThresholdMs =
eventTimeCol.metadata.getLong(EventTimeWatermark.delayKey)
+ private val eventTimeColOrdinal: Int = child.output.indexOf(eventTimeCol)
+
+ protected def initializeReusedDupInfoRow(): Option[UnsafeRow] = {
+ val timeoutToUnsafeRow = UnsafeProjection.create(schemaForValueRow)
+ val timeoutRow = timeoutToUnsafeRow(new
SpecificInternalRow(schemaForValueRow))
+ Some(timeoutRow)
+ }
+
+ protected def putDupInfoIntoState(
+ store: StateStore,
+ data: UnsafeRow,
+ key: UnsafeRow,
+ reusedDupInfoRow: Option[UnsafeRow]): Unit = {
+ assert(reusedDupInfoRow.isDefined, "This should have reused row.")
+ val timeoutRow = reusedDupInfoRow.get
+
+ // We expect data type of event time column to be TimestampType or
TimestampNTZType which both
+ // are internally represented as Long.
+ val timestamp = data.getLong(eventTimeColOrdinal)
+ // The unit of timestamp in Spark is microseconds, convert the delay
threshold to micros.
+ val expiresAt = timestamp + DateTimeUtils.millisToMicros(delayThresholdMs)
Review Comment:
What you can get from watermark is to confirm that the logical time passes
through. Again, this is unavoidable due to events out of order. What if 13:59
comes earlier than 12:00? We will need to ensure the duplication state info to
not be evicted for 2 hours "before". That's due to the contract of API.
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
