Copilot commented on code in PR #7958:
URL: https://github.com/apache/texera/pull/7958#discussion_r3850162309
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/asterixdb/AsterixDBSourceOpExecSpec.scala:
##########
@@ -573,6 +588,20 @@ class AsterixDBSourceOpExecSpec
tableNameRows = Seq("\"other\"\n", "\"twitter\"\n")
exec.open()
exec.tableNames.toList shouldBe List("other", "twitter")
+ }
+
+ it should "refresh a stale cached API version when it opens" in {
+ val exec = newExec()
+ // Asserting the mapping straight after open() would be vacuous: the
+ // constructor's own `schema = desc.sourceSchema()` already went through
+ // queryAsterixDB, which fills a MISSING host entry, so `0.9.9` is cached
+ // before open() is ever entered. Overwriting the entry first makes
open()'s
+ // explicit refresh the only thing that can restore it - queryAsterixDB
+ // never rewrites an entry that is already present, so a stale version
(which
+ // selects a different `format` field) would otherwise survive forever in
+ // this singleton.
+ AsterixDBConnUtil.asterixDBVersionMapping += (host -> "0.0.0")
+ exec.open()
AsterixDBConnUtil.asterixDBVersionMapping.get(host) shouldBe Some("0.9.9")
Review Comment:
This test mutates the global singleton
`AsterixDBConnUtil.asterixDBVersionMapping` but doesn’t restore it if
`exec.open()` throws, which can leak state into later tests and make failures
order-dependent. Capture the previous value and restore/remove it in a
`finally` block for isolation.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/asterixdb/AsterixDBSourceOpDescSpec.scala:
##########
@@ -19,18 +19,140 @@
package org.apache.texera.amber.operator.source.sql.asterixdb
+import com.sun.net.httpserver.{HttpExchange, HttpServer}
import org.apache.texera.amber.core.executor.OpExecWithClassName
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema}
import org.apache.texera.amber.core.workflow.WorkflowContext.{
DEFAULT_EXECUTION_ID,
DEFAULT_WORKFLOW_ID
}
import org.apache.texera.amber.operator.LogicalOp
import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.BeforeAndAfterAll
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
-class AsterixDBSourceOpDescSpec extends AnyFlatSpec with Matchers {
+import java.net.{InetSocketAddress, URLDecoder}
+import java.nio.charset.StandardCharsets
+import scala.collection.mutable
+import scala.util.Try
+
+class AsterixDBSourceOpDescSpec extends AnyFlatSpec with Matchers with
BeforeAndAfterAll {
+
+ //
---------------------------------------------------------------------------
+ // In-process AsterixDB stub
+ //
+ // sourceSchema() resolves the dataset's datatype over HTTP through
+ // AsterixDBConnUtil, so the type-mapping tests need a reachable server. The
+ // stub answers the two metadata statements sourceSchema() issues and nothing
+ // else; the same approach is used by AsterixDBConnUtilSpec and
+ // AsterixDBSourceOpExecSpec. Binding port 0 keeps it off any fixed port.
+ //
---------------------------------------------------------------------------
+
+ /** Field name -> AsterixDB type the stub reports for the dataset's
datatype. */
+ @volatile private var datatypeFields: Seq[(String, String)] = Seq.empty
+
+ /** Decoded `statement` form field of every /query/service request, in
order. */
+ private val recordedStatements = mutable.Buffer[String]()
+
+ private val server: HttpServer = HttpServer.create(new InetSocketAddress(0),
0)
+ server.createContext(
+ "/admin/version",
+ (exchange: HttpExchange) => respond(exchange,
"""{"git.build.version":"0.9.9"}""")
+ )
+ server.createContext(
+ "/query/service",
+ (exchange: HttpExchange) => {
+ val is = exchange.getRequestBody
+ val body =
+ try new String(is.readAllBytes(), StandardCharsets.UTF_8)
+ finally is.close()
+ val statement = formField(body, "statement")
+ recordedStatements.synchronized { recordedStatements += statement }
+ respond(exchange, responseFor(statement))
+ }
+ )
+
+ private val host = "localhost"
+ private def port: String = server.getAddress.getPort.toString
+
+ private def responseFor(statement: String): String =
+ if (statement.contains("Metadata.`Datatype`")) {
+ val fields = datatypeFields
+ .map { case (name, tpe) =>
s"""{"FieldName":"$name","FieldType":"$tpe"}""" }
+ .mkString(",")
+ s"""{"results":[{"Fields":[$fields]}]}"""
+ } else if (statement.contains("Metadata.`Dataset`")) {
+ """{"results":[{"DatatypeName":"tweetType"}]}"""
+ } else {
+ // Deliberately not a fall-through onto the dataset answer: a statement
+ // aimed at the wrong metadata table must come back empty, so a query the
+ // descriptor mistargets cannot still yield a plausible schema.
+ """{"results":[]}"""
+ }
+
+ private def respond(exchange: HttpExchange, body: String): Unit = {
+ val bytes = body.getBytes(StandardCharsets.UTF_8)
+ exchange.getResponseHeaders.add("Content-Type", "application/json")
+ exchange.sendResponseHeaders(200, bytes.length.toLong)
+ val os = exchange.getResponseBody
+ try os.write(bytes)
+ finally os.close()
+ }
+
+ private def formField(body: String, name: String): String =
+ body
+ .split("&")
+ .filter(_.contains("="))
+ .map { pair =>
+ val idx = pair.indexOf('=')
+ URLDecoder.decode(pair.substring(0, idx), StandardCharsets.UTF_8) ->
+ URLDecoder.decode(pair.substring(idx + 1), StandardCharsets.UTF_8)
+ }
+ .toMap
+ .getOrElse(name, "")
+
+ override protected def beforeAll(): Unit = {
+ super.beforeAll()
+ server.start()
+ }
+
+ override protected def afterAll(): Unit = {
+ try {
+ server.stop(0)
+ // Drop only this suite's key from AsterixDBConnUtil's host-keyed version
+ // cache singleton, so a sibling asterixdb suite keeps its own entry.
+ AsterixDBConnUtil.asterixDBVersionMapping -= host
Review Comment:
This cleanup comment claims removing `asterixDBVersionMapping -= host`
preserves a sibling suite’s entry, but the cache is keyed only by host and this
suite hardcodes `host = "localhost"`, so any other suite using the same host
key would also be affected. Update the comment to accurately describe the
intent (preventing cross-test leakage), rather than implying per-suite
isolation that the map keying cannot provide.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/sql/asterixdb/AsterixDBSourceOpDescSpec.scala:
##########
@@ -100,4 +217,183 @@ class AsterixDBSourceOpDescSpec extends AnyFlatSpec with
Matchers {
r.username shouldBe null
r.password shouldBe null
}
+
+ //
---------------------------------------------------------------------------
+ // updatePort
+ //
---------------------------------------------------------------------------
+
+ "AsterixDBSourceOpDesc.updatePort" should
+ "resolve the sentinel `default` to AsterixDB's HTTP API port and leave any
other port alone" in {
+ val d = new AsterixDBSourceOpDesc
+ d.port = "default"
+ d.updatePort()
+ d.port shouldBe "19002"
+
+ // The sentinel is recognised through surrounding whitespace.
+ d.port = " default "
+ d.updatePort()
+ d.port shouldBe "19002"
+
+ // An explicit port is left exactly as configured.
+ d.port = "19004"
+ d.updatePort()
+ d.port shouldBe "19004"
+ }
+
+ it should "be applied by sourceSchema before it issues any query" in {
+ // `default` is the port the shipped descriptor actually carries (see
+ // TestOperatorsSpec), and sourceSchema's call to updatePort is the only
+ // place in production that ever resolves it. Driving updatePort directly,
+ // as the test above does, cannot tell whether that call site still exists.
+ val d = configured()
+ d.port = "default"
+ // updatePort runs after the four requires and before the first HTTP call,
+ // so what the (unserved) real AsterixDB port answers is irrelevant: the
+ // resolution has already happened by the time the connection is refused.
+ Try(d.sourceSchema())
+ d.port shouldBe "19002"
+ }
Review Comment:
This test currently drives `sourceSchema()` with `port = "default"`, which
resolves to `19002` and then makes real HTTP calls via Unirest to
`localhost:19002`. That introduces an external network dependency (and
potential flakiness/slow timeouts) into a unit test. Stand up a minimal
in-process stub on port 19002 for this test (and stop it in a finally) so the
assertion doesn’t depend on the environment.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
