Encoder ttl poc

core/pom.xml

@@ -354,7 +354,18 @@
     </dependency>
     <dependency>
       <groupId>org.seleniumhq.selenium</groupId>
-      <artifactId>htmlunit3-driver</artifactId>
+      <artifactId>htmlunit-driver</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <!-- Coerce sbt into honoring these dependency updates: -->
+    <dependency>
+      <groupId>net.sourceforge.htmlunit</groupId>
+      <artifactId>htmlunit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>net.sourceforge.htmlunit</groupId>
+      <artifactId>htmlunit-core-js</artifactId>
       <scope>test</scope>
     </dependency>
     <!-- at least just for tests, coerce SBT to use the updated httpcore/client version -->
@@ -373,6 +384,12 @@
       <artifactId>httpcore</artifactId>
       <scope>test</scope>
     </dependency>
+    <!-- Added for selenium: -->
+    <dependency>
+      <groupId>xml-apis</groupId>
+      <artifactId>xml-apis</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>

core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala

@@ -561,9 +561,7 @@ abstract class HistoryServerSuite extends SparkFunSuite with BeforeAndAfter with
     // app is no longer incomplete
     listApplications(false) should not contain(appId)
 
-    eventually(stdTimeout, stdInterval) {
-      assert(jobcount === getNumJobs("/jobs"))
-    }
+    assert(jobcount === getNumJobs("/jobs"))
 
     // no need to retain the test dir now the tests complete
     ShutdownHookManager.registerShutdownDeleteDir(logDir)

core/src/test/scala/org/apache/spark/deploy/history/RealBrowserUIHistoryServerSuite.scala

@@ -22,10 +22,8 @@ import javax.servlet.http.HttpServletRequest
 import org.eclipse.jetty.proxy.ProxyServlet
 import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
 import org.openqa.selenium.WebDriver
-import org.scalatest.concurrent.Eventually._
 import org.scalatest.matchers.must.Matchers
 import org.scalatest.matchers.should.Matchers._
-import org.scalatest.time.SpanSugar._
 import org.scalatestplus.selenium.WebBrowser
 
 import org.apache.spark._
@@ -149,9 +147,7 @@ abstract class RealBrowserUIHistoryServerSuite(val driverProp: String)
 
       // there are at least some URL links that were generated via javascript,
       // and they all contain the spark.ui.proxyBase (uiRoot)
-      eventually(timeout(10.seconds)) {
-        links.length should be > 4
-      }
+      links.length should be > 4
       for (link <- links) {
         link should startWith(url + uiRoot)
       }

core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala

@@ -24,9 +24,9 @@ import javax.servlet.http.{HttpServletRequest, HttpServletResponse}
 import scala.io.Source
 import scala.xml.Node
 
+import com.gargoylesoftware.css.parser.CSSParseException
+import com.gargoylesoftware.htmlunit.DefaultCssErrorHandler
 import org.glassfish.jersey.internal.util.collection.MultivaluedStringMap
-import org.htmlunit.DefaultCssErrorHandler
-import org.htmlunit.cssparser.parser.CSSParseException
 import org.json4s._
 import org.json4s.jackson.JsonMethods
 import org.openqa.selenium.{By, WebDriver}

pom.xml

@@ -215,7 +215,8 @@
     <antlr4.version>4.13.1</antlr4.version>
     <jpam.version>1.1</jpam.version>
     <selenium.version>4.12.1</selenium.version>
-    <htmlunit3-driver.version>4.17.0</htmlunit3-driver.version>
+    <htmlunit-driver.version>4.12.0</htmlunit-driver.version>
+    <htmlunit.version>2.70.0</htmlunit.version>
     <maven-antrun.version>3.1.0</maven-antrun.version>
     <commons-crypto.version>1.1.0</commons-crypto.version>
     <commons-cli.version>1.6.0</commons-cli.version>
@@ -709,10 +710,29 @@
       </dependency>
       <dependency>
         <groupId>org.seleniumhq.selenium</groupId>
-        <artifactId>htmlunit3-driver</artifactId>
-        <version>${htmlunit3-driver.version}</version>
+        <artifactId>htmlunit-driver</artifactId>
+        <version>${htmlunit-driver.version}</version>
         <scope>test</scope>
       </dependency>
+      <!-- Update htmlunit dependency that selenium uses for better JS support -->
+      <dependency>
+        <groupId>net.sourceforge.htmlunit</groupId>
+        <artifactId>htmlunit</artifactId>
+        <version>${htmlunit.version}</version>
+        <scope>test</scope>
+      </dependency>
+      <dependency>
+        <groupId>net.sourceforge.htmlunit</groupId>
+        <artifactId>htmlunit-core-js</artifactId>
+        <version>${htmlunit.version}</version>
+        <scope>test</scope>
+      </dependency>
+      <!-- Added for selenium only, and should match its dependent version: -->
+      <dependency>
+        <groupId>xml-apis</groupId>
+        <artifactId>xml-apis</artifactId>
+        <version>1.4.01</version>
+      </dependency>
 
       <!-- log4j -->
       <dependency>

sql/api/src/main/java/org/apache/spark/sql/streaming/TTLMode.java

@@ -0,0 +1,41 @@
+/*
+ * 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.streaming;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.annotation.Experimental;
+import org.apache.spark.sql.catalyst.plans.logical.*;
+
+/**
+ * Represents the type of timeouts possible for the Dataset operations
+ * {@code transformWithState}.
+ */
+@Experimental
+@Evolving
+public class TTLMode {
+    /**
+     * Stateful processor that does not register timers
+     */
+    public static final TTLMode NoTTL() {
+        return NoTTL$.MODULE$;
+    }
+
+    public static final TTLMode ProcessingTimeTTL() {
+        return ProcessingTimeTTL$.MODULE$;
+    }
+}

sql/api/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/TransformWithStateTTLMode.scala

@@ -0,0 +1,24 @@
+/*
+ * 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.catalyst.plans.logical
+
+import org.apache.spark.sql.streaming.TTLMode
+
+/** Types of timeouts used in tranformWithState operator */
+case object NoTTL extends TTLMode
+
+case object ProcessingTimeTTL extends TTLMode

sql/api/src/main/scala/org/apache/spark/sql/streaming/StatefulProcessorHandle.scala

@@ -18,6 +18,8 @@ package org.apache.spark.sql.streaming
 
 import java.io.Serializable
 
+import scala.concurrent.duration.Duration
+
 import org.apache.spark.annotation.{Evolving, Experimental}
 
 /**
@@ -38,6 +40,9 @@ private[sql] trait StatefulProcessorHandle extends Serializable {
    */
   def getValueState[T](stateName: String): ValueState[T]
 
+  def getValueState[T](stateName: String, ttlMode: TTLMode,
+    ttl: Duration): ValueState[T]
+
   /** Function to return queryInfo for currently running task */
   def getQueryInfo(): QueryInfo
 

sql/core/pom.xml

@@ -225,7 +225,7 @@
     </dependency>
     <dependency>
       <groupId>org.seleniumhq.selenium</groupId>
-      <artifactId>htmlunit3-driver</artifactId>
+      <artifactId>htmlunit-driver</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StatefulProcessorHandleImpl.scala

@@ -18,11 +18,13 @@ package org.apache.spark.sql.execution.streaming
 
 import java.util.UUID
 
+import scala.concurrent.duration.Duration
+
 import org.apache.spark.TaskContext
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.execution.streaming.state.StateStore
-import org.apache.spark.sql.streaming.{QueryInfo, StatefulProcessorHandle, ValueState}
+import org.apache.spark.sql.streaming.{QueryInfo, StatefulProcessorHandle, TTLMode, ValueState}
 import org.apache.spark.util.Utils
 
 /**
@@ -119,6 +121,14 @@ class StatefulProcessorHandleImpl(
     resultState
   }
 
+  override def getValueState[T](stateName: String, ttlMode: TTLMode,
+      ttl: Duration): ValueState[T] = {
+    verify(currState == CREATED, s"Cannot create state variable with name=$stateName after " +
+      "initialization is complete")
+    val resultState = new ValueStateImpl[T](store, stateName, keyEncoder, ttlMode, ttl)
+    resultState
+  }
+
   override def getQueryInfo(): QueryInfo = currQueryInfo
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TransformWithStateExec.scala

@@ -80,6 +80,7 @@ case class TransformWithStateExec(
   protected val schemaForKeyRow: StructType = new StructType().add("key", BinaryType)
 
   protected val schemaForValueRow: StructType = new StructType().add("value", BinaryType)
+    .add("ttl", BinaryType)
 
   override def requiredChildDistribution: Seq[Distribution] = {
     StatefulOperatorPartitioning.getCompatibleDistribution(groupingAttributes,