feat(scroll): sliced PIT row extraction — one reader per primary shard, merged into one stream (#238)

README.md

@@ -357,7 +357,8 @@ val results = client.search(SQLQuery("SELECT * FROM users WHERE age > 25"))
 // Type-safe queries with compile-time validation
 case class User(id: String, name: String, age: Int)
 val users: Source[User, NotUsed] = client.scrollAs[User](
-  "SELECT id, name, age FROM users WHERE active = true"
+  "SELECT id, name, age FROM users WHERE active = true",
+  client.defaultScrollConfig // the macro needs an explicit config; this one carries elastic.scroll.*
 )
 ```
 

build.sbt

@@ -20,7 +20,7 @@ ThisBuild / organization := "app.softnetwork"
 
 name := "softclient4es"
 
-ThisBuild / version := "0.20.4"
+ThisBuild / version := "0.21.0-SNAPSHOT"
 
 ThisBuild / scalaVersion := scala213
 
@@ -183,7 +183,9 @@ lazy val testkit = Project(id = "softclient4es-core-testkit", base = file("testk
     buildInfoKeys += BuildInfoKey("elasticVersion" -> elasticSearchVersion.value),
     buildInfoObject := "SoftClient4esCoreTestkitBuildInfo",
     organization := "app.softnetwork.elastic",
-    name := s"softclient4es-core-testkit"
+    name := s"softclient4es-core-testkit",
+    // the template compiles against logback (SlicedScrollCompletenessSpec log capture, #238)
+    libraryDependencies += "ch.qos.logback" % "logback-classic" % Versions.logback
   )
   .enablePlugins(BuildInfoPlugin)
   .dependsOn(
@@ -224,6 +226,9 @@ def testkitProject(esVersion: String, ss: Def.SettingsDefinition*): Project = {
         "org.apache.logging.log4j" % "log4j-api" % Versions.log4j,
         //  "org.apache.logging.log4j" % "log4j-slf4j-impl"  % Versions.log4j,
         "org.apache.logging.log4j" % "log4j-core" % Versions.log4j,
+        // SlicedScrollCompletenessSpec (#238) captures the client log through logback's
+        // ListAppender at compile time — declared, not inherited from persistence-core
+        "ch.qos.logback" % "logback-classic" % Versions.logback,
         "app.softnetwork.persistence" %% "persistence-core-testkit" % Versions.genericPersistence,
         "org.testcontainers" % "testcontainers-elasticsearch" % Versions.testContainers excludeAll (jacksonExclusions: _*),
         "org.testcontainers" % "testcontainers-minio"         % Versions.testContainers,

core/src/main/resources/softnetwork-elastic.conf

@@ -34,6 +34,17 @@ elastic {
   connection-timeout = 5s
   socket-timeout     = 30s
 
+  # Paged row extraction (scroll / PIT + search_after)
+  scroll {
+    # Rows per page. Larger pages cut round-trips linearly and raise in-flight memory linearly.
+    size = 1000
+    size = ${?ELASTIC_SCROLL_SIZE}
+    # Ceiling on concurrent PIT slices for a no-ORDER-BY extraction (ES 7.15+). The effective
+    # count is min(primary shards, max-slices); 1 disables slicing (sequential paging).
+    max-slices = 8
+    max-slices = ${?ELASTIC_SCROLL_MAX_SLICES}
+  }
+
   # When enabled, result rows surface the Elasticsearch document id as an `_id` column.
   # Disabled by default: SQL results carry only the selected columns.
   include-document-id = false

core/src/main/scala-2.12/app/softnetwork/elastic/client/ElasticConfig.scala

@@ -42,6 +42,9 @@ import java.time.Duration
  * @param includeDocumentId
  *   When enabled, result rows surface the Elasticsearch document id as an `_id` column (disabled
  *   by default)
+ * @param scroll
+ *   Paged row extraction settings (`elastic.scroll`: page size and the ceiling on concurrent PIT
+ *   slices, #238)
   */
 case class ElasticConfig(
   credentials: ElasticCredentials = ElasticCredentials(),
@@ -51,7 +54,8 @@ case class ElasticConfig(
   socketTimeout: Duration,
   metrics: MetricsConfig,
   watcher: ElasticCredentials,
-  includeDocumentId: Boolean = false)
+  includeDocumentId: Boolean = false,
+  scroll: ScrollSettings = ScrollSettings())
 
 object ElasticConfig extends StrictLogging {
   def apply(config: Config): ElasticConfig = {

core/src/main/scala-2.13/app/softnetwork/elastic/client/ElasticConfig.scala

@@ -42,6 +42,9 @@ import java.time.Duration
   * @param includeDocumentId
   *   When enabled, result rows surface the Elasticsearch document id as an `_id` column (disabled
   *   by default)
+  * @param scroll
+  *   Paged row extraction settings (`elastic.scroll`: page size and the ceiling on concurrent PIT
+  *   slices, #238)
   */
 case class ElasticConfig(
   credentials: ElasticCredentials = ElasticCredentials(),
@@ -51,7 +54,8 @@ case class ElasticConfig(
   socketTimeout: Duration,
   metrics: MetricsConfig,
   watcher: ElasticCredentials,
-  includeDocumentId: Boolean = false
+  includeDocumentId: Boolean = false,
+  scroll: ScrollSettings = ScrollSettings()
 )
 
 object ElasticConfig extends StrictLogging {

core/src/main/scala/app/softnetwork/elastic/client/ElasticClientApi.scala

@@ -17,6 +17,7 @@
 package app.softnetwork.elastic.client
 
 import app.softnetwork.common.ClientCompanion
+import app.softnetwork.elastic.client.scroll.ScrollConfig
 import app.softnetwork.elastic.licensing.metrics.MetricsApi
 import com.typesafe.config.{Config, ConfigFactory}
 import org.slf4j.Logger
@@ -65,4 +66,14 @@ trait ElasticClientApi
     * `elastic.include-document-id` (disabled by default).
     */
   override protected def includeDocumentId: Boolean = elasticConfig.includeDocumentId
+
+  /** Paged row extraction defaults come from `elastic.scroll` (#238): the page size is applied
+    * here, the slice ceiling is inherited through `maxSlices = None` so an explicit
+    * `ScrollConfig(...)` still honours the HOCON/env opt-out. A `def` — see
+    * [[ScrollApi.defaultScrollConfig]].
+    */
+  override def defaultScrollConfig: ScrollConfig =
+    ScrollConfig(scrollSize = elasticConfig.scroll.size)
+
+  override protected def configuredMaxSlices: Int = elasticConfig.scroll.maxSlices
 }

core/src/main/scala/app/softnetwork/elastic/client/ElasticsearchVersion.scala

@@ -92,6 +92,16 @@ object ElasticsearchVersion {
     isAtLeast(version, 7, 12)
   }
 
+  /** Check if PIT slicing is usable (ES >= 7.15).
+    *
+    * `slice` + `pit` in one search request exists from 7.15 (elastic/elasticsearch#74457); on
+    * 7.12–7.14 a PIT extraction pages sequentially (#238). Under a PIT, slices are contiguous
+    * doc-id ranges split first across shards — not the `_id` hash filter of sliced scroll.
+    */
+  def supportsPitSlicing(version: String): Boolean = {
+    isAtLeast(version, 7, 15)
+  }
+
   /** Check if version is ES 8+
     */
   def isEs8OrHigher(version: String): Boolean = {

core/src/main/scala/app/softnetwork/elastic/client/IndicesApi.scala

@@ -195,6 +195,9 @@ trait IndicesApi extends ElasticClientHelpers {
 
     executeCreateIndex(index, settings, updatedMappings, aliases) match {
       case success @ ElasticSuccess(true) =>
+        // #238 — a shard count cached for an expression this index is the stem of (a name probed
+        // before it existed, `orders*` before the load) must not survive the creation
+        invalidateShardCounts(Some(index))
         logger.info(s"✅ Index '$index' created successfully")
         success
       case success @ ElasticSuccess(_) =>
@@ -230,16 +233,20 @@ trait IndicesApi extends ElasticClientHelpers {
 
   def updateSchema(index: String, schema: Schema): Unit = {
     schemaCache.put(index, (schema, System.currentTimeMillis()))
+    // #238 — ALTER TABLE may have reindexed into a different shard count
+    invalidateShardCounts(Some(index))
     logger.debug(s"📦 Schema cache updated for '$index'")
   }
 
   def invalidateSchema(index: String): Unit = {
     schemaCache.remove(index)
+    invalidateShardCounts(Some(index)) // #238 — the sliced-paging shard counts follow the schema
     logger.info(s"🗑️ Schema cache invalidated for '$index'")
   }
 
   def invalidateAllSchemas(): Unit = {
     schemaCache.clear()
+    invalidateShardCounts()
     logger.info("🗑️ All schema caches invalidated")
   }