From 982a30a5f5c8907ebc470c713ffc23d053700b42 Mon Sep 17 00:00:00 2001 From: Ostrzyciel Date: Thu, 8 Oct 2026 10:22:07 +0200 Subject: [PATCH] Jelly-SPARQL: add support for punctuated streams - `sparql to-jelly` can take multiple files as input - `sparql from-jelly` will just concat the result set files on stdout (RDF Messages-style) - Conversion to Jelly-RDF will put each result set in one frame and set an appropriate logical stream type. - `sparql validate` can take multiple input comparison files OR can compare against another punctuated stream. - `sparql inspect` can handle multiple result sets --- build.sbt | 2 +- .../jelly/cli/command/rdf/RdfToJelly.scala | 4 +- .../cli/command/sparql/SparqlFromJelly.scala | 3 + .../cli/command/sparql/SparqlInspect.scala | 13 +- .../command/sparql/SparqlSerDesCommand.scala | 33 ++- .../cli/command/sparql/SparqlToJelly.scala | 72 +++++- .../cli/command/sparql/SparqlValidate.scala | 222 ++++++++++++------ .../SparqlJellySerializationOptions.scala | 23 +- .../sparql/util/SparqlMetricsPrinter.scala | 57 ++++- .../sparql/util/SparqlResultSetReader.scala | 114 +++++++++ .../cli/util/jena/RdfSparqlConverter.scala | 136 +++++++---- .../cli/util/jena/riot/JellyFrameWriter.scala | 85 +++++++ .../command/helpers/RdfSparqlTestData.scala | 13 +- .../command/sparql/SparqlInspectSpec.scala | 32 +++ .../cli/command/sparql/SparqlSerDesSpec.scala | 192 ++++++++++++++- .../command/sparql/SparqlValidateSpec.scala | 153 ++++++++++-- .../util/jena/RdfSparqlConverterSpec.scala | 121 +++++++++- 17 files changed, 1099 insertions(+), 176 deletions(-) create mode 100644 src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlResultSetReader.scala create mode 100644 src/main/scala/eu/neverblink/jelly/cli/util/jena/riot/JellyFrameWriter.scala diff --git a/build.sbt b/build.sbt index a591ba2..0bbcfb4 100644 --- a/build.sbt +++ b/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") diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJelly.scala b/src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJelly.scala index 8b4a114..a31a967 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJelly.scala +++ b/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("") diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlFromJelly.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlFromJelly.scala index f9a08da..372398b 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlFromJelly.scala +++ b/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.", diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspect.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspect.scala index 556b15e..499e298 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspect.scala +++ b/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 diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesCommand.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesCommand.scala index a929182..66dc729 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesCommand.scala +++ b/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" diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlToJelly.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlToJelly.scala index 62fbc69..479a300 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlToJelly.scala +++ b/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("") +@ArgsName("") 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 => diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidate.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidate.scala index e52f403..0d625f4 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidate.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidate.scala @@ -5,13 +5,16 @@ import eu.neverblink.jelly.cli.* import eu.neverblink.jelly.cli.command.sparql.util.* import eu.neverblink.jelly.cli.util.io.{IoUtil, ProtoText} import eu.neverblink.jelly.convert.jena.sparql.JenaSparqlConverterFactory -import eu.neverblink.jelly.core.proto.v1.sparql.{SparqlResultsFrame, SparqlResultsOptions} +import eu.neverblink.jelly.core.proto.v1.sparql.{ + SparqlResultsFrame, + SparqlResultsOptions, + SparqlStreamType, +} import eu.neverblink.jelly.core.sparql.{JellySparqlOptions, SparqlResultsHandler} import org.apache.jena.graph.Node import org.apache.jena.riot.{Lang, RIOT} -import org.apache.jena.riot.resultset.ResultSetReaderRegistry +import org.apache.jena.riot.rowset.RowSetReaderRegistry import org.apache.jena.sparql.core.Var -import org.apache.jena.sparql.engine.binding.Binding import org.apache.jena.sparql.util.FmtUtils import java.io.InputStream @@ -47,12 +50,13 @@ case class SparqlValidateOptions( @Recurse common: JellyCommandOptions = JellyCommandOptions(), @HelpMessage( - "SPARQL results file to compare the input stream to. If not specified, no comparison is done.", + "SPARQL results file to compare the input stream to. Can be given several times, for the " + + "result sets of a PUNCTUATED stream. If not specified, no comparison is done.", ) - compareToFile: Option[String] = None, + compareToFile: List[String] = Nil, @HelpMessage( - "Format of the SPARQL results file to compare the input stream to. If not specified, " + - "the format is inferred from the file name. Possible values: " + + "Format of the SPARQL results files to compare the input stream to. If not specified, " + + "the format is inferred from the file names. Possible values: " + SparqlValidatePrint.validFormatsString, ) compareToFormat: Option[String] = None, @@ -68,20 +72,16 @@ case class SparqlValidateOptions( delimited: String = "either", @HelpMessage( "Whether the stream must end with a trailer, which tells the reader that the result set " + - "is complete. Default: false", + "is complete. Producers must always write one, so a stream without it is most likely " + + "truncated. Use --require-trailer=false to check such a stream anyway. Default: true", ) - requireTrailer: Boolean = false, + requireTrailer: Boolean = true, ) extends HasJellyCommandOptions object SparqlValidate extends JellyCommand[SparqlValidateOptions]: private enum Delimiting: case Either, Delimited, Undelimited - /** The result set of the reference file. The solutions are read as they are compared. */ - private enum Reference: - case Select(variables: Seq[Var], rows: Iterator[Binding]) - case Ask(value: Boolean) - override def names: List[List[String]] = List(List("sparql", "validate")) override def group = "sparql" @@ -98,18 +98,18 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: Some("Valid values: true, false, either"), ) val expectedOptions = options.optionsFile.map(SparqlJellyUtil.loadOptionsFromFile) - val referenceLang = options.compareToFile.map(referenceFormat(_, options.compareToFormat)) + val referenceFiles = + options.compareToFile.map(f => f -> referenceFormat(f, options.compareToFormat)) val (inputStream, _) = getIoStreamsFromOptions(remainingArgs.remaining.headOption, None) SparqlJellyUtil.translateErrors { val (delimited, frames) = SparqlJellyUtil.iterateSparqlStreamWithDelimitingInfo(inputStream) validateDelimiting(delimiting, delimited) - (options.compareToFile, referenceLang) match - case (Some(fileName), Some(lang)) => - // Keep the reference file open while its solutions are compared - Using.resource(IoUtil.inputStream(fileName)) { is => - validateContent(frames, expectedOptions, Some(readReference(is, lang))) - } - case _ => validateContent(frames, expectedOptions, None) + if referenceFiles.isEmpty then validateContent(frames, expectedOptions, None) + else + // Keep the reference files open while their solutions are compared + Using.resource(References(referenceFiles)) { references => + validateContent(frames, expectedOptions, Some(references)) + } } private def validateDelimiting(expected: Delimiting, delimited: Boolean): Unit = @@ -134,6 +134,9 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: s"Actual: ${printOptions(streamOptions)}", ) } + validateVersion(streamOptions) + + private def validateVersion(streamOptions: SparqlResultsOptions): Unit = if streamOptions.getVersion <= 0 then throw CriticalException( "The version field in SparqlResultsOptions is <= 0. " + @@ -145,19 +148,20 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: /** Pushes all frames through the decoder, which catches most of the errors, and checks the rest. * - * @param reference - * the result set to compare the input to, if any + * @param references + * the result sets to compare the input to, if any */ private def validateContent( frames: Iterator[SparqlResultsFrame], expectedOptions: Option[SparqlResultsOptions], - reference: Option[Reference], + references: Option[References], ): Unit = if !frames.hasNext then throw CriticalException("Empty input stream") - val handler = ComparingHandler(reference) + val handler = ComparingHandler(references) // The stream options are compared with the expected ones separately. Here, we only make sure // that the decoder accepts lookup tables as large as the expected ones. val supported = JellySparqlOptions.DEFAULT_SUPPORTED_OPTIONS.clone() + .setStreamType(SparqlStreamType.PUNCTUATED) expectedOptions.foreach { e => supported .setMaxNameTableSize(e.getMaxNameTableSize.max(supported.getMaxNameTableSize)) @@ -166,11 +170,12 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: } val decoder = JenaSparqlConverterFactory.getInstance().decoder(handler, supported) var firstOptions: Option[SparqlResultsOptions] = None - // Whether the current stream ended with a trailer. Repeated options start a new stream. + // Whether the previous frame ended with a trailer. Repeated options in a FLAT stream start a + // new segment of a concatenated stream, which should also end with a trailer. In a PUNCTUATED + // stream, the decoder only accepts them after a trailer. var trailerSeen = false def checkTrailer(): Unit = - // An ASK result is complete on its own, so it does not need a trailer - if getOptions.requireTrailer && !trailerSeen && handler.askResult.isEmpty then + if getOptions.requireTrailer && !trailerSeen then throw CriticalException("The stream does not end with a trailer") for (frame, i) <- frames.zipWithIndex do @@ -178,13 +183,11 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: case Some(o) if firstOptions.isEmpty => validateOptions(o, expectedOptions) firstOptions = Some(o) + handler.punctuated = o.getStreamType == SparqlStreamType.PUNCTUATED case Some(o) => - if !firstOptions.contains(o) then - throw CriticalException( - s"Later occurrence of stream options in frame $i does not match the first", - ) + // Repeated options need not be the same as the first ones, but must be valid on their own + validateVersion(o) checkTrailer() - trailerSeen = false case None if firstOptions.isEmpty => throw CriticalException("First frame in the input stream does not contain stream options") case None => () @@ -192,32 +195,86 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: Option(frame.getTrailer).foreach { trailer => if trailer.getError.nonEmpty then throw CriticalException( - s"The trailer in frame $i says that the result set is incomplete: ${trailer.getError}", + s"The trailer in frame $i says that ${handler.resultSetName} is incomplete: " + + trailer.getError, ) - trailerSeen = true } + trailerSeen = frame.getTrailer != null checkTrailer() handler.finish() - /** Checks what the decoder reads from the stream against the reference, if there is one. */ - private class ComparingHandler(reference: Option[Reference]) extends SparqlResultsHandler[Node]: - var askResult: Option[Boolean] = None + /** The result sets of the reference files, in order, with the name of the file each one is from. + * The files are opened as they are needed, and the solutions are read as they are compared. + */ + private class References(files: Seq[(String, Lang)]) + extends Iterator[(String, SparqlResultSet)], + AutoCloseable: + private val remainingFiles = files.iterator + private val opened = mutable.ArrayBuffer[InputStream]() + private var fileName = "" + private var resultSets: Iterator[SparqlResultSet] = Iterator.empty + + override def hasNext: Boolean = + while !resultSets.hasNext && remainingFiles.hasNext do + val (name, lang) = remainingFiles.next() + val is = IoUtil.inputStream(name) + opened += is + fileName = name + resultSets = readReference(is, lang) + resultSets.hasNext + + override def next(): (String, SparqlResultSet) = + if !hasNext then throw java.util.NoSuchElementException() + (fileName, resultSets.next()) + + override def close(): Unit = opened.foreach(_.close()) + + /** Checks what the decoder reads from the stream against the references, if there are any. */ + private class ComparingHandler(references: Option[References]) extends SparqlResultsHandler[Node]: + // Set once the stream options are known + var punctuated = false + // Index of the current result set + private var resultSetIndex = -1L + // The result set to compare the current one to, and the file it is from + private var reference: Option[(String, SparqlResultSet)] = None private var variables: IndexedSeq[Var] = IndexedSeq() - // Number of solutions read so far + // Number of solutions read so far in the current result set private var index = 0L - private val blankNodes = BlankNodeMapping() + private var blankNodes = BlankNodeMapping() - private def fileName = getOptions.compareToFile.getOrElse("") + /** How to refer to the current result set in messages. */ + def resultSetName: String = + if punctuated then s"result set ${resultSetIndex.max(0)}" else "the result set" + + // Only in a PUNCTUATED stream, where there may be several result sets + private def inResultSet: String = if punctuated then s" in result set $resultSetIndex" else "" + + /** Moves on to the next result set, and the next reference. */ + private def startResultSet(): Unit = + finishResultSet() + resultSetIndex += 1 + index = 0 + // Blank node labels are scoped to one result set + blankNodes = BlankNodeMapping() + reference = references.map { refs => + if !refs.hasNext then + throw CriticalException( + s"Expected $resultSetIndex result sets, as in the reference files, " + + "but the input stream has more", + ) + refs.next() + } override def handleVariables(variables: java.util.List[String]): Unit = + startResultSet() this.variables = variables.asScala.map(Var.alloc).toIndexedSeq reference match - case Some(Reference.Ask(_)) => - throw CriticalException(s"Expected an ASK result, as in $fileName, but got solutions") - case Some(Reference.Select(expected, _)) if expected != this.variables => + case Some((file, SparqlResultSet.Ask(_))) => + throw CriticalException(s"Expected an ASK result, as in $file, but got solutions") + case Some((file, SparqlResultSet.Select(expected, _))) if expected != this.variables => throw CriticalException( - s"Expected variables ${expected.mkString(" ")}, as in $fileName, " + - s"but got ${this.variables.mkString(" ")}", + s"Expected variables ${expected.mkString(" ")}, as in $file, " + + s"but got ${this.variables.mkString(" ")}$inResultSet", ) case _ => () @@ -225,41 +282,57 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: override def handleRow(row: Array[Node]): Unit = reference match - case Some(Reference.Select(_, rows)) => + case Some((file, SparqlResultSet.Select(_, rows))) => if !rows.hasNext then throw CriticalException( - s"Expected $index solutions, as in $fileName, but the input stream has more", + s"Expected $index solutions, as in $file, but the input stream has more" + + inResultSet, ) val expected = rows.next() - val matches = variables.indices.forall { i => - blankNodes.sameTerm(expected.get(variables(i)), row(i)) - } + val matches = variables.indices.forall(i => blankNodes.sameTerm(expected(i), row(i))) if !matches then throw CriticalException( - s"Solution $index does not match the one in $fileName\n" + - s"Expected: ${formatRow(variables.map(expected.get))}\n" + + s"Solution $index$inResultSet does not match the one in $file\n" + + s"Expected: ${formatRow(expected.toIndexedSeq)}\n" + s"Actual: ${formatRow(row.toIndexedSeq)}", ) case _ => () index += 1 override def handleAskResult(value: Boolean): Unit = - askResult = Some(value) + startResultSet() reference match - case Some(Reference.Ask(expected)) if expected != value => - throw CriticalException(s"Expected the ASK result to be $expected, but it was $value") - case Some(Reference.Select(_, _)) => - throw CriticalException(s"Expected solutions, as in $fileName, but got an ASK result") + case Some((file, SparqlResultSet.Ask(expected))) if expected != value => + throw CriticalException( + s"Expected the ASK result to be $expected, as in $file, but it was $value" + + inResultSet, + ) + case Some((file, SparqlResultSet.Select(_, _))) => + throw CriticalException( + s"Expected solutions, as in $file, but got an ASK result$inResultSet", + ) case _ => () /** Checks that the reference has no more solutions than the input. */ - def finish(): Unit = reference match - case Some(Reference.Select(_, rows)) if rows.hasNext => + private def finishResultSet(): Unit = reference match + case Some((file, SparqlResultSet.Select(_, rows))) if rows.hasNext => throw CriticalException( - s"Expected ${index + rows.size} solutions, as in $fileName, but got $index", + s"Expected ${index + rows.size} solutions, as in $file, but got $index$inResultSet", ) case _ => () + /** Checks that the references have no more result sets or solutions than the input. */ + def finish(): Unit = + finishResultSet() + references.foreach { refs => + if refs.hasNext then + val count = resultSetIndex + 1 + throw CriticalException( + s"Expected ${count + refs.size} result sets, as in the reference files, " + + s"but got $count", + ) + } + private def formatRow(values: Seq[Node]): String = variables.zip(values).map { (v, n) => s"$v=${if n == null then "(unbound)" else FmtUtils.stringForNode(n)}" @@ -311,13 +384,20 @@ object SparqlValidate extends JellyCommand[SparqlValidateOptions]: case f => throw CriticalException(s"Cannot read $f for comparison") /** Starts reading the reference file. The solutions are read lazily, as they are compared. */ - private def readReference(inputStream: InputStream, lang: Lang): Reference = - val result = ResultSetReaderRegistry.getFactory(lang).create(lang) - .readAny(inputStream, RIOT.getContext.copy()) - if result.isBoolean then Reference.Ask(result.getBooleanResult.booleanValue) + private def readReference(inputStream: InputStream, lang: Lang): Iterator[SparqlResultSet] = + if lang == SparqlFormat.JellySparql.jenaLang then + SparqlResultSetReader(SparqlJellyUtil.iterateSparqlStream(inputStream)) else - val rs = result.getResultSet - Reference.Select( - rs.getResultVars.asScala.map(Var.alloc).toSeq, - Iterator.continually(rs).takeWhile(_.hasNext).map(_.nextBinding()), - ) + val result = RowSetReaderRegistry.getFactory(lang).create(lang) + .readAny(inputStream, RIOT.getContext.copy()) + if result.isBoolean then Iterator(SparqlResultSet.Ask(result.booleanResult.booleanValue)) + else + val rowSet = result.rowSet + val vars = rowSet.getResultVars.asScala.toIndexedSeq + Iterator( + SparqlResultSet.Select( + vars, + Iterator.continually(rowSet).takeWhile(_.hasNext).map(_.next()) + .map(b => vars.map(b.get).toArray), + ), + ) diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlJellySerializationOptions.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlJellySerializationOptions.scala index aeab8c0..6be04d3 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlJellySerializationOptions.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlJellySerializationOptions.scala @@ -3,7 +3,7 @@ package eu.neverblink.jelly.cli.command.sparql.util import caseapp.* import eu.neverblink.jelly.cli.InvalidArgument import eu.neverblink.jelly.core.proto.v1.RdfVersion -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.JellySparqlOptions import eu.neverblink.jelly.core.utils.RdfVersionUtils @@ -17,6 +17,11 @@ private val defaultOptions: SparqlResultsOptions = JellySparqlOptions.BIG case class SparqlJellySerializationOptions( @HelpMessage("Name of the output stream (in metadata). Default: (empty)") `opt.streamName`: Option[String] = None, + @HelpMessage( + "Type of the stream. One of: flat (one result set), punctuated (a sequence of result " + + "sets). Default: flat, or punctuated if there is more than one input file", + ) + `opt.streamType`: Option[String] = None, @HelpMessage( "Version of RDF whose terms may occur in the stream. One of: 1.1 (no triple terms, " + "no base directions), 1.2-basic (base directions, no triple terms), 1.2 (all RDF 1.2 " + @@ -39,6 +44,19 @@ case class SparqlJellySerializationOptions( `opt.maxDatatypeTableSize`: Option[Int] = None, ): + /** The stream type set on the command line, if any. */ + def streamType: Option[SparqlStreamType] = + `opt.streamType`.map(_.trim.toLowerCase).map { + case "flat" => SparqlStreamType.FLAT + case "punctuated" => SparqlStreamType.PUNCTUATED + case _ => + throw InvalidArgument( + "--opt.stream-type", + `opt.streamType`.get, + Some("Must be one of: flat, punctuated"), + ) + } + private def rdfVersion: Option[RdfVersion] = `opt.rdfVersion`.map(_.trim.toLowerCase).map { case "unspecified" => RdfVersion.RDF_VERSION_UNSPECIFIED @@ -76,6 +94,7 @@ case class SparqlJellySerializationOptions( checkTableSize("--opt.max-datatype-table-size", `opt.maxDatatypeTableSize`, 0) val options = base.getOrElse(defaultOptions).clone() `opt.streamName`.foreach(options.setStreamName) + streamType.foreach(options.setStreamType) rdfVersion.foreach(options.setRdfVersion) `opt.maxNameTableSize`.foreach(options.setMaxNameTableSize) `opt.maxPrefixTableSize`.foreach(options.setMaxPrefixTableSize) @@ -84,6 +103,6 @@ case class SparqlJellySerializationOptions( /** Whether any of the --opt.* options were set. */ def isAnySet: Boolean = - `opt.streamName`.isDefined || `opt.rdfVersion`.isDefined || + `opt.streamName`.isDefined || `opt.streamType`.isDefined || `opt.rdfVersion`.isDefined || `opt.maxNameTableSize`.isDefined || `opt.maxPrefixTableSize`.isDefined || `opt.maxDatatypeTableSize`.isDefined diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlMetricsPrinter.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlMetricsPrinter.scala index 23f1614..e897c6c 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlMetricsPrinter.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlMetricsPrinter.scala @@ -16,10 +16,20 @@ import scala.jdk.CollectionConverters.* * With the count statistic, lookup entries and column values are counted one by one (a packed * lookup entry holds many values). With the size statistic, their serialized size is summed. */ -final class SparqlFrameInfo(val frameIndex: Long, val metadata: Map[String, ByteString])(using +final class SparqlFrameInfo( + val frameIndex: Long, + val metadata: Map[String, ByteString], + // In a PUNCTUATED stream, the index of the result set the frame belongs to + val resultSetIndex: Option[Long] = None, +)(using statCollector: FrameInfo.StatisticCollector, ): var frameCount: Long = 1 + // Number of result sets that start in these frames + var resultSetCount: Long = 0 + // In a PUNCTUATED stream, the variables or the ASK result of the result set that starts in this + // frame, if any + var resultSetHeader: Seq[(String, YamlValue)] = Seq() // Error message of the last trailer that reported one var trailerError: Option[String] = None private object stat: @@ -39,6 +49,7 @@ final class SparqlFrameInfo(val frameIndex: Long, val metadata: Map[String, Byte def +=(other: SparqlFrameInfo): SparqlFrameInfo = this.frameCount += 1 + this.resultSetCount += other.resultSetCount if other.trailerError.isDefined then this.trailerError = other.trailerError this.stat.frame += other.stat.frame this.stat.row += other.stat.row @@ -72,7 +83,13 @@ final class SparqlFrameInfo(val frameIndex: Long, val metadata: Map[String, Byte if isCount then columns.iterator.map(valueCount).sum else measureAll(columns) - def processFrame(frame: SparqlResultsFrame): Unit = + /** @param resultSetStart + * whether this is the first frame of a result set + */ + def processFrame(frame: SparqlResultsFrame, resultSetStart: Boolean = false): Unit = + if resultSetStart then + resultSetCount += 1 + if resultSetIndex.isDefined then resultSetHeader = SparqlMetricsPrinter.resultSetHeader(frame) stat.frame += statCollector.measure(frame) stat.row += frame.getRowCount Option(frame.getOptions).foreach(o => stat.option += statCollector.measure(o)) @@ -127,14 +144,30 @@ end SparqlFrameInfo */ object SparqlMetricsPrinter: - /** Prints what the first frame says about the whole stream: the options, and the variables or the - * ASK result. + /** The variables or the ASK result, from the first frame of a result set. */ + def resultSetHeader(frame: SparqlResultsFrame): Seq[(String, YamlValue)] = + Option(frame.getAskResult) match + case Some(ask) => + Seq("ask_result" -> YamlBool(ask.getValue)) + case None => + val vars = frame.getVariables.asScala.map(v => YamlListElem(YamlString(v.getName))) + Seq("variables" -> YamlList(vars.toSeq)) + + private def isPunctuated(firstFrame: SparqlResultsFrame): Boolean = + firstFrame.getOptions.getStreamType == SparqlStreamType.PUNCTUATED + + /** Prints what the first frame says about the whole stream: the options, and, if the stream has + * only one result set, its variables or the ASK result. */ private def printHeader(firstFrame: SparqlResultsFrame, o: OutputStream): Unit = val options = firstFrame.getOptions val header = Seq( "stream_options" -> YamlMap( "stream_name" -> YamlString(options.getStreamName), + "stream_type" -> YamlEnum( + Option(options.getStreamType).map(_.toString).getOrElse("UNKNOWN"), + options.getStreamTypeValue, + ), "rdf_version" -> YamlEnum( Option(options.getRdfVersion).map(_.toString).getOrElse("UNKNOWN"), options.getRdfVersionValue, @@ -144,12 +177,7 @@ object SparqlMetricsPrinter: "max_datatype_table_size" -> YamlInt(options.getMaxDatatypeTableSize), "version" -> YamlInt(options.getVersion), ), - ) ++ (Option(firstFrame.getAskResult) match - case Some(ask) => - Seq("ask_result" -> YamlBool(ask.getValue)) - case None => - val vars = firstFrame.getVariables.asScala.map(v => YamlListElem(YamlString(v.getName))) - Seq("variables" -> YamlList(vars.toSeq))) + ) ++ (if isPunctuated(firstFrame) then Seq() else resultSetHeader(firstFrame)) for (key, value) <- header do o.write(YamlDocBuilder.build(YamlMap(key -> value)).getString.getBytes) o.write(System.lineSeparator().getBytes) @@ -166,6 +194,8 @@ object SparqlMetricsPrinter: val yamlFrame = YamlListElem( YamlMap( Seq("frame_index" -> YamlLong(frame.frameIndex)) ++ + frame.resultSetIndex.map("result_set_index" -> YamlLong(_)) ++ + frame.resultSetHeader ++ formatMetadata(frame.metadata).map("metadata" -> _) ++ frame.format()*, ), @@ -183,7 +213,12 @@ object SparqlMetricsPrinter: val sum = iterator.reduce((a, b) => a += b) // Not printing metadata in this case, as there is no upper bound on the number of frames // and thus on the size of the collected metadata. - val stats = YamlMap(Seq("frame_count" -> YamlLong(sum.frameCount)) ++ sum.format()*) + val stats = YamlMap( + Seq("frame_count" -> YamlLong(sum.frameCount)) ++ + (if isPunctuated(firstFrame) then Seq("result_set_count" -> YamlLong(sum.resultSetCount)) + else Seq()) ++ + sum.format()*, + ) o.write(YamlDocBuilder.build(YamlMap("frames" -> stats)).getString.getBytes) private def formatMetadata(metadata: Map[String, ByteString]): Option[YamlMap] = diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlResultSetReader.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlResultSetReader.scala new file mode 100644 index 0000000..e493a15 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlResultSetReader.scala @@ -0,0 +1,114 @@ +package eu.neverblink.jelly.cli.command.sparql.util + +import eu.neverblink.jelly.cli.CriticalException +import eu.neverblink.jelly.convert.jena.sparql.JenaSparqlConverterFactory +import eu.neverblink.jelly.core.proto.v1.sparql.{ + SparqlResultsFrame, + SparqlResultsOptions, + SparqlStreamType, +} +import eu.neverblink.jelly.core.sparql.{JellySparqlOptions, SparqlResultsHandler} +import org.apache.jena.graph.Node +import org.apache.jena.sparql.core.Var + +import scala.annotation.tailrec +import scala.collection.mutable +import scala.jdk.CollectionConverters.* + +/** One result set of a Jelly-SPARQL stream. */ +enum SparqlResultSet: + case Select(variables: IndexedSeq[Var], rows: Iterator[Array[Node]]) + case Ask(value: Boolean) + +/** Reads the result sets of a Jelly-SPARQL stream one by one. + * + * Unlike Jena's reader, this one pulls frames from the stream only when it needs them, so it can + * be read side by side with another stream. + * + * @throws CriticalException + * when it reaches a trailer saying that a result set is incomplete + */ +final class SparqlResultSetReader( + frames: Iterator[SparqlResultsFrame], + supportedOptions: SparqlResultsOptions = JellySparqlOptions.DEFAULT_SUPPORTED_OPTIONS, +) extends Iterator[SparqlResultSet]: + private enum Event: + case Start(variables: IndexedSeq[Var]) + case Row(values: Array[Node]) + case Ask(value: Boolean) + case Trailer(error: String) + + private val events = mutable.Queue[Event]() + + private val handler = new SparqlResultsHandler[Node]: + override def handleVariables(variables: java.util.List[String]): Unit = + events += Event.Start(variables.asScala.map(Var.alloc).toIndexedSeq) + override def createRowBuffer(size: Int): Array[Node] = new Array[Node](size) + // The decoder reuses the row buffer, so it must be copied + override def handleRow(row: Array[Node]): Unit = events += Event.Row(row.clone()) + override def handleAskResult(value: Boolean): Unit = events += Event.Ask(value) + override def handleTrailer(error: String): Unit = events += Event.Trailer(error) + + private val decoder = JenaSparqlConverterFactory.getInstance().decoder( + handler, + supportedOptions.clone().setStreamType(SparqlStreamType.PUNCTUATED), + ) + + // Rows of the current result set, which must be skipped before the next one starts + private var rows: Iterator[Array[Node]] = Iterator.empty + + /** Whether the stream is PUNCTUATED. Known once [[hasNext]] was called. */ + def isPunctuated: Boolean = + Option(decoder.getSparqlOptions).exists(_.getStreamType == SparqlStreamType.PUNCTUATED) + + /** Reads frames until the decoder reports something, or the stream ends. */ + private def peek: Option[Event] = + while events.isEmpty && frames.hasNext do decoder.ingestFrame(frames.next()) + events.headOption + + private def checkError(error: String): Unit = + if error.nonEmpty then + throw CriticalException(s"The producer could not complete the result set: $error") + + @tailrec + override def hasNext: Boolean = + rows.foreach(_ => ()) + peek match + case Some(Event.Trailer(error)) => + // The end of the previous result set + events.dequeue() + checkError(error) + hasNext + case Some(Event.Start(_) | Event.Ask(_)) => true + // The decoder reports rows only after the variables + case Some(Event.Row(_)) => throw IllegalStateException("Row outside of a result set") + case None => false + + override def next(): SparqlResultSet = + if !hasNext then throw java.util.NoSuchElementException() + events.dequeue() match + case Event.Start(variables) => + rows = rowIterator() + SparqlResultSet.Select(variables, rows) + case Event.Ask(value) => SparqlResultSet.Ask(value) + case e => throw IllegalStateException(s"Unexpected $e") + + /** In a PUNCTUATED stream, a trailer ends the result set. In a FLAT stream, it may be followed by + * the repeated stream options and more rows of the same result set. + */ + private def rowIterator(): Iterator[Array[Node]] = new Iterator[Array[Node]]: + override def hasNext: Boolean = peek match + case Some(Event.Row(_)) => true + case Some(Event.Trailer(error)) => + checkError(error) + if isPunctuated then false + else + events.dequeue() + hasNext + case _ => false + + override def next(): Array[Node] = + if !hasNext then throw java.util.NoSuchElementException() + events.dequeue() match + case Event.Row(values) => values + case e => throw IllegalStateException(s"Unexpected $e") diff --git a/src/main/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverter.scala b/src/main/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverter.scala index ae9ff8c..c35ca52 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverter.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverter.scala @@ -2,8 +2,12 @@ package eu.neverblink.jelly.cli.util.jena import eu.neverblink.jelly.cli.* import eu.neverblink.jelly.cli.command.rdf.util.JellyUtil -import eu.neverblink.jelly.cli.command.sparql.util.SparqlJellyUtil -import eu.neverblink.jelly.cli.util.jena.riot.JellyWriterUtil +import eu.neverblink.jelly.cli.command.sparql.util.{ + SparqlJellyUtil, + SparqlResultSet, + SparqlResultSetReader, +} +import eu.neverblink.jelly.cli.util.jena.riot.{JellyFrameWriter, JellyWriterUtil} import eu.neverblink.jelly.convert.jena.JenaConverterFactory import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage import eu.neverblink.jelly.core.JellyOptions @@ -13,7 +17,7 @@ import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsOptions import eu.neverblink.jelly.core.sparql.JellySparqlOptions import org.apache.jena.graph.{Node, Triple} import org.apache.jena.riot.RIOT -import org.apache.jena.riot.resultset.ResultSetReaderRegistry +import org.apache.jena.riot.system.StreamRDF import org.apache.jena.riot.rowset.RowSetWriterRegistry import org.apache.jena.sparql.core.{Quad, Var} import org.apache.jena.sparql.engine.binding.Binding @@ -119,12 +123,16 @@ object RdfSparqlConverter: } (inputOptions, vars, bindings) - /** Converts a Jelly-SPARQL result set to a Jelly-RDF stream, one statement per solution. + /** Converts a Jelly-SPARQL stream to a Jelly-RDF stream, one statement per solution. + * + * A FLAT stream gives a stream with frames of `rowsPerFrame` rows. A PUNCTUATED stream gives one + * frame per result set, with the logical type set to GRAPHS (TRIPLES output) or DATASETS + * (otherwise), unless it was set already. * * @param outputOptions * stream options of the output. The physical type may be unspecified. * @param rowsPerFrame - * target number of rows per output frame + * target number of rows per output frame, for FLAT input * @param delimited * whether the output should be delimited */ @@ -136,78 +144,114 @@ object RdfSparqlConverter: delimited: Boolean, ): Unit = SparqlJellyUtil.translateErrors { - convertSparqlToRdf(inputStream, outputStream, outputOptions, rowsPerFrame, delimited) + val resultSets = SparqlResultSetReader(SparqlJellyUtil.iterateSparqlStream(inputStream)) + if !resultSets.hasNext then throw CriticalException("Empty input stream") + val first = selectResult(resultSets.next(), 0) + val jellyOpt = withPhysicalType(outputOptions, first.variables.size == 4) + if resultSets.isPunctuated then + if jellyOpt.getLogicalType == LogicalStreamType.UNSPECIFIED then + jellyOpt.setLogicalType( + if jellyOpt.getPhysicalType == PhysicalStreamType.TRIPLES then LogicalStreamType.GRAPHS + else LogicalStreamType.DATASETS, + ) + val writer = JellyFrameWriter(jellyOpt, delimited, outputStream) + writeStatements(first, writer, jellyOpt, Some(0)) + writer.endFrame() + for (resultSet, i) <- resultSets.zipWithIndex do + writeStatements(selectResult(resultSet, i + 1), writer, jellyOpt, Some(i + 1)) + writer.endFrame() + writer.finish() + else + val writer = JellyWriterUtil.createWriter( + jellyOpt, + rowsPerFrame, + enableNamespaceDeclarations = false, + delimited, + outputStream, + ) + writer.start() + writeStatements(first, writer, jellyOpt) + writer.finish() outputStream.flush() } - private def convertSparqlToRdf( - inputStream: InputStream, - outputStream: OutputStream, - outputOptions: RdfStreamOptions, - rowsPerFrame: Int, - delimited: Boolean, + /** Checks that the result set is a solution sequence that can be made into statements. */ + private def selectResult(resultSet: SparqlResultSet, index: Int): SparqlResultSet.Select = + resultSet match + case _: SparqlResultSet.Ask => + throw CriticalException( + (if index == 0 then "The input is" else s"Result set $index is") + + " an ASK result, which cannot be converted to RDF", + ) + case select: SparqlResultSet.Select => + val vars = select.variables + if vars.size != 3 && vars.size != 4 then + throw CriticalException( + (if index == 0 then "The result set" else s"Result set $index") + + s" must have 3 or 4 variables, but it has ${vars.size}: " + vars.mkString(", "), + ) + select + + /** Writes the solutions of the result set as statements. + * + * @param resultSetIndex + * the index of the result set in a PUNCTUATED stream, for error messages + */ + private def writeStatements( + resultSet: SparqlResultSet.Select, + writer: StreamRDF, + jellyOpt: RdfStreamOptions, + resultSetIndex: Option[Int] = None, ): Unit = - val lang = JellySparqlLanguage.JELLY_SPARQL - val result = ResultSetReaderRegistry.getFactory(lang).create(lang) - .readAny(inputStream, RIOT.getContext.copy()) - if result.isBoolean then - throw CriticalException("The input is an ASK result, which cannot be converted to RDF") - val resultSet = result.getResultSet - val vars = resultSet.getResultVars.asScala.map(Var.alloc).toIndexedSeq - if vars.size != 3 && vars.size != 4 then + val vars = resultSet.variables + val quads = vars.size == 4 + val triples = jellyOpt.getPhysicalType == PhysicalStreamType.TRIPLES + val where = resultSetIndex.map(i => s" of result set $i").getOrElse("") + if quads && triples then throw CriticalException( - s"The result set must have 3 or 4 variables, but it has ${vars.size}: " + - vars.mkString(", "), + s"Result set${resultSetIndex.map(" " + _).getOrElse("")} has 4 variables, but the output is a " + + "TRIPLES stream. Use --opt.physical-type=QUADS or GRAPHS.", ) - val quads = vars.size == 4 - val jellyOpt = withPhysicalType(outputOptions, quads) - val writer = JellyWriterUtil.createWriter( - jellyOpt, - rowsPerFrame, - enableNamespaceDeclarations = false, - delimited, - outputStream, - ) - writer.start() var index = 0L - while resultSet.hasNext do - val binding = resultSet.nextBinding() - def get(i: Int): Node = binding.get(vars(i)) + for row <- resultSet.rows do def required(i: Int): Node = - val node = get(i) + val node = row(i) if node == null then throw CriticalException( - s"Solution $index has no value for ${vars(i)}, so it cannot be made into a statement", + s"Solution $index$where has no value for ${vars(i)}, so it cannot be made into a " + + "statement", ) node val triple = Triple.create(required(0), required(1), required(2)) - if jellyOpt.getPhysicalType == PhysicalStreamType.TRIPLES then + if triples then checkStatement( triple.toString, StatementUtils.isGeneralized(triple), StatementUtils.hasTripleTerms(triple), - index, + s"$index$where", jellyOpt, ) writer.triple(triple) else - val graph = (if quads then Option(get(3)) else None).getOrElse(Quad.defaultGraphIRI) + val graph = (if quads then Option(row(3)) else None).getOrElse(Quad.defaultGraphIRI) val quad = Quad.create(graph, triple) checkStatement( quad.toString, StatementUtils.isGeneralized(quad), StatementUtils.hasTripleTerms(quad), - index, + s"$index$where", jellyOpt, ) writer.quad(quad) index += 1 - writer.finish() /** The output options, with the physical type set to match the number of variables, unless it was * set already. */ - private def withPhysicalType(options: RdfStreamOptions, quads: Boolean): RdfStreamOptions = + private def withPhysicalType( + options: RdfStreamOptions, + quads: Boolean, + ): RdfStreamOptions.Mutable = val jellyOpt = options.clone() jellyOpt.getPhysicalType match case PhysicalStreamType.UNSPECIFIED => @@ -226,16 +270,16 @@ object RdfSparqlConverter: statement: String, generalized: Boolean, tripleTerms: Boolean, - index: Long, + solution: String, opt: RdfStreamOptions, ): Unit = if !opt.getGeneralizedStatements && generalized then throw CriticalException( - s"Solution $index is not a valid RDF statement: $statement. " + + s"Solution $solution is not a valid RDF statement: $statement. " + "Use --opt.generalized-statements=true to allow generalized statements.", ) if !opt.getRdfStar && tripleTerms then throw CriticalException( - s"Solution $index contains a triple term: $statement. " + + s"Solution $solution contains a triple term: $statement. " + "Use --opt.triple-terms=true to allow triple terms.", ) diff --git a/src/main/scala/eu/neverblink/jelly/cli/util/jena/riot/JellyFrameWriter.scala b/src/main/scala/eu/neverblink/jelly/cli/util/jena/riot/JellyFrameWriter.scala new file mode 100644 index 0000000..618a111 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/util/jena/riot/JellyFrameWriter.scala @@ -0,0 +1,85 @@ +package eu.neverblink.jelly.cli.util.jena.riot + +import eu.neverblink.jelly.cli.CriticalException +import eu.neverblink.jelly.convert.jena.JenaConverterFactory +import eu.neverblink.jelly.core.ProtoEncoder +import eu.neverblink.jelly.core.memory.RowBuffer +import eu.neverblink.jelly.core.proto.v1.{PhysicalStreamType, RdfStreamFrame, RdfStreamOptions} +import org.apache.jena.graph.{Node, Triple} +import org.apache.jena.riot.system.StreamRDF +import org.apache.jena.sparql.core.Quad + +import java.io.OutputStream + +/** Writes Jelly-RDF, ending a frame only when [[endFrame]] is called. This is for grouped streams, + * where each frame is one graph or dataset. + * + * Namespace declarations are not written. + * + * @param options + * stream options, with the physical type set + * @param delimited + * whether the output is delimited. If not, there can be only one frame. + */ +final class JellyFrameWriter(options: RdfStreamOptions, delimited: Boolean, out: OutputStream) + extends StreamRDF: + private val buffer = RowBuffer.newLazyImmutable() + private val encoder = JenaConverterFactory.getInstance().encoder( + ProtoEncoder.Params.of(options, false, buffer), + ) + private val physicalType = options.getPhysicalType + // Only used in GRAPHS streams + private var currentGraph: Node = null + private var frameCount = 0 + + override def start(): Unit = () + + override def triple(triple: Triple): Unit = physicalType match + case PhysicalStreamType.TRIPLES => + encoder.handleTriple(triple.getSubject, triple.getPredicate, triple.getObject) + case PhysicalStreamType.GRAPHS => + handleGraph(Quad.defaultGraphIRI) + encoder.handleTriple(triple.getSubject, triple.getPredicate, triple.getObject) + case _ => + encoder.handleQuad(triple.getSubject, triple.getPredicate, triple.getObject, null) + + override def quad(quad: Quad): Unit = physicalType match + case PhysicalStreamType.TRIPLES => + throw CriticalException("Cannot write quads to a Jelly TRIPLES stream.") + case PhysicalStreamType.GRAPHS => + handleGraph(quad.getGraph) + encoder.handleTriple(quad.getSubject, quad.getPredicate, quad.getObject) + case _ => + encoder.handleQuad(quad.getSubject, quad.getPredicate, quad.getObject, quad.getGraph) + + private def handleGraph(graph: Node): Unit = + if currentGraph == null || currentGraph != graph && + !(Quad.isDefaultGraph(currentGraph) && Quad.isDefaultGraph(graph)) + then + if currentGraph != null then encoder.handleGraphEnd() + encoder.handleGraphStart(graph) + currentGraph = graph + + /** Writes out everything since the previous frame as one frame, even if that's nothing. */ + def endFrame(): Unit = + if currentGraph != null then + encoder.handleGraphEnd() + currentGraph = null + frameCount += 1 + if frameCount == 1 && buffer.isEmpty then buffer.appendMessage().setOptions(encoder.getOptions) + if !delimited && frameCount > 1 then + throw CriticalException( + "The output has more than one frame, so it cannot be written as non-delimited output.", + ) + val frame = RdfStreamFrame.newInstance() + buffer.forEach(row => frame.addRows(row)) + if delimited then frame.writeDelimitedTo(out) else frame.writeTo(out) + buffer.clear() + + override def base(base: String): Unit = () + + override def version(version: String): Unit = () + + override def prefix(prefix: String, iri: String): Unit = () + + override def finish(): Unit = out.flush() diff --git a/src/test/scala/eu/neverblink/jelly/cli/command/helpers/RdfSparqlTestData.scala b/src/test/scala/eu/neverblink/jelly/cli/command/helpers/RdfSparqlTestData.scala index d401be5..a11aba6 100644 --- a/src/test/scala/eu/neverblink/jelly/cli/command/helpers/RdfSparqlTestData.scala +++ b/src/test/scala/eu/neverblink/jelly/cli/command/helpers/RdfSparqlTestData.scala @@ -84,12 +84,19 @@ object RdfSparqlTestData: /** Writes a SELECT result set with the given variables and rows (JSON terms) as Jelly-SPARQL. */ def writeSparql(vars: Seq[String], rows: Seq[Map[String, String]]): Array[Byte] = + SparqlSerDesSpec.writeJelly(sparqlJson(vars, rows)) + + /** A SELECT result set with the given variables and rows (JSON terms) as SPARQL JSON. */ + def sparqlJson(vars: Seq[String], rows: Seq[Map[String, String]]): String = val bindings = rows.map { row => row.map((k, v) => s"\"$k\": $v").mkString("{ ", ", ", " }") } - val json = s"""{ "head": { "vars": [ ${vars.map(v => s"\"$v\"").mkString(", ")} ] }, - | "results": { "bindings": [ ${bindings.mkString(", ")} ] } }""".stripMargin - SparqlSerDesSpec.writeJelly(json) + s"""{ "head": { "vars": [ ${vars.map(v => s"\"$v\"").mkString(", ")} ] }, + | "results": { "bindings": [ ${bindings.mkString(", ")} ] } }""".stripMargin + + /** A solution of ?s ?p ?o, and optionally ?g, with IRIs and a literal object. */ + def statementRow(s: String, p: String, o: String, g: Option[String] = None): Map[String, String] = + Map("s" -> iriJson(s), "p" -> iriJson(p), "o" -> litJson(o)) ++ g.map("g" -> iriJson(_)) def iriJson(s: String): String = s"""{"type":"uri","value":"http://example.org/$s"}""" def litJson(s: String): String = s"""{"type":"literal","value":"$s"}""" diff --git a/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspectSpec.scala b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspectSpec.scala index e33a1a6..9f37221 100644 --- a/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspectSpec.scala +++ b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspectSpec.scala @@ -40,6 +40,7 @@ class SparqlInspectSpec extends AnyWordSpec with TestFixtureHelper with Matchers val parsed = inspect(writeJelly(selectJson)) val options = parsed.get("stream_options").asInstanceOf[YamlMap] options.get("stream_name") shouldBe "" + options.get("stream_type") shouldBe "FLAT (0)" options.get("max_name_table_size") shouldBe JellySparqlOptions.BIG.getMaxNameTableSize options.get("version") shouldBe 1 options.get("rdf_version") shouldBe "RDF_VERSION_UNSPECIFIED (0)" @@ -56,6 +57,7 @@ class SparqlInspectSpec extends AnyWordSpec with TestFixtureHelper with Matchers frames.get("trailer_count") shouldBe 1 frames.get("ask_result_count") shouldBe 0 frames.containsKey("trailer_error") shouldBe false + frames.containsKey("result_set_count") shouldBe false } "print the variables as a plain YAML list" in { @@ -129,6 +131,36 @@ class SparqlInspectSpec extends AnyWordSpec with TestFixtureHelper with Matchers aggregate(parsed).get("row_count") shouldBe 2 } + "count the result sets of a PUNCTUATED stream, instead of printing the variables" in { + val jelly = writePunctuated(Seq(selectJson, askJson(true), selectX()), valuesPerFrame = 4) + val parsed = inspect(jelly) + parsed.get("stream_options").asInstanceOf[YamlMap].get("stream_type") shouldBe + "PUNCTUATED (1)" + parsed.containsKey("variables") shouldBe false + parsed.containsKey("ask_result") shouldBe false + val frames = aggregate(parsed) + frames.get("result_set_count") shouldBe 3 + frames.get("frame_count") shouldBe readFrames(jelly).size + frames.get("row_count") shouldBe 3 + frames.get("trailer_count") shouldBe 3 + } + + "print the result set of each frame of a PUNCTUATED stream in --per-frame" in { + val jelly = writePunctuated(Seq(selectJson, askJson(true), selectX()), valuesPerFrame = 4) + val frames = perFrame(inspect(jelly, "--per-frame")) + frames.size should be > 3 + val resultSets = frames.map(_.get("result_set_index")) + resultSets.distinct shouldBe Seq(0, 1, 2) + // The first frame of each result set has its variables or the ASK result + val firsts = frames.indices.filter(i => i == 0 || resultSets(i) != resultSets(i - 1)) + frames(firsts(0)).get("variables").asInstanceOf[util.List[String]].asScala shouldBe + Seq("s", "label", "num", "bn") + frames(firsts(1)).get("ask_result") shouldBe true + frames(firsts(2)).get("variables").asInstanceOf[util.List[String]].asScala shouldBe Seq("x") + for i <- frames.indices if !firsts.contains(i) do + frames(i).containsKey("variables") shouldBe false + } + "complain about empty input" in { val e = intercept[ExitException](inspect(Array())) e.getCause.getMessage should include("Empty input stream") diff --git a/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesSpec.scala b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesSpec.scala index d9189cb..2f5e2bd 100644 --- a/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesSpec.scala +++ b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlSerDesSpec.scala @@ -3,17 +3,24 @@ package eu.neverblink.jelly.cli.command.sparql import eu.neverblink.jelly.cli.* import eu.neverblink.jelly.cli.command.helpers.TestFixtureHelper import eu.neverblink.jelly.cli.command.sparql.util.{SparqlFormat, SparqlJellyUtil} -import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage +import eu.neverblink.jelly.convert.jena.sparql.{ + JellySparqlLanguage, + JenaSparqlConverterFactory, + RowSetReaderJelly, + RowSetWriterJelly, +} import eu.neverblink.jelly.core.proto.v1.RdfVersion -import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsFrame +import eu.neverblink.jelly.core.proto.v1.sparql.{SparqlResultsFrame, SparqlStreamType} import eu.neverblink.jelly.core.sparql.{JellySparqlConstants, JellySparqlOptions} -import org.apache.jena.query.{ResultSet, ResultSetFactory} +import org.apache.jena.query.{ResultSet, ResultSetFactory, ResultSetRewindable} import org.apache.jena.riot.{Lang, RIOT, ResultSetMgr} import org.apache.jena.riot.resultset.{ ResultSetLang, ResultSetReaderRegistry, ResultSetWriterRegistry, } +import org.apache.jena.riot.rowset.RowSetReaderRegistry +import org.apache.jena.sparql.core.Quad import org.apache.jena.sparql.resultset.ResultsCompare import org.scalatest.matchers.should.Matchers import org.scalatest.wordspec.AnyWordSpec @@ -22,6 +29,7 @@ import java.io.{ByteArrayInputStream, ByteArrayOutputStream} import java.nio.charset.StandardCharsets.UTF_8 import java.nio.file.{Files, Path} import java.util.UUID.randomUUID +import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters.* object SparqlSerDesSpec: @@ -74,6 +82,62 @@ object SparqlSerDesSpec: else writer.write(out, result.getResultSet, context) out.toByteArray + /** Writes SPARQL JSON results (SELECT or ASK) as one PUNCTUATED Jelly-SPARQL stream, in order. */ + def writePunctuated( + jsons: Seq[String], + valuesPerFrame: Int = JellySparqlConstants.DEFAULT_MAX_VALUES_PER_FRAME, + ): Array[Byte] = + val out = ByteArrayOutputStream() + val context = RIOT.getContext.copy() + .set(JellySparqlLanguage.SYMBOL_MAX_VALUES_PER_FRAME, valuesPerFrame) + val writer = + RowSetWriterJelly(RowSetWriterJelly.Options(), JenaSparqlConverterFactory.getInstance()) + .resultSetsWriter(out, context) + for json <- jsons do + val result = RowSetReaderRegistry + .getFactory(ResultSetLang.RS_JSON) + .create(ResultSetLang.RS_JSON) + .readAny(ByteArrayInputStream(json.getBytes(UTF_8)), context) + if result.isBoolean then writer.write(result.booleanResult.booleanValue) + else writer.write(result.rowSet) + out.toByteArray + + /** Reads all result sets of a Jelly-SPARQL stream of any type: the value of an ASK result, or the + * solutions. + */ + def readResultSets(bytes: Array[Byte]): Seq[Either[Boolean, ResultSetRewindable]] = + val results = ArrayBuffer[Either[Boolean, ResultSetRewindable]]() + RowSetReaderJelly(RowSetReaderJelly.Options(), JenaSparqlConverterFactory.getInstance()) + .readAll( + ByteArrayInputStream(bytes), + null, + r => + results += ( + if r.isBoolean then Left(r.booleanResult.booleanValue) + else Right(ResultSetFactory.makeRewindable(ResultSet.adapt(r.rowSet))) + ), + ) + results.toSeq + + /** Whether the result sets are the same as the SPARQL JSON results, in the same order. */ + def sameResultSets( + actual: Seq[Either[Boolean, ResultSetRewindable]], + expected: Seq[String], + ): Boolean = + actual.size == expected.size && actual.zip(expected).forall { + case (Left(value), json) => + ResultSetMgr.readBoolean(ByteArrayInputStream(json.getBytes(UTF_8)), ResultSetLang.RS_JSON) + == value + case (Right(rs), json) => + rs.reset() + ResultsCompare.equalsByTermAndOrder(rs, parse(json, ResultSetLang.RS_JSON)) + } + + /** A SELECT result set with one variable, ?x, and one solution per value (JSON terms). */ + def selectX(values: String*): String = + val rows = values.map(v => s"""{ "x": $v }""") + s"""{ "head": { "vars": [ "x" ] }, "results": { "bindings": [ ${rows.mkString(", ")} ] } }""" + /** Reads the frames of a delimited Jelly-SPARQL stream. */ def readFrames(bytes: Array[Byte]): Seq[SparqlResultsFrame] = SparqlJellyUtil.iterateSparqlStream(ByteArrayInputStream(bytes)).toSeq @@ -367,6 +431,128 @@ class SparqlSerDesSpec extends AnyWordSpec with TestFixtureHelper with Matchers: } } + /** Runs `sparql to-jelly` over several files, given as (content, extension) pairs. */ + private def toJellyFiles(files: Seq[(String, String)], args: List[String] = Nil): Array[Byte] = + val paths = files.map { (content, extension) => + val file = Files.createTempFile(tmpDir, randomUUID.toString, extension) + Files.write(file, content.getBytes(UTF_8)) + file + } + try + SparqlToJelly.runTestCommand(List("sparql", "to-jelly") ++ paths.map(_.toString) ++ args) + SparqlToJelly.getOutBytes + finally paths.foreach(Files.deleteIfExists) + + "PUNCTUATED streams" should { + val second = selectX("""{"type":"bnode","value":"b0"}""", """{"type":"literal","value":"1"}""") + val xml = ResultSetMgr.asString(parse(second, ResultSetLang.RS_JSON), ResultSetLang.RS_XML) + + "be written by to-jelly from several files, one result set per file" in { + val jelly = toJellyFiles(Seq(selectJson -> ".srj", askJson(true) -> ".srj", xml -> ".srx")) + readFrames(jelly).head.getOptions.getStreamType shouldBe SparqlStreamType.PUNCTUATED + sameResultSets(readResultSets(jelly), Seq(selectJson, askJson(true), second)) shouldBe true + } + + "be written by to-jelly from one file with --opt.stream-type=punctuated" in { + for json <- Seq(selectJson, askJson(false)) do + val jelly = toJelly(json, ".srj", List("--opt.stream-type=punctuated")) + readFrames(jelly).head.getOptions.getStreamType shouldBe SparqlStreamType.PUNCTUATED + sameResultSets(readResultSets(jelly), Seq(json)) shouldBe true + } + + "keep the other stream options and split the frames in to-jelly" in { + val jelly = toJellyFiles( + Seq(selectJson -> ".srj", second -> ".srj"), + List("--opt.stream-name=many", "--values-per-frame=2", "--quiet"), + ) + val frames = readFrames(jelly) + frames.head.getOptions.getStreamName shouldBe "many" + frames.size should be > 2 + sameResultSets(readResultSets(jelly), Seq(selectJson, second)) shouldBe true + } + + "not be mentioned in the warning about options that default readers do not accept" in { + val (_, err) = withFile(selectJson, ".srj") { f => + SparqlToJelly.runTestCommand( + List("sparql", "to-jelly", f, "--opt.stream-type=punctuated"), + ) + } + err shouldBe empty + } + + "be refused by to-jelly with --opt.stream-type=flat and several files" in { + val e = intercept[ExitException] { + toJellyFiles(Seq(selectJson -> ".srj", second -> ".srj"), List("--opt.stream-type=flat")) + } + e.getCause shouldBe a[InvalidArgument] + e.getCause.getMessage should include("PUNCTUATED") + } + + "be refused by to-jelly with --delimited=false" in { + val e = intercept[ExitException] { + toJellyFiles(Seq(selectJson -> ".srj", second -> ".srj"), List("--delimited=false")) + } + e.getCause shouldBe a[InvalidArgument] + } + + "not be made by to-jelly from several text or Jelly-RDF files" in { + val text = toText(toJelly(selectJson, ".srj")) + val e = intercept[ExitException] { + toJellyFiles(Seq(text -> ".jellys.txt", text -> ".jellys.txt"), List("--quiet")) + } + e.getCause.getMessage should include("can only be read from one file") + } + + "reject an unknown stream type in to-jelly" in { + val e = intercept[ExitException] { + toJelly(selectJson, ".srj", List("--opt.stream-type=wavy")) + } + e.getCause shouldBe a[InvalidArgument] + } + + "be written by from-jelly to one output, one result set after another" in { + val jelly = writePunctuated(Seq(selectJson, askJson(true), second)) + SparqlFromJelly.setStdIn(ByteArrayInputStream(jelly)) + val (json, _) = SparqlFromJelly.runTestCommand(List("sparql", "from-jelly")) + "\"head\"".r.findAllMatchIn(json).size shouldBe 3 + json should include("\"boolean\" : true") + // The CSV header of each SELECT result set marks where it starts + SparqlFromJelly.setStdIn(ByteArrayInputStream(writePunctuated(Seq(selectJson, second)))) + val (csv, _) = + SparqlFromJelly.runTestCommand(List("sparql", "from-jelly", "--out-format", "csv")) + csv.linesIterator.toSeq.filter(_.nonEmpty) should contain inOrder ("s,label,num,bn", "x") + csv should include("http://example.org/c") + } + + "round trip through the text format" in { + val jelly = writePunctuated(Seq(selectJson, askJson(false), second), valuesPerFrame = 3) + val back = fromText(toText(jelly)) + readFrames(back).size shouldBe readFrames(jelly).size + sameResultSets(readResultSets(back), Seq(selectJson, askJson(false), second)) shouldBe true + } + + "be written by from-jelly as Jelly-RDF, one frame per result set" in { + import eu.neverblink.jelly.cli.command.helpers.RdfSparqlTestData.* + import eu.neverblink.jelly.cli.command.rdf.util.JellyUtil + import eu.neverblink.jelly.core.proto.v1.{LogicalStreamType, PhysicalStreamType} + val resultSets = Seq( + sparqlJson(Seq("s", "p", "o", "g"), Seq(statementRow("a", "p", "v", Some("g")))), + sparqlJson(Seq("s", "p", "o"), Seq(statementRow("b", "p", "w"))), + ) + SparqlFromJelly.setStdIn(ByteArrayInputStream(writePunctuated(resultSets))) + SparqlFromJelly.runTestCommand(List("sparql", "from-jelly", "--out-format", "jelly-rdf")) + val rdf = SparqlFromJelly.getOutBytes + val options = rdfStreamOptions(rdf) + options.getPhysicalType shouldBe PhysicalStreamType.QUADS + options.getLogicalType shouldBe LogicalStreamType.DATASETS + JellyUtil.iterateRdfStream(ByteArrayInputStream(rdf)).size shouldBe 2 + readRdf(rdf) shouldBe Set( + Quad.create(iri("g"), iri("a"), iri("p"), lit("v")), + Quad.create(Quad.defaultGraphIRI, iri("b"), iri("p"), lit("w")), + ) + } + } + private def withBytesFile[T](content: Array[Byte], extension: String)(testCode: String => T): T = val file = Files.createTempFile(tmpDir, randomUUID.toString, extension) Files.write(file, content) diff --git a/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidateSpec.scala b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidateSpec.scala index 67fc69f..b21ed18 100644 --- a/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidateSpec.scala +++ b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidateSpec.scala @@ -2,7 +2,12 @@ package eu.neverblink.jelly.cli.command.sparql import eu.neverblink.jelly.cli.* import eu.neverblink.jelly.cli.command.helpers.TestFixtureHelper -import eu.neverblink.jelly.core.proto.v1.sparql.{SparqlResultsFrame, SparqlResultsTrailer} +import eu.neverblink.jelly.core.proto.v1.sparql.{ + SparqlResultsFrame, + SparqlResultsOptions, + SparqlResultsTrailer, + SparqlStreamType, +} import org.scalatest.matchers.should.Matchers import org.scalatest.wordspec.AnyWordSpec @@ -66,7 +71,7 @@ class SparqlValidateSpec extends AnyWordSpec with TestFixtureHelper with Matcher "accept a valid ASK result stream" in { validate(writeJelly(askJson(true))) - validate(writeJelly(askJson(false)), "--require-trailer") + validate(writeJelly(askJson(false))) } "complain about empty input" in { @@ -113,19 +118,26 @@ class SparqlValidateSpec extends AnyWordSpec with TestFixtureHelper with Matcher "the stream options are repeated and match" in { val frames = readFrames(writeJelly(selectJson)) // Concatenating two streams of the same query is allowed - validate(writeFrames(frames ++ frames), "--require-trailer") + validate(writeFrames(frames ++ frames)) } "the stream options are repeated and differ" in { + def withOptions(change: SparqlResultsOptions.Mutable => Unit) = + readFrames(writeJelly(selectJson)).map { f => + val options = f.getOptions.clone() + change(options) + val c = f.clone().setOptions(options) + c.resetCachedSize() + c + } val first = readFrames(writeJelly(selectJson)) - val second = readFrames(writeJelly(selectJson)).map { f => - val c = f.clone().setOptions(f.getOptions.clone().setStreamName("other")) - c.resetCachedSize() - c - } - validateFails(writeFrames(first ++ second)).getMessage should include( - "Later occurrence of stream options in frame 1 does not match the first", - ) + // Allowed, as long as the new options are valid on their own + validate(writeFrames(first ++ withOptions(_.setStreamName("other")))) + validateFails(writeFrames(first ++ withOptions(_.setVersion(0)))) + .getMessage should include("version") + validateFails( + writeFrames(first ++ withOptions(_.setStreamType(SparqlStreamType.PUNCTUATED))), + ).getMessage should include("stream type must be the same") } "the version in the options is 0" in { @@ -142,8 +154,8 @@ class SparqlValidateSpec extends AnyWordSpec with TestFixtureHelper with Matcher "the stream has no trailer" in { val frames = readFrames(writeJelly(selectJson)).map(withTrailer(_, None)) val jelly = writeFrames(frames) - validate(jelly) - validateFails(jelly, "--require-trailer").getMessage should include( + validate(jelly, "--require-trailer=false") + validateFails(jelly).getMessage should include( "does not end with a trailer", ) } @@ -151,8 +163,8 @@ class SparqlValidateSpec extends AnyWordSpec with TestFixtureHelper with Matcher "one of the concatenated streams has no trailer" in { val frames = readFrames(writeJelly(selectJson)) val jelly = writeFrames(frames.map(withTrailer(_, None)) ++ frames) - validate(jelly) - validateFails(jelly, "--require-trailer").getMessage should include( + validate(jelly, "--require-trailer=false") + validateFails(jelly).getMessage should include( "does not end with a trailer", ) } @@ -163,6 +175,117 @@ class SparqlValidateSpec extends AnyWordSpec with TestFixtureHelper with Matcher } } + "check a PUNCTUATED stream" when { + val second = + selectX("""{"type":"bnode","value":"b0"}""", """{"type":"literal","value":"1"}""") + val jelly = writePunctuated(Seq(selectJson, askJson(true), second)) + + "it is valid" in { + validate(jelly) + validate(writePunctuated(Seq(selectJson, second), valuesPerFrame = 2)) + } + + "the last result set has no trailer" in { + val frames = readFrames(jelly) + val noTrailer = writeFrames(frames.init :+ withTrailer(frames.last, None)) + validate(noTrailer, "--require-trailer=false") + validateFails(noTrailer).getMessage should include( + "does not end with a trailer", + ) + } + + "a trailer reports an error" in { + val frames = readFrames(jelly) + val failed = writeFrames( + frames.head +: withTrailer(frames(1), Some("query timeout")) +: frames.drop(2), + ) + val e = validateFails(failed) + e.getMessage should include("result set 1 is incomplete") + e.getMessage should include("query timeout") + } + + "each result set is compared to its own reference file" in { + withFile(selectJson.getBytes(UTF_8), ".srj") { ref1 => + withFile(askJson(true).getBytes(UTF_8), ".srj") { ref2 => + withFile(second.getBytes(UTF_8), ".srj") { ref3 => + validate( + jelly, + "--compare-to-file", + ref1, + "--compare-to-file", + ref2, + "--compare-to-file", + ref3, + ) + val e = validateFails( + jelly, + "--compare-to-file", + ref1, + "--compare-to-file", + ref2, + "--compare-to-file", + ref1, + ) + e.getMessage should include("Expected variables ?s ?label ?num ?bn") + e.getMessage should include("in result set 2") + } + } + } + } + + "the reference files have a different number of result sets" in { + withFile(selectJson.getBytes(UTF_8), ".srj") { ref => + validateFails(jelly, "--compare-to-file", ref).getMessage should include( + "Expected 1 result sets, as in the reference files, but the input stream has more", + ) + validateFails( + writeJelly(selectJson), + "--compare-to-file", + ref, + "--compare-to-file", + ref, + ).getMessage should include( + "Expected 2 result sets, as in the reference files, but got 1", + ) + } + } + + "a Jelly-SPARQL reference file has several result sets" in { + withFile(writePunctuated(Seq(selectJson, askJson(true))), ".jellys") { ref1 => + withFile(second.getBytes(UTF_8), ".srj") { ref2 => + validate(jelly, "--compare-to-file", ref1, "--compare-to-file", ref2) + validateFails(jelly, "--compare-to-file", ref2, "--compare-to-file", ref1) + .getMessage should include("Expected variables ?x") + } + } + } + + "the solutions of a later result set differ" in { + val other = + selectX("""{"type":"bnode","value":"b0"}""", """{"type":"literal","value":"2"}""") + withFile(writePunctuated(Seq(selectJson, askJson(true), other)), ".jellys") { ref => + val e = validateFails(jelly, "--compare-to-file", ref) + e.getMessage should include("Solution 1 in result set 2 does not match") + } + } + + "the same blank node label is used in different result sets" in { + val bnode = selectX("""{"type":"bnode","value":"a"}""") + // Blank nodes are scoped to one result set, so these need not be the same node + withFile( + writePunctuated( + Seq( + selectX("""{"type":"bnode","value":"x"}"""), + selectX("""{"type":"bnode","value":"y"}"""), + ), + ), + ".jellys", + ) { ref => + validate(writePunctuated(Seq(bnode, bnode)), "--compare-to-file", ref) + } + } + } + "check the stream options against an options file" when { "the options match" in { val jelly = writeJelly(selectJson) diff --git a/src/test/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverterSpec.scala b/src/test/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverterSpec.scala index c87042e..49878b3 100644 --- a/src/test/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverterSpec.scala +++ b/src/test/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverterSpec.scala @@ -5,16 +5,20 @@ import eu.neverblink.jelly.cli.command.helpers.RdfSparqlTestData.* import eu.neverblink.jelly.cli.command.helpers.TestFixtureHelper import eu.neverblink.jelly.cli.command.rdf.util.{JellyUtil, RdfJellySerializationOptions} import eu.neverblink.jelly.cli.command.sparql.SparqlSerDesSpec.* -import eu.neverblink.jelly.core.proto.v1.{PhysicalStreamType, RdfStreamOptions, RdfVersion} -import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsOptions +import eu.neverblink.jelly.convert.jena.JenaConverterFactory +import eu.neverblink.jelly.core.JellyOptions +import eu.neverblink.jelly.core.RdfHandler.AnyStatementHandler +import eu.neverblink.jelly.core.proto.v1.* +import eu.neverblink.jelly.core.proto.v1.sparql.{SparqlResultsOptions, SparqlResultsTrailer} import eu.neverblink.jelly.core.sparql.{JellySparqlConstants, JellySparqlOptions} -import org.apache.jena.graph.{NodeFactory, Triple} +import org.apache.jena.graph.{Node, NodeFactory, Triple} import org.apache.jena.sparql.core.Quad import org.scalatest.matchers.should.Matchers import org.scalatest.wordspec.AnyWordSpec import java.io.{ByteArrayInputStream, ByteArrayOutputStream} import java.nio.charset.StandardCharsets.UTF_8 +import scala.jdk.CollectionConverters.* class RdfSparqlConverterSpec extends AnyWordSpec with TestFixtureHelper with Matchers: @@ -43,11 +47,31 @@ class RdfSparqlConverterSpec extends AnyWordSpec with TestFixtureHelper with Mat private def toRdf( jelly: Array[Byte], options: RdfStreamOptions = defaultRdfOptions, + delimited: Boolean = true, ): Array[Byte] = val out = ByteArrayOutputStream() - RdfSparqlConverter.sparqlToRdf(ByteArrayInputStream(jelly), out, options, 256, true) + RdfSparqlConverter.sparqlToRdf(ByteArrayInputStream(jelly), out, options, 256, delimited) out.toByteArray + /** Reads each frame of a Jelly-RDF stream into its own set of quads. */ + private def readRdfFrames(jelly: Array[Byte]): Seq[Set[Quad]] = + var current = Set[Quad]() + val handler = new AnyStatementHandler[Node]: + override def handleTriple(s: Node, p: Node, o: Node): Unit = + current += Quad.create(Quad.defaultGraphIRI, s, p, o) + override def handleQuad(s: Node, p: Node, o: Node, g: Node): Unit = + current += Quad.create(if Quad.isDefaultGraph(g) then Quad.defaultGraphIRI else g, s, p, o) + val decoder = JenaConverterFactory.getInstance() + .anyStatementDecoder(handler, JellyOptions.DEFAULT_SUPPORTED_OPTIONS) + JellyUtil.iterateRdfStream(ByteArrayInputStream(jelly)).map { frame => + current = Set() + frame.getRows.forEach(decoder.ingestRow) + current + }.toSeq + + private def rdfFrames(jelly: Array[Byte]): Seq[RdfStreamFrame] = + JellyUtil.iterateRdfStream(ByteArrayInputStream(jelly)).toSeq + "rdfToSparql" should { "turn a TRIPLES stream into ?s ?p ?o solutions" in { val (vars, rows) = readSparql(toSparql(writeRdf(PhysicalStreamType.TRIPLES, triples))) @@ -253,3 +277,92 @@ class RdfSparqlConverterSpec extends AnyWordSpec with TestFixtureHelper with Mat intercept[InvalidJellyFile](toRdf("not Jelly".getBytes(UTF_8))) } } + + "sparqlToRdf with a PUNCTUATED stream" should { + val spo = Seq("s", "p", "o") + val spog = Seq("s", "p", "o", "g") + val ab = sparqlJson(spo, Seq(statementRow("a", "p", "1"), statementRow("b", "p", "2"))) + val empty = sparqlJson(spo, Nil) + val cd = + sparqlJson(spog, Seq(statementRow("c", "p", "3", Some("g")), statementRow("d", "p", "4"))) + def quad(s: String, o: String, g: Option[String] = None) = + Quad.create(g.map(iri).getOrElse(Quad.defaultGraphIRI), iri(s), iri("p"), lit(o)) + val abQuads = Set(quad("a", "1"), quad("b", "2")) + val cdQuads = Set(quad("c", "3", Some("g")), quad("d", "4")) + + "write one frame per result set, as a GRAPHS stream for 3 variables" in { + val jelly = toRdf(writePunctuated(Seq(ab, empty, ab), valuesPerFrame = 1)) + val options = rdfStreamOptions(jelly) + options.getPhysicalType shouldBe PhysicalStreamType.TRIPLES + options.getLogicalType shouldBe LogicalStreamType.GRAPHS + readRdfFrames(jelly) shouldBe Seq(abQuads, Set(), abQuads) + } + + "write an empty first result set as a frame with only the stream options" in { + val jelly = toRdf(writePunctuated(Seq(empty, ab))) + val first = rdfFrames(jelly).head.getRows.asScala.toSeq + first.size shouldBe 1 + first.head.hasOptions shouldBe true + readRdfFrames(jelly) shouldBe Seq(Set(), abQuads) + } + + "write a DATASETS stream when the first result set has 4 variables" in { + val jelly = toRdf(writePunctuated(Seq(cd, ab))) + val options = rdfStreamOptions(jelly) + options.getPhysicalType shouldBe PhysicalStreamType.QUADS + options.getLogicalType shouldBe LogicalStreamType.DATASETS + readRdfFrames(jelly) shouldBe Seq(cdQuads, abQuads) + } + + "end the graphs with each frame of a GRAPHS physical stream" in { + val jelly = toRdf( + writePunctuated(Seq(cd, ab, cd)), + defaultRdfOptions.setPhysicalType(PhysicalStreamType.GRAPHS), + ) + val options = rdfStreamOptions(jelly) + options.getPhysicalType shouldBe PhysicalStreamType.GRAPHS + options.getLogicalType shouldBe LogicalStreamType.DATASETS + rdfFrames(jelly).map(_.getRows.asScala.last.hasGraphEnd) shouldBe Seq(true, true, true) + readRdfFrames(jelly) shouldBe Seq(cdQuads, abQuads, cdQuads) + } + + "keep the logical type it is given" in { + val jelly = toRdf( + writePunctuated(Seq(ab, ab)), + defaultRdfOptions.setLogicalType(LogicalStreamType.SUBJECT_GRAPHS), + ) + rdfStreamOptions(jelly).getLogicalType shouldBe LogicalStreamType.SUBJECT_GRAPHS + } + + "refuse 4 variables after a first result set with 3" in { + val e = intercept[CriticalException](toRdf(writePunctuated(Seq(ab, cd)))) + e.getMessage should include("Result set 1 has 4 variables") + e.getMessage should include("--opt.physical-type") + } + + "reject an ASK result set" in { + val e = intercept[CriticalException](toRdf(writePunctuated(Seq(ab, askJson(true))))) + e.getMessage should include("Result set 1 is an ASK result") + } + + "say which result set a solution that cannot be converted is in" in { + val noPredicate = sparqlJson(spo, Seq(Map("s" -> iriJson("a"), "o" -> litJson("v")))) + val e = intercept[CriticalException](toRdf(writePunctuated(Seq(ab, noPredicate)))) + e.getMessage should include("Solution 0 of result set 1 has no value for ?p") + } + + "refuse to write several frames as non-delimited output" in { + toRdf(writePunctuated(Seq(ab)), delimited = false) + val e = intercept[CriticalException](toRdf(writePunctuated(Seq(ab, ab)), delimited = false)) + e.getMessage should include("more than one frame") + } + + "report a result set that the producer could not complete" in { + val frames = readFrames(writePunctuated(Seq(ab, ab))) + val failed = frames.head.clone() + .setTrailer(SparqlResultsTrailer.newInstance().setError("query timeout")) + failed.resetCachedSize() + val e = intercept[CriticalException](toRdf(writeFrames(failed +: frames.tail))) + e.getMessage should include("query timeout") + } + }