Jelly-SPARQL: add support for punctuated streams

build.sbt

@@ -11,7 +11,7 @@ resolvers +=
   "Sonatype OSS Snapshots" at "https://central.sonatype.com/repository/maven-snapshots"
 
 lazy val jenaV = "6.2.0"
-lazy val jellyV = "3.7.3+71-c920dacf-SNAPSHOT"
+lazy val jellyV = "3.7.3+75-03c595ae-SNAPSHOT"
 lazy val graalvmV = "25.4.4.1.1"
 
 addCommandAlias("fixAll", "scalafixAll; scalafmtAll")

src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJelly.scala

@@ -26,7 +26,9 @@ object RdfToJellyPrint extends RdfCommandPrintUtil[RdfFormat.Readable]:
     "The input can also be a Jelly-SPARQL result set (the jelly-sparql format) with 3 or 4 " +
     "variables, which are taken by position, not by name: subject, predicate, object, and graph. " +
     "Each solution becomes one statement. With 3 variables, the output is a TRIPLES stream, " +
-    "with 4 a QUADS stream, where an unbound graph means the default graph.\n" +
+    "with 4 a QUADS stream, where an unbound graph means the default graph. Each result set of " +
+    "a PUNCTUATED stream becomes one frame, and the logical type is GRAPHS or DATASETS, unless " +
+    "set with --opt.logical-type.\n" +
     "Note: this command works in a streaming manner and scales well to large files. ",
 )
 @ArgsName("<file-to-convert>")

src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlFromJelly.scala

@@ -18,6 +18,9 @@ object SparqlFromJellyPrint:
     "If no input file is specified, the input is read from stdin.\n" +
     "If no output file is specified, the output is written to stdout.\n" +
     "Both SELECT results (bindings) and ASK results (a boolean) are supported.\n" +
+    "The result sets of a PUNCTUATED stream are written one after another to the same output. " +
+    "With the jelly-rdf output format, each of them becomes one frame of a GRAPHS or DATASETS " +
+    "stream.\n" +
     "The jelly-sparql-text format is a human-readable version of Jelly-SPARQL, for debugging.\n" +
     "If an error is detected, the program will exit with a non-zero code.\n" +
     "Otherwise, the program will exit with code 0.",

src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspect.scala

@@ -4,6 +4,7 @@ import caseapp.*
 import eu.neverblink.jelly.cli.*
 import eu.neverblink.jelly.cli.command.rdf.util.FrameInfo
 import eu.neverblink.jelly.cli.command.sparql.util.*
+import eu.neverblink.jelly.core.proto.v1.sparql.SparqlStreamType
 
 import scala.jdk.CollectionConverters.*
 
@@ -12,6 +13,8 @@ import scala.jdk.CollectionConverters.*
     "The statistics are returned as a valid YAML. \n" +
     "If no input file is specified, the input is read from stdin.\n" +
     "If no output file is specified, the output is written to stdout.\n" +
+    "For a PUNCTUATED stream (a sequence of result sets), the number of result sets is printed " +
+    "instead of the variables.\n" +
     "If an error is detected, the program will exit with a non-zero code.\n",
   "Note: this command works in a streaming manner and scales well to large files. " +
     "It only reads the structure of the frames, it does not decode the RDF terms – " +
@@ -55,10 +58,16 @@ object SparqlInspect extends JellyCommand[SparqlInspectOptions]:
       val firstFrame = frames.head
       if firstFrame.getOptions == null then
         throw CriticalException("First frame in the input stream does not contain stream options")
+      val punctuated = firstFrame.getOptions.getStreamType == SparqlStreamType.PUNCTUATED
+      // In a PUNCTUATED stream, every frame after a trailer starts a new result set
+      var resultSetIndex = -1L
+      var resultSetStart = true
       val frameInfos = frames.zipWithIndex.map { (frame, i) =>
+        if resultSetStart then resultSetIndex += 1
         val metadata = frame.getMetadata.asScala.map(e => e.getKey -> e.getValue).toMap
-        val info = SparqlFrameInfo(i, metadata)
-        info.processFrame(frame)
+        val info = SparqlFrameInfo(i, metadata, Option.when(punctuated)(resultSetIndex))
+        info.processFrame(frame, resultSetStart)
+        resultSetStart = punctuated && frame.getTrailer != null
         info
       }
       if options.perFrame then

src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesCommand.scala

@@ -4,9 +4,11 @@ import caseapp.*
 import eu.neverblink.jelly.cli.*
 import eu.neverblink.jelly.cli.command.sparql.util.{SparqlFormat, SparqlJellyUtil}
 import eu.neverblink.jelly.cli.util.io.ProtoText
+import eu.neverblink.jelly.convert.jena.sparql.{JenaSparqlConverterFactory, RowSetReaderJelly}
 import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsFrame
 import org.apache.jena.riot.RIOT
-import org.apache.jena.riot.resultset.{ResultSetReaderRegistry, ResultSetWriterRegistry}
+import org.apache.jena.riot.rowset.{RowSetReaderRegistry, RowSetWriterRegistry}
+import org.apache.jena.sparql.exec.QueryExecResult
 import org.apache.jena.sparql.util.Context
 
 import java.io.{BufferedReader, InputStream, InputStreamReader, OutputStream}
@@ -61,7 +63,8 @@ abstract class SparqlSerDesCommand[T <: HasJellyCommandOptions: {Parser, Help}]
 
   /** Reads a result set in one format and writes it back out in another.
     *
-    * Both SELECT results (bindings) and ASK results (a single boolean) are handled.
+    * Both SELECT results (bindings) and ASK results (a single boolean) are handled. All result sets
+    * of a PUNCTUATED stream are written to the same output, one after another.
     */
   final def convert(
       from: SparqlFormat,
@@ -93,12 +96,26 @@ abstract class SparqlSerDesCommand[T <: HasJellyCommandOptions: {Parser, Help}]
       outputStream: OutputStream,
   ): Unit =
     val context = jenaContext
-    val reader = ResultSetReaderRegistry.getFactory(from.jenaLang).create(from.jenaLang)
-    val writer = ResultSetWriterRegistry.getFactory(to.jenaLang).create(to.jenaLang)
-    val result = reader.readAny(inputStream, context)
-    if result.isBoolean then
-      writer.write(outputStream, result.getBooleanResult.booleanValue, context)
-    else writer.write(outputStream, result.getResultSet, context)
+    val writer = RowSetWriterRegistry.getFactory(to.jenaLang).create(to.jenaLang)
+    def write(result: QueryExecResult): Unit =
+      if result.isBoolean then
+        writer.write(outputStream, result.booleanResult.booleanValue, context)
+      else writer.write(outputStream, result.rowSet, context)
+    from match
+      case SparqlFormat.JellySparql =>
+        // A PUNCTUATED stream has many result sets, which are written one after another
+        RowSetReaderJelly(RowSetReaderJelly.Options(), JenaSparqlConverterFactory.getInstance())
+          .readAll(inputStream, context, write)
+      case _ => write(readResult(from, inputStream, context))
+
+  /** Reads a result set in a format other than Jelly-SPARQL. */
+  protected final def readResult(
+      from: SparqlFormat.Jena,
+      inputStream: InputStream,
+      context: Context,
+  ): QueryExecResult =
+    RowSetReaderRegistry.getFactory(from.jenaLang).create(from.jenaLang)
+      .readAny(inputStream, context)
 
   private val frameCommentPrefix = "# Frame"
 

src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlToJelly.scala

@@ -4,13 +4,19 @@ import caseapp.*
 import eu.neverblink.jelly.cli.*
 import eu.neverblink.jelly.cli.command.sparql.util.*
 import eu.neverblink.jelly.cli.util.jena.RdfSparqlConverter
-import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage
+import eu.neverblink.jelly.cli.util.io.IoUtil
+import eu.neverblink.jelly.convert.jena.sparql.{
+  JellySparqlLanguage,
+  JenaSparqlConverterFactory,
+  RowSetWriterJelly,
+}
 import eu.neverblink.jelly.core.RdfProtoDeserializationError
-import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsOptions
+import eu.neverblink.jelly.core.proto.v1.sparql.{SparqlResultsOptions, SparqlStreamType}
 import eu.neverblink.jelly.core.sparql.{JellySparqlConstants, JellySparqlOptions}
 import org.apache.jena.sparql.util.Context
 
 import java.io.{InputStream, OutputStream}
+import scala.util.Using
 
 object SparqlToJellyPrint:
   val validFormats: List[SparqlFormat] = SparqlFormat.readable
@@ -23,6 +29,8 @@ object SparqlToJellyPrint:
     "If no output file is specified, the output is written to stdout.\n" +
     "Both SELECT results (bindings) and ASK results (a boolean) are supported.\n" +
     "The input can also be in the jelly-sparql-text format, which is written by sparql from-jelly.\n" +
+    "With several input files, the output is a PUNCTUATED stream, with one result set per file, " +
+    "in order. jelly-sparql-text and jelly-rdf input can only be read from one file.\n" +
     "With the jelly-rdf input format, a Jelly-RDF stream is turned into a result set with one " +
     "solution per statement: ?s ?p ?o for TRIPLES streams, and ?s ?p ?o ?g for QUADS and GRAPHS " +
     "streams, where ?g is unbound for the default graph. Namespace declarations are dropped. " +
@@ -31,7 +39,7 @@ object SparqlToJellyPrint:
     "If an error is detected, the program will exit with a non-zero code.\n" +
     "Otherwise, the program will exit with code 0.",
 )
-@ArgsName("<file-to-convert>")
+@ArgsName("<files-to-convert>")
 case class SparqlToJellyOptions(
     @Recurse
     common: JellyCommandOptions = JellyCommandOptions(),
@@ -87,8 +95,10 @@ object SparqlToJelly extends SparqlSerDesCommand[SparqlToJellyOptions]:
   override protected def delimitedOutput: Boolean = getOptions.delimited
 
   override def doRun(options: SparqlToJellyOptions, remainingArgs: RemainingArgs): Unit =
-    val inputFile = remainingArgs.remaining.headOption
-    val inputFormat = resolveFormat(options.inputFormat, inputFile)
+    val inputFiles = remainingArgs.remaining
+    val inputFormats =
+      if inputFiles.isEmpty then Seq(resolveFormat(options.inputFormat, None))
+      else inputFiles.map(f => resolveFormat(options.inputFormat, Some(f)))
     if options.valuesPerFrame < 1 then
       throw InvalidArgument(
         "--values-per-frame",
@@ -97,9 +107,51 @@ object SparqlToJelly extends SparqlSerDesCommand[SparqlToJellyOptions]:
       )
     baseOptions = options.optionsFrom.map(SparqlJellyUtil.loadOptionsFromFile)
     streamOptions = options.jellySerializationOptions.toSparqlResultsOptions(baseOptions)
-    if !isQuietMode then checkAndWarnOptions(inputFormat)
-    val (inputStream, outputStream) = getIoStreamsFromOptions(inputFile, options.outputFile)
-    convert(inputFormat, SparqlFormat.JellySparql, inputStream, outputStream)
+    if inputFiles.size > 1 then
+      if options.jellySerializationOptions.streamType.contains(SparqlStreamType.FLAT) then
+        throw InvalidArgument(
+          "--opt.stream-type",
+          options.jellySerializationOptions.`opt.streamType`.get,
+          Some("Several input files can only be written as a PUNCTUATED stream"),
+        )
+      streamOptions = streamOptions.clone().setStreamType(SparqlStreamType.PUNCTUATED)
+      inputFormats.find(!_.isInstanceOf[SparqlFormat.Jena]).foreach { f =>
+        throw CriticalException(s"$f input can only be read from one file, not several.")
+      }
+    val punctuated = streamOptions.getStreamType == SparqlStreamType.PUNCTUATED
+    if punctuated && !options.delimited && inputFormats.head != SparqlFormat.JellySparqlText then
+      throw InvalidArgument(
+        "--delimited",
+        "false",
+        Some("PUNCTUATED streams are always written delimited"),
+      )
+    if !isQuietMode then checkAndWarnOptions(inputFormats.head)
+    val (inputStream, outputStream) =
+      getIoStreamsFromOptions(inputFiles.headOption, options.outputFile)
+    inputFormats.head match
+      case format: SparqlFormat.Jena if punctuated =>
+        val writer = resultSetsWriter(outputStream)
+        writeResultSet(writer, format, inputStream)
+        // The formats of the other files were checked above
+        for case (file, format: SparqlFormat.Jena) <- inputFiles.zip(inputFormats).drop(1) do
+          Using.resource(IoUtil.inputStream(file))(writeResultSet(writer, format, _))
+        outputStream.flush()
+      case format => convert(format, SparqlFormat.JellySparql, inputStream, outputStream)
+
+  private def resultSetsWriter(outputStream: OutputStream): RowSetWriterJelly.ResultSetsWriter =
+    RowSetWriterJelly(RowSetWriterJelly.Options(), JenaSparqlConverterFactory.getInstance())
+      .resultSetsWriter(outputStream, jenaContext)
+
+  private def writeResultSet(
+      writer: RowSetWriterJelly.ResultSetsWriter,
+      format: SparqlFormat.Jena,
+      inputStream: InputStream,
+  ): Unit =
+    SparqlJellyUtil.translateErrors {
+      val result = readResult(format, inputStream, jenaContext)
+      if result.isBoolean then writer.write(result.booleanResult.booleanValue)
+      else writer.write(result.rowSet)
+    }
 
   override protected def jellyRdfToSparql(
       inputStream: InputStream,
@@ -126,9 +178,11 @@ object SparqlToJelly extends SparqlSerDesCommand[SparqlToJellyOptions]:
         )
     else
       try
+        // Readers of PUNCTUATED streams must ask for them, so only the rest is checked here
         JellySparqlOptions.checkCompatibility(
           streamOptions,
-          JellySparqlOptions.DEFAULT_SUPPORTED_OPTIONS,
+          JellySparqlOptions.DEFAULT_SUPPORTED_OPTIONS.clone()
+            .setStreamType(streamOptions.getStreamType),
         )
       catch
         case e: RdfProtoDeserializationError =>