From 3ab6e393282828d11c8a78e8adc2415da4f21dec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Manciot?= Date: Thu, 30 Jul 2026 11:38:29 +0200 Subject: [PATCH] feat(client): eager extension init off the prompt thread + init observability (REPL.3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ExtensionApi.initializeExtensions(): default method forcing extension discovery + initialization on a caller-supplied ExecutionContext instead of the first query (~442ms ServiceLoader scan over a 278-jar REPL classpath) - ElasticClientDelegator forwards initializeExtensions() to the delegate — the wrapper's inherited registry is a second, unused one; run() consults the delegate's registry, so eager init must warm THAT one - ExtensionRegistry: per-extension initialize() and total scan durations logged (grep-able 'Extension discovery + initialization completed in …ms') - Cli: fire-and-forget gateway.initializeExtensions() after gateway creation, before REPL/batch dispatch — prompt appears immediately, extension-side warm-ups (e.g. arrow JoinExtension's DuckDB native load) start at startup - New ExtensionEagerInitSpec covering the count and the delegator forwarding - Version 0.20.1 -> 0.20.2-SNAPSHOT Refs #163 (fixes 1+2: async warm-up trigger point + eager init; the arrow-side DuckDBWarmup lands in softclient4es-arrow) Co-Authored-By: Claude Fable 5 --- build.sbt | 2 +- .../app/softnetwork/elastic/client/Cli.scala | 23 +++++++ .../client/ElasticClientDelegator.scala | 9 +++ .../elastic/client/ExtensionApi.scala | 14 ++++ .../elastic/client/ExtensionRegistry.scala | 19 +++++- .../client/ExtensionEagerInitSpec.scala | 67 +++++++++++++++++++ 6 files changed, 130 insertions(+), 4 deletions(-) create mode 100644 core/src/test/scala/app/softnetwork/elastic/client/ExtensionEagerInitSpec.scala diff --git a/build.sbt b/build.sbt index 807c6178..22c7824b 100644 --- a/build.sbt +++ b/build.sbt @@ -20,7 +20,7 @@ ThisBuild / organization := "app.softnetwork" name := "softclient4es" -ThisBuild / version := "0.20.1" +ThisBuild / version := "0.20.2-SNAPSHOT" ThisBuild / scalaVersion := scala213 diff --git a/core/src/main/scala/app/softnetwork/elastic/client/Cli.scala b/core/src/main/scala/app/softnetwork/elastic/client/Cli.scala index 173bf175..426a226e 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/Cli.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/Cli.scala @@ -24,6 +24,8 @@ import scala.concurrent.ExecutionContext object Cli extends App { + private val logger = org.slf4j.LoggerFactory.getLogger(getClass) + implicit val system: ActorSystem = ActorSystem("softclient4es-sql-cli") implicit val ec: ExecutionContext = system.dispatcher @@ -33,6 +35,27 @@ object Cli extends App { try { val gateway = ElasticClientFactory.createWithMonitoring(config.elasticConfig) + // #163 — eager extension init OFF the prompt thread: fold the ~442ms ServiceLoader scan (and + // kick off any extension-side warm-up, e.g. the arrow JoinExtension's DuckDB native load) + // into startup instead of the first query. Fire-and-forget: the REPL prompt (or the batch + // statement) proceeds immediately; a query racing this simply blocks on the lazy-val monitor + // (single scan, no double-init). Goes through the delegator forwarding so the DELEGATE's + // registry — the one run() consults — is the one warmed. + locally { + val initStart = System.nanoTime() + gateway.initializeExtensions().onComplete { + case scala.util.Success(count) => + logger.info( + s"🔌 $count extension(s) ready in ${(System.nanoTime() - initStart) / 1000000L}ms" + ) + case scala.util.Failure(e) => + logger.warn( + "⚠️ Eager extension initialization failed — extensions will initialize on first query", + e + ) + } + } + val executor = new StreamingReplExecutor(gateway) val repl = new Repl(executor, config.replConfig) diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala index 2b9e8f8a..9b9fade2 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala @@ -1907,6 +1907,15 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes { ): Future[ElasticResult[QueryResult]] = delegate.run(statement) + // ==================== Extensions (delegate) ==================== + + /** #163 — the delegator inherits ExtensionApi's lazy vals, so `this.extensionRegistry` would be a + * SECOND registry that no query path consults (`run` delegates to `delegate.run`, which uses the + * delegate's registry). Eager init must warm THAT one. + */ + override def initializeExtensions()(implicit ec: ExecutionContext): Future[Int] = + delegate.initializeExtensions() + // ==================== Transform (delegate) ==================== override def createTransform( diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ExtensionApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/ExtensionApi.scala index 04c25816..4a2a5851 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ExtensionApi.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ExtensionApi.scala @@ -23,6 +23,8 @@ import app.softnetwork.elastic.licensing.{ LicenseRefreshStrategyFactory } +import scala.concurrent.{ExecutionContext, Future} + trait ExtensionApi { self: ElasticClientApi => /** License refresh strategy resolved via LicenseRefreshStrategyFactory. The factory uses SPI @@ -55,4 +57,16 @@ trait ExtensionApi { self: ElasticClientApi => /** Extension registry (lazy loaded) */ lazy val extensionRegistry: ExtensionRegistry = new ExtensionRegistry(config, licenseRefreshStrategy) + + /** #163 — eagerly force extension discovery + initialization on the given execution context, + * instead of paying the ServiceLoader scan (~442ms measured over a 278-jar REPL classpath) on + * the first query. Additive: the underlying lazy vals stay the single initialization path — a + * concurrent first query simply blocks on the lazy-val monitor until the one scan finishes (and + * a FAILED lazy init is not cached, so a later query retries). + * + * @return + * the number of successfully initialized extensions. + */ + def initializeExtensions()(implicit ec: ExecutionContext): Future[Int] = + Future(extensionRegistry.extensions.size) } diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ExtensionRegistry.scala b/core/src/main/scala/app/softnetwork/elastic/client/ExtensionRegistry.scala index 2ccdf903..84d0e029 100644 --- a/core/src/main/scala/app/softnetwork/elastic/client/ExtensionRegistry.scala +++ b/core/src/main/scala/app/softnetwork/elastic/client/ExtensionRegistry.scala @@ -35,6 +35,7 @@ class ExtensionRegistry( /** All loaded extensions. */ lazy val extensions: Seq[ExtensionSpi] = { + val scanStart = System.nanoTime() val loader = ServiceLoader.load(classOf[ExtensionSpi]) val loaded = loader.iterator().asScala.toSeq.flatMap { ext => @@ -42,21 +43,33 @@ class ExtensionRegistry( s"🔌 Discovered extension: ${ext.extensionName} v${ext.version} (priority: ${ext.priority})" ) + val initStart = System.nanoTime() ext.initialize(config, licenseRefreshStrategy) match { case Right(_) => - logger.info(s"✅ Extension ${ext.extensionName} initialized successfully") + logger.info( + s"✅ Extension ${ext.extensionName} initialized successfully in ${elapsedMs(initStart)}ms" + ) Some(ext) case Left(error) => - logger.warn(s"⚠️ Failed to initialize extension ${ext.extensionName}: $error") + logger.warn( + s"⚠️ Failed to initialize extension ${ext.extensionName} after ${elapsedMs(initStart)}ms: $error" + ) None } } // ✅ Sort by priority (lower = higher priority) - loaded.sortBy(_.priority) + val sorted = loaded.sortBy(_.priority) + logger.info( + s"🔌 Extension discovery + initialization completed in ${elapsedMs(scanStart)}ms " + + s"(${sorted.size} extension(s) active)" + ) + sorted } + private def elapsedMs(t0: Long): Long = (System.nanoTime() - t0) / 1000000L + /** Find extension that can handle a statement. */ def findHandler(statement: Statement): Option[ExtensionSpi] = { diff --git a/core/src/test/scala/app/softnetwork/elastic/client/ExtensionEagerInitSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/ExtensionEagerInitSpec.scala new file mode 100644 index 00000000..0153005e --- /dev/null +++ b/core/src/test/scala/app/softnetwork/elastic/client/ExtensionEagerInitSpec.scala @@ -0,0 +1,67 @@ +/* + * Copyright 2025 SOFTNETWORK + * + * Licensed 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 app.softnetwork.elastic.client + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.slf4j.{Logger, LoggerFactory} + +import scala.concurrent.duration._ +import scala.concurrent.{Await, ExecutionContext, Future} + +/** #163 — eager extension init. Two guarantees: + * 1. `initializeExtensions()` forces the ServiceLoader scan and reports the count (core's own + * META-INF/services registers CoreDdlExtension + CoreDqlExtension → always >= 2 here). + * 1. THE TRAP — a delegator forwards to its delegate — the wrapper's own (inherited, unused) + * registry must NOT be the one warmed, because run() delegates to delegate.run(). + */ +class ExtensionEagerInitSpec extends AnyFlatSpec with Matchers { + + implicit val ec: ExecutionContext = ExecutionContext.global + + private val testLogger: Logger = LoggerFactory.getLogger(getClass) + + // `protected def logger` is the ONLY abstract member NopeClientApi leaves open; `config` and + // `metrics` have concrete defaults, so licenseRefreshStrategy resolves from ConfigFactory.load() + // → Community fallback, and clusterUuid/clusterName/version no-op failures are absorbed by + // ExtensionApi's `case _ =>`. Same recipe as CoreDqlExtensionSpec's RecordingClient. + private def newNopeClient(): ElasticClientApi = new NopeClientApi { + override protected def logger: Logger = testLogger + } + + "initializeExtensions" should "force the registry and report the loaded-extension count" in { + val client = newNopeClient() + val n = Await.result(client.initializeExtensions(), 30.seconds) + n should be >= 2 // CoreDdlExtension + CoreDqlExtension from core's META-INF/services + } + + it should "be forwarded to the delegate by ElasticClientDelegator" in { + @volatile var forwarded = false + val probe: ElasticClientApi = new NopeClientApi { + override protected def logger: Logger = testLogger + override def initializeExtensions()(implicit ec: ExecutionContext): Future[Int] = { + forwarded = true + Future.successful(42) + } + } + val wrapper = new ElasticClientDelegator { + override val delegate: ElasticClientApi = probe + } + Await.result(wrapper.initializeExtensions(), 5.seconds) shouldBe 42 + forwarded shouldBe true + } +}