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 2021/03/29 04:45:37 UTC

[GitHub] [spark] HeartSaVioR opened a new pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

HeartSaVioR opened a new pull request #31989:
URL: https://github.com/apache/spark/pull/31989


   Introduction: this PR is a part of SPARK-10816 (`EventTime based sessionization (session window)`). Please refer #31937 to see the overall view of the code change. (Note that code diff could be diverged a bit.)
   
   ### What changes were proposed in this pull request?
   
   This PR introduces state store manager for session window in streaming query. Session window in batch query wouldn't need to leverage state store manager.
   
   This PR ensures versioning on state format for state store manager, so that we can apply further optimization after releasing Spark version. StreamingSessionWindowStateManager is a trait defining the available methods in session window state store manager. StreamingSessionWindowStateManagerBaseImpl and its subclasses are classes implementing the trait with versioning.
   
   The format of version 1 leverages two state stores to represent the session windows:
   
   * key -> list of start times (in session window spec)
   * key + start time in session window -> value
   
   This structure is simpler compared to what we tried to implement in history, and also less sub-optimal as it doesn't require all values to be rewritten when any of session window is added/modified/removed.
   
   ### Why are the changes needed?
   
   This part is a one of required on implementing SPARK-10816.
   
   ### Does this PR introduce _any_ user-facing change?
   
   No.
   
   ### How was this patch tested?
   
   WIP (new test suite is expected to be added, or can be skipped if we agree it can be skipped)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822220644


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42130/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878637665






-- 
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867494004


   **[Test build #140257 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140257/testReport)** for PR 31989 at commit [`efab51d`](https://github.com/apache/spark/commit/efab51d6783651d529cc1cc000e77ef89ce5b3cc).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822352693


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137560/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822229073


   **[Test build #137584 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137584/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868457383


   **[Test build #140313 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140313/testReport)** for PR 31989 at commit [`8fe3e23`](https://github.com/apache/spark/commit/8fe3e236ff1e83feafe3c409a2dae26083b9c875).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds the following public classes _(experimental)_:
     * `sealed trait StreamingSessionWindowStateManager extends Serializable `
     * `class StreamingSessionWindowStateManagerImplV1(`
     * `class StreamingSessionWindowHelper(sessionExpression: Attribute, inputSchema: Seq[Attribute]) `
     * `  case class WindowRecord(start: Long, end: Long, isNew: Boolean, row: UnsafeRow)`


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r668492722



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */

Review comment:
       Proposed sentence looks nice. Thanks!

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   * @param sessions The all sessions including existing sessions if it's active.
+   *                 Existing sessions which aren't included in this parameter will be removed.
+   */
+  def updateSessions(store: StateStore, key: UnsafeRow, sessions: Seq[UnsafeRow]): Long
+
+  /**
+   * Removes using a predicate on values, with returning removed values via iterator.
+   *
+   * At a high level, this produces an iterator over the (key, value, matched) tuples such that
+   * value satisfies the predicate, where producing an element removes the value from the
+   * state store and producing all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   *
+   * @param removalCondition The predicate on removing the key-value.
+   */
+  def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]
+
+  /**
+   * Return an iterator containing all the sessions. Implementations must ensure that updates
+   * (puts, removes) can be made while iterating over this iterator.
+   */
+  def iterator(store: ReadStateStore): Iterator[UnsafeRow]
+
+  /**
+   * Commits the change.
+   */
+  def commit(store: StateStore): Long
+
+  /**
+   * Aborts the change.
+   */
+  def abortIfNeeded(store: StateStore): Unit
+}
+
+object StreamingSessionWindowStateManager {
+  val supportedVersions = Seq(1)
+
+  def createStateManager(
+      keyWithoutSessionExpressions: Seq[Attribute],
+      sessionExpression: Attribute,
+      inputRowAttributes: Seq[Attribute],
+      stateFormatVersion: Int): StreamingSessionWindowStateManager = {
+    stateFormatVersion match {
+      case 1 => new StreamingSessionWindowStateManagerImplV1(
+        keyWithoutSessionExpressions, sessionExpression, inputRowAttributes)
+      case _ => throw new IllegalArgumentException(s"Version $stateFormatVersion is invalid")
+    }
+  }
+}
+
+class StreamingSessionWindowStateManagerImplV1(
+    keyWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    valueAttributes: Seq[Attribute])
+  extends StreamingSessionWindowStateManager with Logging {
+
+  @transient private lazy val keyWithoutSessionRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+
+  private val stateKeyStructType = keyWithoutSessionExpressions.toStructType
+    .add("sessionStartTime", TimestampType, nullable = false)
+
+  private val stateKeyExprs = keyWithoutSessionExpressions :+ Literal(1L)
+  private val indexOrdinalInSessionStart = keyWithoutSessionExpressions.size
+  @transient private lazy val keyRowGenerator = UnsafeProjection.create(

Review comment:
       Nice finding. Looks like I missed to remove one later.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.

Review comment:
       The new number of session windows (state rows) for the key after applying this method.
   
   I'll update the doc. Nice finding.




-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812964782


   **[Test build #136887 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136887/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878577173


   **[Test build #140936 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140936/testReport)** for PR 31989 at commit [`8638baf`](https://github.com/apache/spark/commit/8638baf0066b9b49983f525b9f0db230aca4201c).


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879187376


   **[Test build #140971 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140971/testReport)** for PR 31989 at commit [`f12cda8`](https://github.com/apache/spark/commit/f12cda8151abdcd25ab4cd88a53e3f4bb6e725b8).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867233236


   I haven't submitted two PRs based on this change. We will have more PRs soon.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810790360


   **[Test build #136742 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136742/testReport)** for PR 31989 at commit [`25bbd47`](https://github.com/apache/spark/commit/25bbd4772a08b0c19d1cd305ef82d26b922a21e9).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867679763


   **[Test build #140257 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140257/testReport)** for PR 31989 at commit [`efab51d`](https://github.com/apache/spark/commit/efab51d6783651d529cc1cc000e77ef89ce5b3cc).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds the following public classes _(experimental)_:
     * `sealed trait StreamingSessionWindowStateManager extends Serializable `
     * `class StreamingSessionWindowStateManagerImplV1(`
     * `class StreamingSessionWindowHelper(sessionExpression: Attribute, inputSchema: Seq[Attribute]) `
     * `  case class WindowRecord(start: Long, end: Long, isNew: Boolean, row: UnsafeRow)`


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822298092


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42140/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810891685


   **[Test build #136756 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136756/testReport)** for PR 31989 at commit [`da08bd4`](https://github.com/apache/spark/commit/da08bd408ac74d8cdccf3ebf50d8fb3591642f5f).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811665284


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41375/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878609113


   **[Test build #140937 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140937/testReport)** for PR 31989 at commit [`8638baf`](https://github.com/apache/spark/commit/8638baf0066b9b49983f525b9f0db230aca4201c).


-- 
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] SparkQA commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810713055


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41324/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811659949


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41375/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878692990


   **[Test build #140935 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140935/testReport)** for PR 31989 at commit [`7372012`](https://github.com/apache/spark/commit/737201249e2f243dce5f3d9b522eac2bb252885b).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds the following public classes _(experimental)_:
     * `sealed trait StreamingSessionWindowStateManager extends Serializable `
     * `class StreamingSessionWindowStateManagerImplV1(`
     * `class StreamingSessionWindowHelper(sessionExpression: Attribute, inputSchema: Seq[Attribute]) `
     * `  case class WindowRecord(start: Long, end: Long, isNew: Boolean, row: UnsafeRow)`


-- 
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] AmplabJenkins commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810814443


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136742/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868296368


   **[Test build #140313 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140313/testReport)** for PR 31989 at commit [`8fe3e23`](https://github.com/apache/spark/commit/8fe3e236ff1e83feafe3c409a2dae26083b9c875).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 a change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r668462823



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManagerSuite.scala
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.state
+
+import java.util.UUID
+
+import org.apache.hadoop.conf.Configuration
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.execution.streaming.StatefulOperatorStateInfo
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.streaming.StreamTest
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+class StreamingSessionWindowStateManagerSuite extends StreamTest with BeforeAndAfter {
+  private val rowSchema = new StructType().add("key1", StringType).add("key2", IntegerType)
+    .add("session", new StructType().add("start", LongType).add("end", LongType))
+    .add("value", LongType)
+  private val rowAttributes = rowSchema.toAttributes
+
+  private val keysWithoutSessionAttributes = rowAttributes.filter { attr =>
+    List("key1", "key2").contains(attr.name)
+  }
+
+  private val sessionAttribute = rowAttributes.filter(_.name == "session").head
+
+  private val inputValueGen = UnsafeProjection.create(rowAttributes.map(_.dataType).toArray)
+  private val inputKeyGen = UnsafeProjection.create(
+    keysWithoutSessionAttributes.map(_.dataType).toArray)
+
+  before {
+    SparkSession.setActiveSession(spark) // set this before force initializing 'joinExec'

Review comment:
       `joinExec`?




-- 
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 change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r606885811



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,445 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.hadoop.conf.Configuration
+
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, SpecificInternalRow, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.execution.streaming.StatefulOperatorStateInfo
+import org.apache.spark.sql.types.{ArrayType, StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+/**
+ * Base trait for state manager purposed to be used from streaming session window aggregation.
+ */
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+
+  def getKey(row: InternalRow): UnsafeRow
+
+  def getStartTime(row: InternalRow): Long
+
+  /**
+   * Returns all stored keys.
+   */
+  def getAllKeys(): Iterator[UnsafeRow]
+
+  /**
+   * Returns a list of states for the key. These states are candidates for session window
+   * merging.
+   */
+  def getStates(key: UnsafeRow): Seq[UnsafeRow]
+
+  /**
+   * Returns a list of start times for session windows belonging to the given key.
+   */
+  def getStartTimeList(key: UnsafeRow): Seq[Long]
+
+  /**
+   * Returns the state of given key and start time.
+   */
+  def getState(key: UnsafeRow, startTime: Long): UnsafeRow
+
+  /**
+   * Returns a list of states for all keys.
+   */
+  def getStates(): Seq[UnsafeRow]
+
+  /**
+   * Puts a state row into the state with given key and start time. Note that this method
+   * does not update the start time into the list of start times for the given key. To update
+   * the list, please call `putStartTimeList`.
+   */
+  def putState(key: UnsafeRow, startTime: Long, value: UnsafeRow): Unit
+
+  /**
+   * Puts a list of states for given key. This method will update the list of start times
+   * for the given key.
+   */
+  def putStates(key: UnsafeRow, values: Seq[UnsafeRow]): Unit
+
+  /**
+   * Puts a list of start times of session windows for given key into the state. The list
+   * of start times must be sorted.
+   */
+  def putStartTimeList(key: UnsafeRow, startTimes: Seq[Long]): Unit
+
+  /**
+   * Removes specified state for the given key and start time.
+   */
+  def removeState(key: UnsafeRow, startTime: Long): Unit
+
+  /**
+   * Removes given key from the state store. This method will remove all states associated
+   * with the given key if any.
+   */
+  def removeKey(key: UnsafeRow): Unit
+
+  /**
+   * Given a callback function used to update state store metrics, updates the metrics of all
+   * state stores.
+   */
+  def updateMetrics(updateFunc: StateStoreMetrics => Unit)
+
+  /**
+   * Remove using a predicate on values.
+   *
+   * At a high level, this produces an iterator over the values such that value satisfies the
+   * predicate, where producing an element removes the value from the state store and producing
+   * all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   */
+  def removeByValueCondition(removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]

Review comment:
       I moved the code on evicting state rows from store physical exec to state store impl, to separate the concern. Streaming join state store also does the same.
   
   I think store exec shouldn't know about the details for state store itself, that said, this may need to be applied to store new state rows as well which is also in store exec. 
   (I don't remember I addressed it in this PR. I'm sorry I'm not on my laptop and only mobile for following 3~4 days.)




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810715946


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41324/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867123703


   I think this is the last open sub-PRs for #31937. @HeartSaVioR @xuanyuanking so will this be changed after #33038 is merged?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868296368


   **[Test build #140313 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140313/testReport)** for PR 31989 at commit [`8fe3e23`](https://github.com/apache/spark/commit/8fe3e236ff1e83feafe3c409a2dae26083b9c875).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822349565


   **[Test build #137610 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137610/testReport)** for PR 31989 at commit [`d4f326a`](https://github.com/apache/spark/commit/d4f326afb016f43792810ac1782e1928a893ba4f).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810695572


   **[Test build #136742 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136742/testReport)** for PR 31989 at commit [`25bbd47`](https://github.com/apache/spark/commit/25bbd4772a08b0c19d1cd305ef82d26b922a21e9).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822257605


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42137/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822349565


   **[Test build #137610 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137610/testReport)** for PR 31989 at commit [`d4f326a`](https://github.com/apache/spark/commit/d4f326afb016f43792810ac1782e1928a893ba4f).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878572120


   **[Test build #140935 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140935/testReport)** for PR 31989 at commit [`7372012`](https://github.com/apache/spark/commit/737201249e2f243dce5f3d9b522eac2bb252885b).


-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878945636


   Thanks @viirya for reviewing! I reflected review comments.
   
   Btw, I also made a change via f12cda8 as I realized we don't need to union windows from both (old/new) sides.
   (Actually I found this during experiment but I forgot to reflect here. My bad.)
   
   Simply thinking, given old windows and new windows, we can simply remove all old windows and add all new windows as new windows are the thing we want to "retain" in state store. (See method contract on `updateSessions`.) Adding some optimization on reducing "delta"/"tombstone", we can remove old windows which state keys are not updated by new windows, and add new windows to do "upsert".


-- 
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822581785


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137610/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822274890






-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822414283


   **[Test build #137584 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137584/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810936321


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41339/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878632330


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45449/
   


-- 
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 edited a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR edited a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-813142499


   I addressed it in this PR. Please look for the recent commit to narrow the diff.
   
   https://github.com/apache/spark/commit/25bbd4772a08b0c19d1cd305ef82d26b922a21e9


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878600359


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45447/
   


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867486372


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44783/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 a change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r668488747



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */

Review comment:
       This looks confusing. Maybe "Returns true if the session of the given value doesn't exist in the store, or the value in the session is different to the stored value of the session in the store."




-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878582626


   retest this, please


-- 
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868372621


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44845/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878710457


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140937/
   


-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811622439


   retest this, please


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867687888


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140257/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867564596


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44786/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809062214


   **[Test build #136632 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136632/testReport)** for PR 31989 at commit [`a7bd8a9`](https://github.com/apache/spark/commit/a7bd8a91c52970165e668b6a8d07ade2899c915e).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867564596


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44786/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810936321


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41339/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822216802


   gentle ping @xuanyuanking 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822298092


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42140/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822416014


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137584/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878579841


   Thanks @HeartSaVioR. I'll review this today/tomorrow.


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822563424


   **[Test build #137610 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137610/testReport)** for PR 31989 at commit [`d4f326a`](https://github.com/apache/spark/commit/d4f326afb016f43792810ac1782e1928a893ba4f).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822350578


   **[Test build #137560 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137560/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810936283


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41339/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811078011


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136756/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811058596


   **[Test build #136756 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136756/testReport)** for PR 31989 at commit [`da08bd4`](https://github.com/apache/spark/commit/da08bd408ac74d8cdccf3ebf50d8fb3591642f5f).
    * This patch **fails PySpark unit tests**.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812936839


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41463/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878646422


   Kubernetes integration test status success
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45449/
   


-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878577022


   cc. @viirya @xuanyuanking I've rebased this one and this is ready to review. Please take a look. 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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868470452


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140313/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879189149


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140971/
   


-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867225823


   > I think this is the last open sub-PRs for #31937.
   
   Unfortunately no. We have 4 open sub-tasks in [SPARK-10816](https://issues.apache.org/jira/browse/SPARK-10816) including newly added prefix scan, and I plan to submit a PR per sub-task. So in overall I still need to submit two more PRs.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878972400


   **[Test build #140971 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140971/testReport)** for PR 31989 at commit [`f12cda8`](https://github.com/apache/spark/commit/f12cda8151abdcd25ab4cd88a53e3f4bb6e725b8).


-- 
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810891685


   **[Test build #136756 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136756/testReport)** for PR 31989 at commit [`da08bd4`](https://github.com/apache/spark/commit/da08bd408ac74d8cdccf3ebf50d8fb3591642f5f).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811665311


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41375/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 edited a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR edited a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-813142499


   I addressed it in this PR. Please look for the recent commits to narrow the diff.
   
   https://github.com/apache/spark/commit/25bbd4772a08b0c19d1cd305ef82d26b922a21e9


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810695572


   **[Test build #136742 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136742/testReport)** for PR 31989 at commit [`25bbd47`](https://github.com/apache/spark/commit/25bbd4772a08b0c19d1cd305ef82d26b922a21e9).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867452234


   **[Test build #140254 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140254/testReport)** for PR 31989 at commit [`efab51d`](https://github.com/apache/spark/commit/efab51d6783651d529cc1cc000e77ef89ce5b3cc).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867492787


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44783/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822257605


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42137/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-813135718


   > I just simply remedied write amplification a bit via determining existing values to be "deleted" or "replaced", instead of blindly judging it as "deleted".
   
   Does this reflect in this PR or original one?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879347792


   Thanks all for reviewing and merging! I'll update the next PR. We have two PRs now.


-- 
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] AmplabJenkins commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809230269


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41215/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811635030


   **[Test build #136793 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136793/testReport)** for PR 31989 at commit [`da08bd4`](https://github.com/apache/spark/commit/da08bd408ac74d8cdccf3ebf50d8fb3591642f5f).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822581785


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137610/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878972400


   **[Test build #140971 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140971/testReport)** for PR 31989 at commit [`f12cda8`](https://github.com/apache/spark/commit/f12cda8151abdcd25ab4cd88a53e3f4bb6e725b8).


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878606667


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45448/
   


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878709830


   **[Test build #140937 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140937/testReport)** for PR 31989 at commit [`8638baf`](https://github.com/apache/spark/commit/8638baf0066b9b49983f525b9f0db230aca4201c).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878572120


   **[Test build #140935 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140935/testReport)** for PR 31989 at commit [`7372012`](https://github.com/apache/spark/commit/737201249e2f243dce5f3d9b522eac2bb252885b).


-- 
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812966731


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136887/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868361374


   Kubernetes integration test status success
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44845/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878577173


   **[Test build #140936 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140936/testReport)** for PR 31989 at commit [`8638baf`](https://github.com/apache/spark/commit/8638baf0066b9b49983f525b9f0db230aca4201c).


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810930949


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41339/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822229073


   **[Test build #137584 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137584/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822165654


   **[Test build #137560 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137560/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-813142499


   I addressed it in this PR. Please look for the recent commits to narrow the diff.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811635030


   **[Test build #136793 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136793/testReport)** for PR 31989 at commit [`da08bd4`](https://github.com/apache/spark/commit/da08bd408ac74d8cdccf3ebf50d8fb3591642f5f).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809217116


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41215/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811791017


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136793/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878710457


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140937/
   


-- 
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811791017


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136793/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812936839


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41463/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867455800


   Changed this to draft for now to prevent merging before merging #33038


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 edited a comment on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR edited a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809067233


   Except the test suite, one more thing worths to address here is write amplification; we "blindly" replace all start times and all sessions. This could bring unnecessary writes on "unmodified" existing sessions. In many cases we expect the new inputs will be bound and expanding to the existing sessions, but with very long watermark gap and old inputs which have various timestamps, the case could still happen.
   
   EDIT: I realized the logic is bound to the physical plan. Though it seems OK to move the logic to here so that the logic to store new session windows efficiently can be bound to the state format.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868372621


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44845/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867494004


   **[Test build #140257 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140257/testReport)** for PR 31989 at commit [`efab51d`](https://github.com/apache/spark/commit/efab51d6783651d529cc1cc000e77ef89ce5b3cc).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 a change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r668460586



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   * @param sessions The all sessions including existing sessions if it's active.
+   *                 Existing sessions which aren't included in this parameter will be removed.
+   */
+  def updateSessions(store: StateStore, key: UnsafeRow, sessions: Seq[UnsafeRow]): Long
+
+  /**
+   * Removes using a predicate on values, with returning removed values via iterator.
+   *
+   * At a high level, this produces an iterator over the (key, value, matched) tuples such that
+   * value satisfies the predicate, where producing an element removes the value from the
+   * state store and producing all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   *
+   * @param removalCondition The predicate on removing the key-value.
+   */
+  def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]
+
+  /**
+   * Return an iterator containing all the sessions. Implementations must ensure that updates
+   * (puts, removes) can be made while iterating over this iterator.
+   */
+  def iterator(store: ReadStateStore): Iterator[UnsafeRow]
+
+  /**
+   * Commits the change.
+   */
+  def commit(store: StateStore): Long
+
+  /**
+   * Aborts the change.
+   */
+  def abortIfNeeded(store: StateStore): Unit
+}
+
+object StreamingSessionWindowStateManager {
+  val supportedVersions = Seq(1)
+
+  def createStateManager(
+      keyWithoutSessionExpressions: Seq[Attribute],
+      sessionExpression: Attribute,
+      inputRowAttributes: Seq[Attribute],
+      stateFormatVersion: Int): StreamingSessionWindowStateManager = {
+    stateFormatVersion match {
+      case 1 => new StreamingSessionWindowStateManagerImplV1(
+        keyWithoutSessionExpressions, sessionExpression, inputRowAttributes)
+      case _ => throw new IllegalArgumentException(s"Version $stateFormatVersion is invalid")
+    }
+  }
+}
+
+class StreamingSessionWindowStateManagerImplV1(
+    keyWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    valueAttributes: Seq[Attribute])
+  extends StreamingSessionWindowStateManager with Logging {
+
+  @transient private lazy val keyWithoutSessionRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+
+  private val stateKeyStructType = keyWithoutSessionExpressions.toStructType
+    .add("sessionStartTime", TimestampType, nullable = false)
+
+  private val stateKeyExprs = keyWithoutSessionExpressions :+ Literal(1L)
+  private val indexOrdinalInSessionStart = keyWithoutSessionExpressions.size
+  @transient private lazy val keyRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+  @transient private lazy val stateKeyRowGenerator = UnsafeProjection.create(stateKeyExprs,
+    keyWithoutSessionExpressions)
+
+  @transient private lazy val helper = new StreamingSessionWindowHelper(
+    sessionExpression, valueAttributes)
+
+  override def getStateKeySchema: StructType = stateKeyStructType
+
+  override def getStateValueSchema: StructType = valueAttributes.toStructType
+
+  override def getNumColsForPrefixKey: Int = keyWithoutSessionExpressions.length
+
+  override def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow = {
+    keyWithoutSessionRowGenerator(value)
+  }
+
+  override def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean = {
+    val sessionStart = helper.extractTimePair(value)._1
+    val stateKey = getStateKey(getKey(value), sessionStart)
+    val stateRow = store.get(stateKey)
+    stateRow == null || !stateRow.equals(value)
+  }
+
+  override def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow] =
+    getSessionsWithKeys(store, key).map(_.value)
+
+  private def getSessionsWithKeys(
+      store: ReadStateStore,
+      key: UnsafeRow): Iterator[UnsafeRowPair] = {
+    store.prefixScan(key)
+  }
+
+  override def updateSessions(
+      store: StateStore,
+      key: UnsafeRow,
+      sessions: Seq[UnsafeRow]): Long = {
+    replaceModifyWindow(key, sessions, store)
+  }
+
+  override def commit(store: StateStore): Long = store.commit()
+
+  override def iterator(store: ReadStateStore): Iterator[UnsafeRow] =
+    iteratorWithKeys(store).map(_.value)
+
+  private def iteratorWithKeys(store: ReadStateStore): Iterator[UnsafeRowPair] = {
+    store.iterator()
+  }
+
+  override def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow] = {
+    new NextIterator[UnsafeRow] {
+      private val rangeIter = iteratorWithKeys(store)
+
+      override protected def getNext(): UnsafeRow = {
+        var removedValueRow: UnsafeRow = null
+        while(rangeIter.hasNext && removedValueRow == null) {
+          val rowPair = rangeIter.next()
+          if (removalCondition(rowPair.value)) {
+            store.remove(rowPair.key)
+            removedValueRow = rowPair.value
+          }
+        }
+
+        if (removedValueRow == null) {
+          finished = true
+          null
+        } else {
+          removedValueRow
+        }
+      }
+
+      override protected def close(): Unit = {}
+    }
+  }
+
+  private def getKey(value: UnsafeRow): UnsafeRow = keyRowGenerator(value)
+
+  private def getStateKey(key: UnsafeRow, sessionStart: Long): UnsafeRow = {
+    val stateKey = stateKeyRowGenerator(key)
+    stateKey.setLong(indexOrdinalInSessionStart, sessionStart)
+    stateKey.copy()
+  }
+
+  /**
+   * add to store, only update the changed window elements

Review comment:
       Could you rephrase the method doc? Maybe "Updates windows from the values overlapped with existing values in the store and adds new windows from the values."?




-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812935993






-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809230269


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41215/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878624695


   Kubernetes integration test status success
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45448/
   


-- 
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867456793


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140254/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867452234


   **[Test build #140254 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140254/testReport)** for PR 31989 at commit [`efab51d`](https://github.com/apache/spark/commit/efab51d6783651d529cc1cc000e77ef89ce5b3cc).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878705167


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140935/
   


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867456739


   **[Test build #140254 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140254/testReport)** for PR 31989 at commit [`efab51d`](https://github.com/apache/spark/commit/efab51d6783651d529cc1cc000e77ef89ce5b3cc).
    * This patch **fails to build**.
    * This patch merges cleanly.
    * This patch adds the following public classes _(experimental)_:
     * `sealed trait StreamingSessionWindowStateManager extends Serializable `
     * `class StreamingSessionWindowStateManagerImplV1(`
     * `class StreamingSessionWindowHelper(sessionExpression: Attribute, inputSchema: Seq[Attribute]) `
     * `  case class WindowRecord(start: Long, end: Long, isNew: Boolean, row: UnsafeRow)`


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878581254


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140936/
   


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812936934


   **[Test build #136887 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136887/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867456793


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140254/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811773041


   **[Test build #136793 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136793/testReport)** for PR 31989 at commit [`da08bd4`](https://github.com/apache/spark/commit/da08bd408ac74d8cdccf3ebf50d8fb3591642f5f).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879043117


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45485/
   


-- 
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822165654


   **[Test build #137560 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/137560/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822416014


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137584/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822199099


   Kubernetes integration test unable to build dist.
   
   exiting with code: 1
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/42130/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 a change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r606880784



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,445 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.hadoop.conf.Configuration
+
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, SpecificInternalRow, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.execution.streaming.StatefulOperatorStateInfo
+import org.apache.spark.sql.types.{ArrayType, StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+/**
+ * Base trait for state manager purposed to be used from streaming session window aggregation.
+ */
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+
+  def getKey(row: InternalRow): UnsafeRow
+
+  def getStartTime(row: InternalRow): Long
+
+  /**
+   * Returns all stored keys.
+   */
+  def getAllKeys(): Iterator[UnsafeRow]
+
+  /**
+   * Returns a list of states for the key. These states are candidates for session window
+   * merging.
+   */
+  def getStates(key: UnsafeRow): Seq[UnsafeRow]
+
+  /**
+   * Returns a list of start times for session windows belonging to the given key.
+   */
+  def getStartTimeList(key: UnsafeRow): Seq[Long]
+
+  /**
+   * Returns the state of given key and start time.
+   */
+  def getState(key: UnsafeRow, startTime: Long): UnsafeRow
+
+  /**
+   * Returns a list of states for all keys.
+   */
+  def getStates(): Seq[UnsafeRow]
+
+  /**
+   * Puts a state row into the state with given key and start time. Note that this method
+   * does not update the start time into the list of start times for the given key. To update
+   * the list, please call `putStartTimeList`.
+   */
+  def putState(key: UnsafeRow, startTime: Long, value: UnsafeRow): Unit
+
+  /**
+   * Puts a list of states for given key. This method will update the list of start times
+   * for the given key.
+   */
+  def putStates(key: UnsafeRow, values: Seq[UnsafeRow]): Unit
+
+  /**
+   * Puts a list of start times of session windows for given key into the state. The list
+   * of start times must be sorted.
+   */
+  def putStartTimeList(key: UnsafeRow, startTimes: Seq[Long]): Unit
+
+  /**
+   * Removes specified state for the given key and start time.
+   */
+  def removeState(key: UnsafeRow, startTime: Long): Unit
+
+  /**
+   * Removes given key from the state store. This method will remove all states associated
+   * with the given key if any.
+   */
+  def removeKey(key: UnsafeRow): Unit
+
+  /**
+   * Given a callback function used to update state store metrics, updates the metrics of all
+   * state stores.
+   */
+  def updateMetrics(updateFunc: StateStoreMetrics => Unit)
+
+  /**
+   * Remove using a predicate on values.
+   *
+   * At a high level, this produces an iterator over the values such that value satisfies the
+   * predicate, where producing an element removes the value from the state store and producing
+   * all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   */
+  def removeByValueCondition(removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]

Review comment:
       Seems this is not in my implementation. I also don't find it in #31937. Am I missing 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.

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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822352693


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/137560/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812966731


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136887/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867463664


   retest this, please


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878662960


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45449/
   


-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867226014


   > so will this be changed after #33038 is merged?
   
   Yes, I'll update this on top of #33038.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r606898837



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,445 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.hadoop.conf.Configuration
+
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, SpecificInternalRow, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.execution.streaming.StatefulOperatorStateInfo
+import org.apache.spark.sql.types.{ArrayType, StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+/**
+ * Base trait for state manager purposed to be used from streaming session window aggregation.
+ */
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+
+  def getKey(row: InternalRow): UnsafeRow
+
+  def getStartTime(row: InternalRow): Long
+
+  /**
+   * Returns all stored keys.
+   */
+  def getAllKeys(): Iterator[UnsafeRow]
+
+  /**
+   * Returns a list of states for the key. These states are candidates for session window
+   * merging.
+   */
+  def getStates(key: UnsafeRow): Seq[UnsafeRow]
+
+  /**
+   * Returns a list of start times for session windows belonging to the given key.
+   */
+  def getStartTimeList(key: UnsafeRow): Seq[Long]
+
+  /**
+   * Returns the state of given key and start time.
+   */
+  def getState(key: UnsafeRow, startTime: Long): UnsafeRow
+
+  /**
+   * Returns a list of states for all keys.
+   */
+  def getStates(): Seq[UnsafeRow]
+
+  /**
+   * Puts a state row into the state with given key and start time. Note that this method
+   * does not update the start time into the list of start times for the given key. To update
+   * the list, please call `putStartTimeList`.
+   */
+  def putState(key: UnsafeRow, startTime: Long, value: UnsafeRow): Unit
+
+  /**
+   * Puts a list of states for given key. This method will update the list of start times
+   * for the given key.
+   */
+  def putStates(key: UnsafeRow, values: Seq[UnsafeRow]): Unit
+
+  /**
+   * Puts a list of start times of session windows for given key into the state. The list
+   * of start times must be sorted.
+   */
+  def putStartTimeList(key: UnsafeRow, startTimes: Seq[Long]): Unit
+
+  /**
+   * Removes specified state for the given key and start time.
+   */
+  def removeState(key: UnsafeRow, startTime: Long): Unit
+
+  /**
+   * Removes given key from the state store. This method will remove all states associated
+   * with the given key if any.
+   */
+  def removeKey(key: UnsafeRow): Unit
+
+  /**
+   * Given a callback function used to update state store metrics, updates the metrics of all
+   * state stores.
+   */
+  def updateMetrics(updateFunc: StateStoreMetrics => Unit)
+
+  /**
+   * Remove using a predicate on values.
+   *
+   * At a high level, this produces an iterator over the values such that value satisfies the
+   * predicate, where producing an element removes the value from the state store and producing
+   * all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   */
+  def removeByValueCondition(removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]

Review comment:
       And this change is in #31937.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879208893


   Thanks @HeartSaVioR. Merging to master/3.2.


-- 
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867231688


   > We have 4 open sub-tasks in SPARK-10816
   
   Do you mean open PRs or JIRAs? Based on #31937, I only see three open PRs and two PRs were merged.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867492787


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44783/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867534912


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44786/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809163182


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41215/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878620043


   Kubernetes integration test status success
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45447/
   


-- 
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 #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810697899


   I just simply remedied write amplification a bit via determining existing values to be "deleted" or "replaced", instead of blindly judging it as "deleted".


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822238843






-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879028687


   Kubernetes integration test status success
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45485/
   


-- 
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 change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r668496627



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.

Review comment:
       The return value is just same as sessions.length so not that useful.
   
   I think we'd want to measure the number of rows on "added/updated" vs "deleted" and we have both cases on executing method, which requires another sort of return type. Let's keep this as Unit and change it in session window implementation 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] AmplabJenkins removed a comment on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809086999


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136632/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-812936934


   **[Test build #136887 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/136887/testReport)** for PR 31989 at commit [`1ecfed4`](https://github.com/apache/spark/commit/1ecfed4fc952d0d3cb702a0faf33dbbf40a58040).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868470452


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140313/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 a change in pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #31989:
URL: https://github.com/apache/spark/pull/31989#discussion_r668446915



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   * @param sessions The all sessions including existing sessions if it's active.
+   *                 Existing sessions which aren't included in this parameter will be removed.
+   */
+  def updateSessions(store: StateStore, key: UnsafeRow, sessions: Seq[UnsafeRow]): Long
+
+  /**
+   * Removes using a predicate on values, with returning removed values via iterator.
+   *
+   * At a high level, this produces an iterator over the (key, value, matched) tuples such that
+   * value satisfies the predicate, where producing an element removes the value from the
+   * state store and producing all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   *
+   * @param removalCondition The predicate on removing the key-value.
+   */
+  def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]
+
+  /**
+   * Return an iterator containing all the sessions. Implementations must ensure that updates
+   * (puts, removes) can be made while iterating over this iterator.
+   */
+  def iterator(store: ReadStateStore): Iterator[UnsafeRow]
+
+  /**
+   * Commits the change.
+   */
+  def commit(store: StateStore): Long
+
+  /**
+   * Aborts the change.
+   */
+  def abortIfNeeded(store: StateStore): Unit
+}
+
+object StreamingSessionWindowStateManager {
+  val supportedVersions = Seq(1)
+
+  def createStateManager(
+      keyWithoutSessionExpressions: Seq[Attribute],
+      sessionExpression: Attribute,
+      inputRowAttributes: Seq[Attribute],
+      stateFormatVersion: Int): StreamingSessionWindowStateManager = {
+    stateFormatVersion match {
+      case 1 => new StreamingSessionWindowStateManagerImplV1(
+        keyWithoutSessionExpressions, sessionExpression, inputRowAttributes)
+      case _ => throw new IllegalArgumentException(s"Version $stateFormatVersion is invalid")
+    }
+  }
+}
+
+class StreamingSessionWindowStateManagerImplV1(
+    keyWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    valueAttributes: Seq[Attribute])
+  extends StreamingSessionWindowStateManager with Logging {
+
+  @transient private lazy val keyWithoutSessionRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+
+  private val stateKeyStructType = keyWithoutSessionExpressions.toStructType
+    .add("sessionStartTime", TimestampType, nullable = false)
+
+  private val stateKeyExprs = keyWithoutSessionExpressions :+ Literal(1L)
+  private val indexOrdinalInSessionStart = keyWithoutSessionExpressions.size
+  @transient private lazy val keyRowGenerator = UnsafeProjection.create(

Review comment:
       `keyRowGenerator` looks the same as `keyWithoutSessionRowGenerator`?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   * @param sessions The all sessions including existing sessions if it's active.
+   *                 Existing sessions which aren't included in this parameter will be removed.
+   */
+  def updateSessions(store: StateStore, key: UnsafeRow, sessions: Seq[UnsafeRow]): Long
+
+  /**
+   * Removes using a predicate on values, with returning removed values via iterator.
+   *
+   * At a high level, this produces an iterator over the (key, value, matched) tuples such that
+   * value satisfies the predicate, where producing an element removes the value from the
+   * state store and producing all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   *
+   * @param removalCondition The predicate on removing the key-value.
+   */
+  def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]
+
+  /**
+   * Return an iterator containing all the sessions. Implementations must ensure that updates
+   * (puts, removes) can be made while iterating over this iterator.
+   */
+  def iterator(store: ReadStateStore): Iterator[UnsafeRow]
+
+  /**
+   * Commits the change.
+   */
+  def commit(store: StateStore): Long
+
+  /**
+   * Aborts the change.
+   */
+  def abortIfNeeded(store: StateStore): Unit
+}
+
+object StreamingSessionWindowStateManager {
+  val supportedVersions = Seq(1)
+
+  def createStateManager(
+      keyWithoutSessionExpressions: Seq[Attribute],
+      sessionExpression: Attribute,
+      inputRowAttributes: Seq[Attribute],
+      stateFormatVersion: Int): StreamingSessionWindowStateManager = {
+    stateFormatVersion match {
+      case 1 => new StreamingSessionWindowStateManagerImplV1(
+        keyWithoutSessionExpressions, sessionExpression, inputRowAttributes)
+      case _ => throw new IllegalArgumentException(s"Version $stateFormatVersion is invalid")
+    }
+  }
+}
+
+class StreamingSessionWindowStateManagerImplV1(
+    keyWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    valueAttributes: Seq[Attribute])
+  extends StreamingSessionWindowStateManager with Logging {
+
+  @transient private lazy val keyWithoutSessionRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+
+  private val stateKeyStructType = keyWithoutSessionExpressions.toStructType
+    .add("sessionStartTime", TimestampType, nullable = false)
+
+  private val stateKeyExprs = keyWithoutSessionExpressions :+ Literal(1L)
+  private val indexOrdinalInSessionStart = keyWithoutSessionExpressions.size
+  @transient private lazy val keyRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+  @transient private lazy val stateKeyRowGenerator = UnsafeProjection.create(stateKeyExprs,
+    keyWithoutSessionExpressions)
+
+  @transient private lazy val helper = new StreamingSessionWindowHelper(
+    sessionExpression, valueAttributes)
+
+  override def getStateKeySchema: StructType = stateKeyStructType
+
+  override def getStateValueSchema: StructType = valueAttributes.toStructType
+
+  override def getNumColsForPrefixKey: Int = keyWithoutSessionExpressions.length
+
+  override def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow = {
+    keyWithoutSessionRowGenerator(value)
+  }
+
+  override def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean = {
+    val sessionStart = helper.extractTimePair(value)._1
+    val stateKey = getStateKey(getKey(value), sessionStart)
+    val stateRow = store.get(stateKey)
+    stateRow == null || !stateRow.equals(value)
+  }
+
+  override def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow] =
+    getSessionsWithKeys(store, key).map(_.value)
+
+  private def getSessionsWithKeys(
+      store: ReadStateStore,
+      key: UnsafeRow): Iterator[UnsafeRowPair] = {
+    store.prefixScan(key)
+  }
+
+  override def updateSessions(
+      store: StateStore,
+      key: UnsafeRow,
+      sessions: Seq[UnsafeRow]): Long = {
+    replaceModifyWindow(key, sessions, store)
+  }
+
+  override def commit(store: StateStore): Long = store.commit()
+
+  override def iterator(store: ReadStateStore): Iterator[UnsafeRow] =
+    iteratorWithKeys(store).map(_.value)
+
+  private def iteratorWithKeys(store: ReadStateStore): Iterator[UnsafeRowPair] = {
+    store.iterator()
+  }
+
+  override def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow] = {
+    new NextIterator[UnsafeRow] {
+      private val rangeIter = iteratorWithKeys(store)
+
+      override protected def getNext(): UnsafeRow = {
+        var removedValueRow: UnsafeRow = null
+        while(rangeIter.hasNext && removedValueRow == null) {
+          val rowPair = rangeIter.next()
+          if (removalCondition(rowPair.value)) {
+            store.remove(rowPair.key)
+            removedValueRow = rowPair.value
+          }
+        }
+
+        if (removedValueRow == null) {
+          finished = true
+          null
+        } else {
+          removedValueRow
+        }
+      }
+
+      override protected def close(): Unit = {}
+    }
+  }
+
+  private def getKey(value: UnsafeRow): UnsafeRow = keyRowGenerator(value)
+
+  private def getStateKey(key: UnsafeRow, sessionStart: Long): UnsafeRow = {
+    val stateKey = stateKeyRowGenerator(key)
+    stateKey.setLong(indexOrdinalInSessionStart, sessionStart)
+    stateKey.copy()
+  }
+
+  /**
+   * add to store, only update the changed window elements
+   */
+  private def replaceModifyWindow(
+      key: UnsafeRow,
+      values: Seq[UnsafeRow],
+      store: StateStore): Long = {
+    // Below two will be used multiple times - need to make sure this is not a stream or iterator.
+    val newValues = values.toList
+    val savedStates = getSessionsWithKeys(store, key)
+      .map(pair => (pair.key.copy(), pair.value.copy())).toList
+
+    if (savedStates.isEmpty) {
+      putRows(store, key, savedStates, newValues)
+      newValues.length
+    } else {
+      // get origin state from state store by specified key
+      // filter out the window which have overlapped with new state
+      val newStates = helper.removeOverlapWindow(newValues, savedStates.map(_._2)).toList
+      putRows(store, key, savedStates, newStates)
+      newStates.length
+    }
+  }
+
+  private def putRows(
+      store: StateStore,
+      key: UnsafeRow,
+      oldValues: List[(UnsafeRow, UnsafeRow)],
+      values: List[UnsafeRow]): Unit = {
+    // Here the key doesn't represent the state key - we need to construct the key for state
+    val keyAndValues = values.map { row =>
+      val sessionStart = helper.extractTimePair(row)._1
+      val stateKey = getStateKey(key, sessionStart)
+      (stateKey, row)
+    }
+
+    val keysForValues = keyAndValues.map(_._1)
+    val keysForOldValues = oldValues.map(_._1)
+
+    // We should "replace" the value instead of "delete" and "put" if the start time
+    // equals to. This will remove unnecessary tombstone being written to the delta, which is
+    // implementation details on state store implementations.
+    keysForOldValues.filterNot(keysForValues.contains).foreach { oldKey =>
+      store.remove(oldKey)
+    }
+
+    keyAndValues.foreach { case (key, value) =>
+      store.put(key, value)
+    }
+  }
+
+  override def abortIfNeeded(store: StateStore): Unit = {
+    if (!store.hasCommitted) {
+      logInfo(s"Aborted store ${store.id}")
+      store.abort()
+    }
+  }
+}
+
+class StreamingSessionWindowHelper(sessionExpression: Attribute, inputSchema: Seq[Attribute]) {
+
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  // extract session_window (start, end) from UnsafeRow
+  def extractTimePair(value: InternalRow): (Long, Long) = {
+    val window = sessionProjection(value).getStruct(0, 2)
+    (window.getLong(0), window.getLong(1))
+  }
+
+  def removeOverlapWindow(

Review comment:
       Could you add some comment on this method?

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   * @param sessions The all sessions including existing sessions if it's active.
+   *                 Existing sessions which aren't included in this parameter will be removed.
+   */
+  def updateSessions(store: StateStore, key: UnsafeRow, sessions: Seq[UnsafeRow]): Long
+
+  /**
+   * Removes using a predicate on values, with returning removed values via iterator.
+   *
+   * At a high level, this produces an iterator over the (key, value, matched) tuples such that
+   * value satisfies the predicate, where producing an element removes the value from the
+   * state store and producing all elements with a given key updates it accordingly.
+   *
+   * This implies the iterator must be consumed fully without any other operations on this manager
+   * or the underlying store being interleaved.
+   *
+   * @param removalCondition The predicate on removing the key-value.
+   */
+  def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow]
+
+  /**
+   * Return an iterator containing all the sessions. Implementations must ensure that updates
+   * (puts, removes) can be made while iterating over this iterator.
+   */
+  def iterator(store: ReadStateStore): Iterator[UnsafeRow]
+
+  /**
+   * Commits the change.
+   */
+  def commit(store: StateStore): Long
+
+  /**
+   * Aborts the change.
+   */
+  def abortIfNeeded(store: StateStore): Unit
+}
+
+object StreamingSessionWindowStateManager {
+  val supportedVersions = Seq(1)
+
+  def createStateManager(
+      keyWithoutSessionExpressions: Seq[Attribute],
+      sessionExpression: Attribute,
+      inputRowAttributes: Seq[Attribute],
+      stateFormatVersion: Int): StreamingSessionWindowStateManager = {
+    stateFormatVersion match {
+      case 1 => new StreamingSessionWindowStateManagerImplV1(
+        keyWithoutSessionExpressions, sessionExpression, inputRowAttributes)
+      case _ => throw new IllegalArgumentException(s"Version $stateFormatVersion is invalid")
+    }
+  }
+}
+
+class StreamingSessionWindowStateManagerImplV1(
+    keyWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    valueAttributes: Seq[Attribute])
+  extends StreamingSessionWindowStateManager with Logging {
+
+  @transient private lazy val keyWithoutSessionRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+
+  private val stateKeyStructType = keyWithoutSessionExpressions.toStructType
+    .add("sessionStartTime", TimestampType, nullable = false)
+
+  private val stateKeyExprs = keyWithoutSessionExpressions :+ Literal(1L)
+  private val indexOrdinalInSessionStart = keyWithoutSessionExpressions.size
+  @transient private lazy val keyRowGenerator = UnsafeProjection.create(
+    keyWithoutSessionExpressions, valueAttributes)
+  @transient private lazy val stateKeyRowGenerator = UnsafeProjection.create(stateKeyExprs,
+    keyWithoutSessionExpressions)
+
+  @transient private lazy val helper = new StreamingSessionWindowHelper(
+    sessionExpression, valueAttributes)
+
+  override def getStateKeySchema: StructType = stateKeyStructType
+
+  override def getStateValueSchema: StructType = valueAttributes.toStructType
+
+  override def getNumColsForPrefixKey: Int = keyWithoutSessionExpressions.length
+
+  override def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow = {
+    keyWithoutSessionRowGenerator(value)
+  }
+
+  override def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean = {
+    val sessionStart = helper.extractTimePair(value)._1
+    val stateKey = getStateKey(getKey(value), sessionStart)
+    val stateRow = store.get(stateKey)
+    stateRow == null || !stateRow.equals(value)
+  }
+
+  override def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow] =
+    getSessionsWithKeys(store, key).map(_.value)
+
+  private def getSessionsWithKeys(
+      store: ReadStateStore,
+      key: UnsafeRow): Iterator[UnsafeRowPair] = {
+    store.prefixScan(key)
+  }
+
+  override def updateSessions(
+      store: StateStore,
+      key: UnsafeRow,
+      sessions: Seq[UnsafeRow]): Long = {
+    replaceModifyWindow(key, sessions, store)
+  }
+
+  override def commit(store: StateStore): Long = store.commit()
+
+  override def iterator(store: ReadStateStore): Iterator[UnsafeRow] =
+    iteratorWithKeys(store).map(_.value)
+
+  private def iteratorWithKeys(store: ReadStateStore): Iterator[UnsafeRowPair] = {
+    store.iterator()
+  }
+
+  override def removeByValueCondition(
+      store: StateStore,
+      removalCondition: UnsafeRow => Boolean): Iterator[UnsafeRow] = {
+    new NextIterator[UnsafeRow] {
+      private val rangeIter = iteratorWithKeys(store)
+
+      override protected def getNext(): UnsafeRow = {
+        var removedValueRow: UnsafeRow = null
+        while(rangeIter.hasNext && removedValueRow == null) {

Review comment:
       `while (...)`

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StreamingSessionWindowStateManager.scala
##########
@@ -0,0 +1,370 @@
+/*
+ * 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.state
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Literal, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.types.{StructType, TimestampType}
+import org.apache.spark.util.NextIterator
+
+sealed trait StreamingSessionWindowStateManager extends Serializable {
+  /**
+   * Returns the schema for key of the state.
+   */
+  def getStateKeySchema: StructType
+
+  /**
+   * Returns the schema for value of the state.
+   */
+  def getStateValueSchema: StructType
+
+  /**
+   * Returns the number of columns for `prefix key` in key schema.
+   */
+  def getNumColsForPrefixKey: Int
+
+  /**
+   * Extracts the key without session window from the row.
+   * This can be used to group session windows by key.
+   */
+  def extractKeyWithoutSession(value: UnsafeRow): UnsafeRow
+
+  /**
+   * Check whether the given value is a new session, or the session has been
+   * updated.
+   * This can be used to control the output in UPDATE mode.
+   */
+  def newOrModified(store: ReadStateStore, value: UnsafeRow): Boolean
+
+  /**
+   * Returns all sessions for the key.
+   *
+   * @param key The key without session, which can be retrieved from
+   *            {@code extractKeyWithoutSession}.
+   */
+  def getSessions(store: ReadStateStore, key: UnsafeRow): Iterator[UnsafeRow]
+
+  /**
+   * Replaces all sessions for the key to given one.

Review comment:
       What the returned long is?




-- 
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878637665






-- 
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867687888


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140257/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810814443


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136742/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-822220644


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/42130/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867540984


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44786/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-868330591


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44845/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878662960


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45449/
   


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879008298


   Kubernetes integration test starting
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45485/
   


-- 
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867492756


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44783/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] xuanyuanking commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-823135370


   @viirya Sorry for the late. Just back from the work on open-sourcing RocksDBStateStore. Since the design has been published. I can back to this now.
   I'm +1 for the `StreamingSessionWindowStateManager` abstraction. Only have some concern about the StreamingSessionWindowStateManagerImplV1. It would be great if we can have a better state store API to reduce the complexity of the implementation. I'll start working on it in these two weeks (in parallel with open-sourcing the RocksDB state store).


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878705167


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140935/
   


-- 
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-879189149


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140971/
   


-- 
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] AmplabJenkins removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811665311


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41375/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-811078011


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136756/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] SparkQA commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878581202


   **[Test build #140936 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140936/testReport)** for PR 31989 at commit [`8638baf`](https://github.com/apache/spark/commit/8638baf0066b9b49983f525b9f0db230aca4201c).
    * This patch **fails to build**.
    * This patch merges cleanly.
    * This patch adds no public classes.


-- 
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] SparkQA removed a comment on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878609113


   **[Test build #140937 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140937/testReport)** for PR 31989 at commit [`8638baf`](https://github.com/apache/spark/commit/8638baf0066b9b49983f525b9f0db230aca4201c).


-- 
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] SparkQA commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810715929


   Kubernetes integration test status failure
   URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/41324/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 closed pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
viirya closed pull request #31989:
URL: https://github.com/apache/spark/pull/31989


   


-- 
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] AmplabJenkins commented on pull request #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-878581254


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140936/
   


-- 
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] AmplabJenkins commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809086999


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/136632/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-809067233


   Except the test suite, one more thing worths to address here is write amplification; we "blindly" replace all start times and all sessions. This could bring unnecessary writes on "unmodified" existing sessions. In many cases we expect the new inputs will be bound and expanding to the existing sessions, but with very long watermark gap and old inputs which have various timestamps, the case could still happen.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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] AmplabJenkins commented on pull request #31989: [WIP][SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-810715946


   
   Refer to this link for build results (access rights to CI server needed): 
   https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/41324/
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
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 #31989: [SPARK-34891][SS] Introduce state store manager for session window in streaming query

Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #31989:
URL: https://github.com/apache/spark/pull/31989#issuecomment-867443816


   UPDATE: This is now on top of SPARK-35861 (#33038). I'll rebase this once we merge #33038.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org