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/06/25 05:49:08 UTC

[GitHub] [spark] HeartSaVioR opened a new pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

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

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 MergingSortWithSessionWindowStateIterator, which does "merge sort" between input rows and sessions in state based on group key and session's start time.

Note that the iterator does merge sort among input rows and sessions grouped by grouping key. The iterator doesn't provide sessions in state which keys don't exist in input rows. For input rows, the iterator will provide all rows regardless of the existence of matching sessions in state.

MergingSortWithSessionWindowStateIterator works on the precondition that given iterator is sorted by "group keys + start time of session window", and the iterator still retains the characteristic of the sort.

### 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?

New UT added.

--
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879512560


   **[Test build #140989 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140989/testReport)** for PR 33077 at commit [`e4a74a3`](https://github.com/apache/spark/commit/e4a74a3375ff0de4ac72aa951d36aa36f2492e01).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879755143


   Thanks for reviewing! I merged this to master/3.2. I'll rebase #33081 to reflect the latest master branch.


-- 
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 a change in pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

viirya commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669320305



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIteratorSuite.scala
##########
@@ -0,0 +1,231 @@
+/*
+ * 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
+
+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.state.{HDFSBackedStateStoreProvider, RocksDBStateStoreProvider, StateStore, StateStoreConf, StateStoreId, StateStoreProviderId, StreamingSessionWindowStateManager}
+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 MergingSortWithSessionWindowStateIteratorSuite 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)
+    spark.streams.stateStoreCoordinator // initialize the lazy coordinator
+  }
+
+  private val providerOptions = Seq(
+    classOf[HDFSBackedStateStoreProvider].getCanonicalName,
+    classOf[RocksDBStateStoreProvider].getCanonicalName).map { value =>
+    (SQLConf.STATE_STORE_PROVIDER_CLASS.key, value.stripSuffix("$"))
+  }
+
+  private val availableOptions = for (
+    opt1 <- providerOptions;
+    opt2 <- StreamingSessionWindowStateManager.supportedVersions
+  ) yield (opt1, opt2)
+
+  availableOptions.foreach { case (providerOpt, version) =>
+    withSQLConf(providerOpt) {
+      test(s"StreamingSessionWindowStateManager " +
+        s"provider ${providerOpt._2} state version v${version} - rows only in state") {
+        testRowsOnlyInState(version)
+      }
+
+      test(s"StreamingSessionWindowStateManager " +
+        s"provider ${providerOpt._2} state version v${version} - rows in both input and state") {
+        testRowsInBothInputAndState(version)
+      }
+
+      test(s"StreamingSessionWindowStateManager " +
+        s"provider ${providerOpt._2} state version v${version} - rows only in input") {
+        testRowsOnlyInInput(version)
+      }
+    }
+  }
+
+  private def testRowsOnlyInState(stateFormatVersion: Int): Unit = {
+    withStateManager(stateFormatVersion) { case (stateManager, store) =>
+      val key = createKeyRow("a", 1)
+      val values = Seq(
+        createRow("a", 1, 100, 110, 1),
+        createRow("a", 1, 120, 130, 2),
+        createRow("a", 1, 140, 150, 3))
+
+      stateManager.updateSessions(store, key, values)
+
+      val iter = new MergingSortWithSessionWindowStateIterator(
+        Iterator.empty,
+        stateManager,
+        store,
+        keysWithoutSessionAttributes,
+        sessionAttribute,
+        rowAttributes)
+
+      val actual = iter.map(_.copy()).toList
+      assert(actual.isEmpty)

Review comment:
       Oh, nvm. I see.




-- 
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 a change in pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

viirya commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669315813



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _

Review comment:
       A few suggestions:
   
   currentRow -> currentRowFromInput
   currentStateRow -> currentRowFromState
   currentStateIter -> sessionIterFromState
   currentStateFetchedKey -> currentSessionKey

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _
+
+  override def hasNext: Boolean = {
+    currentRow != null || currentStateRow != null ||
+      (currentStateIter != null && currentStateIter.hasNext) || iter.hasNext
+  }
+
+  override def next(): InternalRow = {
+    if (currentRow == null) {
+      mayFillCurrentRow()
+    }
+
+    if (currentStateRow == null) {
+      mayFillCurrentStateRow()
+    }
+
+    if (currentRow == null && currentStateRow == null) {
+      throw new IllegalStateException("No Row to provide in next() which should not happen!")
+    }
+
+    // return current row vs current state row, should return smaller key, earlier session start
+    val returnCurrentRow: Boolean = {
+      if (currentRow == null) {
+        false
+      } else if (currentStateRow == null) {
+        true
+      } else {
+        // compare
+        if (currentRow.keys != currentStateRow.keys) {
+          // state row cannot advance to row in input, so state row should be lower

Review comment:
       Does this case mean, the input iterator advances to new keys other than current sessions from the state? So we should output from current sessions until it ends and retrieves new sessions from the state again? 




-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879367696


   cc. @viirya @xuanyuanking 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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879536574


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879391296


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879665131


   **[Test build #141006 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141006/testReport)** for PR 33077 at commit [`c674046`](https://github.com/apache/spark/commit/c674046b78d0b590ba78df4812d507d4833b9b1a).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868291726


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868359697


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879386639


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868470456


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879577172


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879607967


   **[Test build #140989 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140989/testReport)** for PR 33077 at commit [`e4a74a3`](https://github.com/apache/spark/commit/e4a74a3375ff0de4ac72aa951d36aa36f2492e01).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds the following public classes _(experimental)_:
     * `class MergingSortWithSessionWindowStateIterator(`


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879665131


   **[Test build #141006 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141006/testReport)** for PR 33077 at commit [`c674046`](https://github.com/apache/spark/commit/c674046b78d0b590ba78df4812d507d4833b9b1a).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879564143


   **[Test build #140987 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140987/testReport)** for PR 33077 at commit [`b540632`](https://github.com/apache/spark/commit/b540632e1180dcb1ce9f49626a9fb925fd503742).
    * 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] AmplabJenkins removed a comment on pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879742707


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868223797


   **[Test build #140306 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140306/testReport)** for PR 33077 at commit [`eb433d7`](https://github.com/apache/spark/commit/eb433d78ba352927e4a5ec0c3681abafcbeee148).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669340938



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _
+
+  override def hasNext: Boolean = {
+    currentRow != null || currentStateRow != null ||
+      (currentStateIter != null && currentStateIter.hasNext) || iter.hasNext
+  }
+
+  override def next(): InternalRow = {
+    if (currentRow == null) {
+      mayFillCurrentRow()
+    }
+
+    if (currentStateRow == null) {
+      mayFillCurrentStateRow()
+    }
+
+    if (currentRow == null && currentStateRow == null) {
+      throw new IllegalStateException("No Row to provide in next() which should not happen!")
+    }
+
+    // return current row vs current state row, should return smaller key, earlier session start
+    val returnCurrentRow: Boolean = {
+      if (currentRow == null) {
+        false
+      } else if (currentStateRow == null) {
+        true
+      } else {
+        // compare
+        if (currentRow.keys != currentStateRow.keys) {
+          // state row cannot advance to row in input, so state row should be lower

Review comment:
       Exactly. We retrieve state rows for the key only when there's a new key from input side, so the case is not possible state side advances compared to input side. If the keys differ, either state side also doesn't have any rows to process, or there're rows to process in state side.




-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868263221


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868295790


   **[Test build #140311 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140311/testReport)** for PR 33077 at commit [`83cc78a`](https://github.com/apache/spark/commit/83cc78ab42b1b776ba2a4978f8b971e8fe3be282).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879529600


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


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

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

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



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

[GitHub] [spark] HeartSaVioR closed pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR closed pull request #33077:
URL: https://github.com/apache/spark/pull/33077


   


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879381320


   **[Test build #140981 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140981/testReport)** for PR 33077 at commit [`60f6114`](https://github.com/apache/spark/commit/60f6114285b10ddae07b95fbc4ceb2e03a698173).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879471127


   **[Test build #140987 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140987/testReport)** for PR 33077 at commit [`b540632`](https://github.com/apache/spark/commit/b540632e1180dcb1ce9f49626a9fb925fd503742).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868222682


   Same here; I marked this as draft as other PRs has to be reviewed and merged earlier. I'll rebase this PR once all other PRs are 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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868295790


   **[Test build #140311 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140311/testReport)** for PR 33077 at commit [`83cc78a`](https://github.com/apache/spark/commit/83cc78ab42b1b776ba2a4978f8b971e8fe3be282).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879608822


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879753466


   Please bear with me on merging this one now to unblock the last PR #33081. We can do post-review for any part of session window changes even it's merged during QA period (and even RC period if someone catches defect.) 


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879386639


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879825000


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669340938



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _
+
+  override def hasNext: Boolean = {
+    currentRow != null || currentStateRow != null ||
+      (currentStateIter != null && currentStateIter.hasNext) || iter.hasNext
+  }
+
+  override def next(): InternalRow = {
+    if (currentRow == null) {
+      mayFillCurrentRow()
+    }
+
+    if (currentStateRow == null) {
+      mayFillCurrentStateRow()
+    }
+
+    if (currentRow == null && currentStateRow == null) {
+      throw new IllegalStateException("No Row to provide in next() which should not happen!")
+    }
+
+    // return current row vs current state row, should return smaller key, earlier session start
+    val returnCurrentRow: Boolean = {
+      if (currentRow == null) {
+        false
+      } else if (currentStateRow == null) {
+        true
+      } else {
+        // compare
+        if (currentRow.keys != currentStateRow.keys) {
+          // state row cannot advance to row in input, so state row should be lower

Review comment:
       Exactly. We retrieve state rows for the key only when there's a new key from input side, so the case is not possible state side advances compared to input side. If the keys differ, either state side also doesn't have any rows to process, or there're rows to process in state side. The opposite case cannot be possible.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _
+
+  override def hasNext: Boolean = {
+    currentRow != null || currentStateRow != null ||
+      (currentStateIter != null && currentStateIter.hasNext) || iter.hasNext
+  }
+
+  override def next(): InternalRow = {
+    if (currentRow == null) {
+      mayFillCurrentRow()
+    }
+
+    if (currentStateRow == null) {
+      mayFillCurrentStateRow()
+    }
+
+    if (currentRow == null && currentStateRow == null) {
+      throw new IllegalStateException("No Row to provide in next() which should not happen!")
+    }
+
+    // return current row vs current state row, should return smaller key, earlier session start
+    val returnCurrentRow: Boolean = {
+      if (currentRow == null) {
+        false
+      } else if (currentStateRow == null) {
+        true
+      } else {
+        // compare
+        if (currentRow.keys != currentStateRow.keys) {
+          // state row cannot advance to row in input, so state row should be lower

Review comment:
       Exactly. We retrieve state rows for the key only when there's a new key from input side, so the case is not possible state side advances compared to input side. If the keys differ, either state side also doesn't have any rows to process, or there're rows to process in state side. The opposite case is not possible.




-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879381320


   **[Test build #140981 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140981/testReport)** for PR 33077 at commit [`60f6114`](https://github.com/apache/spark/commit/60f6114285b10ddae07b95fbc4ceb2e03a698173).


-- 
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 a change in pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

viirya commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669319633



##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIteratorSuite.scala
##########
@@ -0,0 +1,231 @@
+/*
+ * 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
+
+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.state.{HDFSBackedStateStoreProvider, RocksDBStateStoreProvider, StateStore, StateStoreConf, StateStoreId, StateStoreProviderId, StreamingSessionWindowStateManager}
+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 MergingSortWithSessionWindowStateIteratorSuite 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)
+    spark.streams.stateStoreCoordinator // initialize the lazy coordinator
+  }
+
+  private val providerOptions = Seq(
+    classOf[HDFSBackedStateStoreProvider].getCanonicalName,
+    classOf[RocksDBStateStoreProvider].getCanonicalName).map { value =>
+    (SQLConf.STATE_STORE_PROVIDER_CLASS.key, value.stripSuffix("$"))
+  }
+
+  private val availableOptions = for (
+    opt1 <- providerOptions;
+    opt2 <- StreamingSessionWindowStateManager.supportedVersions
+  ) yield (opt1, opt2)
+
+  availableOptions.foreach { case (providerOpt, version) =>
+    withSQLConf(providerOpt) {
+      test(s"StreamingSessionWindowStateManager " +
+        s"provider ${providerOpt._2} state version v${version} - rows only in state") {
+        testRowsOnlyInState(version)
+      }
+
+      test(s"StreamingSessionWindowStateManager " +
+        s"provider ${providerOpt._2} state version v${version} - rows in both input and state") {
+        testRowsInBothInputAndState(version)
+      }
+
+      test(s"StreamingSessionWindowStateManager " +
+        s"provider ${providerOpt._2} state version v${version} - rows only in input") {
+        testRowsOnlyInInput(version)
+      }
+    }
+  }
+
+  private def testRowsOnlyInState(stateFormatVersion: Int): Unit = {
+    withStateManager(stateFormatVersion) { case (stateManager, store) =>
+      val key = createKeyRow("a", 1)
+      val values = Seq(
+        createRow("a", 1, 100, 110, 1),
+        createRow("a", 1, 120, 130, 2),
+        createRow("a", 1, 140, 150, 3))
+
+      stateManager.updateSessions(store, key, values)
+
+      val iter = new MergingSortWithSessionWindowStateIterator(
+        Iterator.empty,
+        stateManager,
+        store,
+        keysWithoutSessionAttributes,
+        sessionAttribute,
+        rowAttributes)
+
+      val actual = iter.map(_.copy()).toList
+      assert(actual.isEmpty)

Review comment:
       Hm? Why it is empty? If input is empty, doesn't it output sorted sessions in the state?




-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879490397


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879742707


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR edited a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879755143


   Thanks @viirya for reviewing! I merged this to master/3.2. I'll rebase #33081 to reflect the latest master branch.


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879386603


   **[Test build #140981 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140981/testReport)** for PR 33077 at commit [`60f6114`](https://github.com/apache/spark/commit/60f6114285b10ddae07b95fbc4ceb2e03a698173).
    * 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] AmplabJenkins commented on pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879536574


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868328320


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868372620


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879577172


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879502187


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879825000


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879720094


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


-- 
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 a change in pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

viirya commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669321330



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _

Review comment:
       And maybe add a few comments explaining these variables for readability.




-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879512560


   **[Test build #140989 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140989/testReport)** for PR 33077 at commit [`e4a74a3`](https://github.com/apache/spark/commit/e4a74a3375ff0de4ac72aa951d36aa36f2492e01).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868437499


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879823869


   **[Test build #141006 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141006/testReport)** for PR 33077 at commit [`c674046`](https://github.com/apache/spark/commit/c674046b78d0b590ba78df4812d507d4833b9b1a).
    * 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 commented on pull request #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879471127


   **[Test build #140987 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140987/testReport)** for PR 33077 at commit [`b540632`](https://github.com/apache/spark/commit/b540632e1180dcb1ce9f49626a9fb925fd503742).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

viirya commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879666657


   I will merge this tomorrow. Thanks @HeartSaVioR 


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868420671


   **[Test build #140306 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140306/testReport)** for PR 33077 at commit [`eb433d7`](https://github.com/apache/spark/commit/eb433d78ba352927e4a5ec0c3681abafcbeee148).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds the following public classes _(experimental)_:
     * `class MergingSortWithSessionWindowStateIterator(`


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

viirya commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879368499


   Thanks @HeartSaVioR. I will review this today.


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669340938



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _
+
+  override def hasNext: Boolean = {
+    currentRow != null || currentStateRow != null ||
+      (currentStateIter != null && currentStateIter.hasNext) || iter.hasNext
+  }
+
+  override def next(): InternalRow = {
+    if (currentRow == null) {
+      mayFillCurrentRow()
+    }
+
+    if (currentStateRow == null) {
+      mayFillCurrentStateRow()
+    }
+
+    if (currentRow == null && currentStateRow == null) {
+      throw new IllegalStateException("No Row to provide in next() which should not happen!")
+    }
+
+    // return current row vs current state row, should return smaller key, earlier session start
+    val returnCurrentRow: Boolean = {
+      if (currentRow == null) {
+        false
+      } else if (currentStateRow == null) {
+        true
+      } else {
+        // compare
+        if (currentRow.keys != currentStateRow.keys) {
+          // state row cannot advance to row in input, so state row should be lower

Review comment:
       Exactly. We retrieve state rows for the specific key only when there's a new key from input side, so the case is not possible state side advances compared to input side. If the keys differ, there're rows to process in state side. The opposite case is not possible.




-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879511382


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868470456


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

HeartSaVioR commented on a change in pull request #33077:
URL: https://github.com/apache/spark/pull/33077#discussion_r669340938



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MergingSortWithSessionWindowStateIterator.scala
##########
@@ -0,0 +1,161 @@
+/*
+ * 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
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.streaming.state.{ReadStateStore, StreamingSessionWindowStateManager}
+
+/**
+ * This class technically does the merge sort between input rows and existing sessions in state,
+ * to optimize the cost of sort on "input rows + existing sessions". This is based on the
+ * precondition that input rows are sorted by "group keys + start time of session window".
+ *
+ * This only materializes the existing sessions into memory, which are tend to be not many per
+ * group key. The cost of sorting existing sessions would be also minor based on the assumption.
+ *
+ * The output rows are sorted with "group keys + start time of session window", which is same as
+ * the sort condition on input rows.
+ */
+class MergingSortWithSessionWindowStateIterator(
+    iter: Iterator[InternalRow],
+    stateManager: StreamingSessionWindowStateManager,
+    store: ReadStateStore,
+    groupWithoutSessionExpressions: Seq[Attribute],
+    sessionExpression: Attribute,
+    inputSchema: Seq[Attribute]) extends Iterator[InternalRow] with Logging {
+
+  private val keysProjection: UnsafeProjection = GenerateUnsafeProjection.generate(
+    groupWithoutSessionExpressions, inputSchema)
+  private val sessionProjection: UnsafeProjection =
+    GenerateUnsafeProjection.generate(Seq(sessionExpression), inputSchema)
+
+  private case class SessionRowInformation(
+      keys: UnsafeRow,
+      sessionStart: Long,
+      sessionEnd: Long,
+      row: InternalRow)
+
+  private object SessionRowInformation {
+    def of(row: InternalRow): SessionRowInformation = {
+      val keys = keysProjection(row).copy()
+      val session = sessionProjection(row).copy()
+      val sessionRow = session.getStruct(0, 2)
+      val sessionStart = sessionRow.getLong(0)
+      val sessionEnd = sessionRow.getLong(1)
+
+      SessionRowInformation(keys, sessionStart, sessionEnd, row)
+    }
+  }
+
+  private var currentRow: SessionRowInformation = _
+  private var currentStateRow: SessionRowInformation = _
+  private var currentStateIter: Iterator[InternalRow] = _
+  private var currentStateFetchedKey: UnsafeRow = _
+
+  override def hasNext: Boolean = {
+    currentRow != null || currentStateRow != null ||
+      (currentStateIter != null && currentStateIter.hasNext) || iter.hasNext
+  }
+
+  override def next(): InternalRow = {
+    if (currentRow == null) {
+      mayFillCurrentRow()
+    }
+
+    if (currentStateRow == null) {
+      mayFillCurrentStateRow()
+    }
+
+    if (currentRow == null && currentStateRow == null) {
+      throw new IllegalStateException("No Row to provide in next() which should not happen!")
+    }
+
+    // return current row vs current state row, should return smaller key, earlier session start
+    val returnCurrentRow: Boolean = {
+      if (currentRow == null) {
+        false
+      } else if (currentStateRow == null) {
+        true
+      } else {
+        // compare
+        if (currentRow.keys != currentStateRow.keys) {
+          // state row cannot advance to row in input, so state row should be lower

Review comment:
       Exactly. We retrieve state rows for the specific key only when there's a new key from input side, so the case is not possible state side advances compared to input side. If the keys differ, either state side also doesn't have any rows to process, or there're rows to process in state side. The opposite case is not possible.




-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868291726


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879391296


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868372620


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879511382


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868223797


   **[Test build #140306 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140306/testReport)** for PR 33077 at commit [`eb433d7`](https://github.com/apache/spark/commit/eb433d78ba352927e4a5ec0c3681abafcbeee148).


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins removed a comment on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868437499


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868457990


   **[Test build #140311 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140311/testReport)** for PR 33077 at commit [`83cc78a`](https://github.com/apache/spark/commit/83cc78ab42b1b776ba2a4978f8b971e8fe3be282).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds the following public classes _(experimental)_:
     * `class MergingSortWithSessionWindowStateIterator(`


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879697927


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

AmplabJenkins commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879608822


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-879391268


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


-- 
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 #33077: [SPARK-34892][SS] Introduce MergingSortWithSessionWindowStateIterator sorting input rows and rows in state efficiently

Posted by GitBox <gi...@apache.org>.

SparkQA commented on pull request #33077:
URL: https://github.com/apache/spark/pull/33077#issuecomment-868279983


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


-- 
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