diff --git a/README.md b/README.md index 80a4f9d..77d8399 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,8 @@ jelly-cli rdf inspect --help jelly-cli rdf validate --help jelly-cli sparql to-jelly --help jelly-cli sparql from-jelly --help +jelly-cli sparql inspect --help +jelly-cli sparql validate --help ``` And use the `--debug` option to get more information about any exceptions you encounter. diff --git a/build.sbt b/build.sbt index 65c0880..659c68b 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+67-fa45890d-SNAPSHOT" +lazy val jellyV = "3.7.3+68-d52f3f12-SNAPSHOT" lazy val graalvmV = "25.4.4.1.1" addCommandAlias("fixAll", "scalafixAll; scalafmtAll") diff --git a/src/main/scala/eu/neverblink/jelly/cli/App.scala b/src/main/scala/eu/neverblink/jelly/cli/App.scala index 1f7377f..903b5a4 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/App.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/App.scala @@ -35,4 +35,6 @@ object App extends CommandsEntryPoint: RdfValidate, SparqlFromJelly, SparqlToJelly, + SparqlInspect, + SparqlValidate, ) diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfFromJelly.scala b/src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfFromJelly.scala index d6752e7..cbbb8a8 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfFromJelly.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/command/rdf/RdfFromJelly.scala @@ -8,6 +8,7 @@ import eu.neverblink.jelly.cli.util.args.IndexRange import eu.neverblink.jelly.cli.util.io.ProtoText import eu.neverblink.jelly.cli.util.jena.{ JenaSystemOptions, + RdfSparqlConverter, StreamRdfBatchWriter, StreamRdfCombiningBatchWriter, } @@ -15,6 +16,7 @@ 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.RdfStreamFrame +import eu.neverblink.jelly.core.sparql.JellySparqlConstants import org.apache.jena.graph.{Node, Triple} import org.apache.jena.riot.system.StreamRDF import org.apache.jena.riot.system.StreamRDFWriter @@ -118,6 +120,23 @@ object RdfFromJelly extends RdfSerDesCommand[RdfFromJellyOptions, RdfFormat.Writ case (j: RdfFormat.Jena.BatchWriteable, false) => Some((in, out, opt) => jellyToLang(in, StreamRdfBatchWriter(out, j.jenaLang), j, opt)) case (RdfFormat.JellyText, _) => Some(jellyBinaryToText) + case (RdfFormat.JellySparql, _) => Some(jellyToSparql) + + /** Converts the Jelly-RDF stream to a Jelly-SPARQL result set of ?s ?p ?o (?g), with the default + * options. To set them, use sparql to-jelly --in-format jelly-rdf. + */ + private def jellyToSparql( + inputStream: InputStream, + outputStream: OutputStream, + opt: RdfFromJellyOptions, + ): Unit = + RdfSparqlConverter.rdfToSparql( + inputStream, + outputStream, + RdfSparqlConverter.defaultSparqlOptions, + JellySparqlConstants.DEFAULT_MAX_VALUES_PER_FRAME, + delimited = true, + ) /** This method reads the Jelly file, rewrites it to specified format and writes it to some output * stream 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 1986d13..8b4a114 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 @@ -5,17 +5,12 @@ import eu.neverblink.jelly.cli.* import eu.neverblink.jelly.cli.command.rdf.util.* import eu.neverblink.jelly.cli.command.rdf.util.RdfFormat.* import eu.neverblink.jelly.cli.util.io.ProtoText -import eu.neverblink.jelly.cli.util.jena.JenaSystemOptions -import eu.neverblink.jelly.cli.util.jena.riot.{JellyStreamWriterGraphs, RiotParserUtil} -import eu.neverblink.jelly.convert.jena.JenaConverterFactory -import eu.neverblink.jelly.convert.jena.riot.{JellyFormatVariant, JellyLanguage, JellyStreamWriter} +import eu.neverblink.jelly.cli.util.jena.{JenaSystemOptions, RdfSparqlConverter} +import eu.neverblink.jelly.cli.util.jena.riot.{JellyWriterUtil, RiotParserUtil} import eu.neverblink.jelly.core.{JellyOptions, RdfProtoDeserializationError} import eu.neverblink.jelly.core.proto.v1.* -import eu.neverblink.jelly.core.utils.IoUtils -import org.apache.jena.riot.system.StreamRDFWriter -import org.apache.jena.riot.RIOT -import java.io.{BufferedReader, FileInputStream, InputStream, InputStreamReader, OutputStream} +import java.io.{BufferedReader, InputStream, InputStreamReader, OutputStream} import java.nio.charset.StandardCharsets.UTF_8 import scala.util.Using @@ -28,6 +23,10 @@ object RdfToJellyPrint extends RdfCommandPrintUtil[RdfFormat.Readable]: "If no output file is specified, the output is written to stdout.\n" + "If an error is detected, the program will exit with a non-zero code.\n" + "Otherwise, the program will exit with code 0.\n" + + "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" + "Note: this command works in a streaming manner and scales well to large files. ", ) @ArgsName("") @@ -79,20 +78,11 @@ object RdfToJelly extends RdfSerDesCommand[RdfToJellyOptions, RdfFormat.Readable val defaultAction: WriteAction = langToJelly(RdfFormat.NQuads, _, _, _) - private def loadOptionsFromFile(filename: String): RdfStreamOptions = - val inputStream = new FileInputStream(filename) - val response = IoUtils.autodetectDelimiting(inputStream) - val frame = - if response.isDelimited then Using(response.newInput())(RdfStreamFrame.parseDelimitedFrom) - else Using(response.newInput())(RdfStreamFrame.parseFrom) - - frame.get.getRows.iterator().next().getOptions - override def doRun(options: RdfToJellyOptions, remainingArgs: RemainingArgs): Unit = if !options.rdfPerformanceOptions.validateTerms.getOrElse(false) then JenaSystemOptions.disableTermValidation() // Infer before touching options - options.optionsFrom.map(loadOptionsFromFile).foreach( + options.optionsFrom.map(JellyUtil.loadOptionsFromFile).foreach( options.jellySerializationOptions.setOptions, ) options.jellySerializationOptions.inferGeneralized( @@ -117,8 +107,23 @@ object RdfToJelly extends RdfSerDesCommand[RdfToJellyOptions, RdfFormat.Readable ): Option[WriteAction] = format match { case f: RdfFormat.Jena.Readable => Some(langToJelly(f, _, _, _)) case f: RdfFormat.JellyText.type => Some(jellyTextToJelly) + case RdfFormat.JellySparql => Some(jellySparqlToJelly) } + /** Convert a Jelly-SPARQL result set of ?s ?p ?o (?g) to Jelly-RDF. */ + private def jellySparqlToJelly( + inputStream: InputStream, + outputStream: OutputStream, + opt: RdfToJellyOptions, + ): Unit = + RdfSparqlConverter.sparqlToRdf( + inputStream, + outputStream, + getOptions.jellySerializationOptions.asRdfStreamOptions, + getOptions.rowsPerFrame, + getOptions.delimited, + ) + /** This method reads the file, rewrites it to Jelly and writes it to some output stream * @param format * Language that should be converted to Jelly @@ -134,61 +139,22 @@ object RdfToJelly extends RdfSerDesCommand[RdfToJellyOptions, RdfFormat.Readable opt: RdfToJellyOptions, ): Unit = val jellyOpt = getOptions.jellySerializationOptions.asRdfStreamOptions - // Configure the writer - val jellyWriter = - if jellyOpt.getPhysicalType == PhysicalStreamType.GRAPHS then - // GRAPHS - JellyStreamWriterGraphs( - JellyFormatVariant - .builder() - .options( - jellyOpt.clone.setLogicalType( - if jellyOpt.getLogicalType == LogicalStreamType.UNSPECIFIED then - LogicalStreamType.FLAT_QUADS - else jellyOpt.getLogicalType, - ), - ) - .frameSize(getOptions.rowsPerFrame) - .enableNamespaceDeclarations(getOptions.enableNamespaceDeclarations) - .isDelimited(getOptions.delimited) - .build(), - out = outputStream, - ) - else - // TRIPLES or QUADS - if jellyOpt.getPhysicalType == PhysicalStreamType.UNSPECIFIED then - if !isQuietMode && isLogicalGrouped(jellyOpt) then - printLine( - "WARNING: Logical type setting ignored because physical type is not set. " + - "Set the physical type to properly pass on the logical type." + - "Use --quiet to silence this warning.", - true, - ) - val writerContext = RIOT.getContext.copy() - .set( - JellyLanguage.SYMBOL_STREAM_OPTIONS, - jellyOpt, - ) - .set(JellyLanguage.SYMBOL_FRAME_SIZE, getOptions.rowsPerFrame) - .set( - JellyLanguage.SYMBOL_ENABLE_NAMESPACE_DECLARATIONS, - getOptions.enableNamespaceDeclarations, - ).set(JellyLanguage.SYMBOL_DELIMITED_OUTPUT, getOptions.delimited) - StreamRDFWriter.getWriterStream( - outputStream, - JellyLanguage.JELLY, - writerContext, - ) - else - // If the physical type is specified, we can just construct the writer - val variant = JellyFormatVariant - .builder() - .options(jellyOpt) - .frameSize(getOptions.rowsPerFrame) - .enableNamespaceDeclarations(getOptions.enableNamespaceDeclarations) - .isDelimited(getOptions.delimited) - .build() - JellyStreamWriter.create(JenaConverterFactory.getInstance(), variant, outputStream) + if jellyOpt.getPhysicalType == PhysicalStreamType.UNSPECIFIED && !isQuietMode && + isLogicalGrouped(jellyOpt) + then + printLine( + "WARNING: Logical type setting ignored because physical type is not set. " + + "Set the physical type to properly pass on the logical type." + + "Use --quiet to silence this warning.", + true, + ) + val jellyWriter = JellyWriterUtil.createWriter( + jellyOpt, + getOptions.rowsPerFrame, + getOptions.enableNamespaceDeclarations, + getOptions.delimited, + outputStream, + ) RiotParserUtil.parse( getOptions.rdfPerformanceOptions.resolveIris, diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/JellyUtil.scala b/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/JellyUtil.scala index 542cd52..314ca15 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/JellyUtil.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/JellyUtil.scala @@ -1,9 +1,11 @@ package eu.neverblink.jelly.cli.command.rdf.util +import eu.neverblink.jelly.cli.util.io.IoUtil import eu.neverblink.jelly.core.utils.IoUtils -import eu.neverblink.jelly.core.proto.v1.RdfStreamFrame +import eu.neverblink.jelly.core.proto.v1.{RdfStreamFrame, RdfStreamOptions} import java.io.InputStream +import scala.util.Using object JellyUtil: /** Reads the Jelly file and returns an iterator of RdfStreamFrame @@ -38,3 +40,14 @@ object JellyUtil: // In this case, we can only read one frame (false, Iterator(RdfStreamFrame.parseFrom(delimitingResponse.newInput))) } + + /** Reads the stream options from the first row of a Jelly-RDF file. + * @param fileName + * file to read + * @return + * the stream options + */ + def loadOptionsFromFile(fileName: String): RdfStreamOptions = + Using.resource(IoUtil.inputStream(fileName)) { is => + iterateRdfStream(is).next().getRows.iterator().next().getOptions + } diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/RdfFormat.scala b/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/RdfFormat.scala index 1403e26..bd70d03 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/RdfFormat.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/command/rdf/util/RdfFormat.scala @@ -1,6 +1,7 @@ package eu.neverblink.jelly.cli.command.rdf.util import eu.neverblink.jelly.convert.jena.riot.JellyLanguage +import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage import org.apache.jena.riot.{Lang, RDFLanguages} sealed trait RdfFormat: @@ -113,8 +114,30 @@ object RdfFormat: override val supportsBaseIri: Boolean = false override val supportsQuads: Boolean = true + /** Jelly-SPARQL result sets of ?s ?p ?o (?g). See + * [[eu.neverblink.jelly.cli.util.jena.RdfSparqlConverter]]. + */ + case object JellySparql extends RdfFormat, RdfFormat.Writeable, RdfFormat.Readable: + override val fullName: String = "Jelly-SPARQL" + override val cliOptions: List[String] = List("jelly-sparql") + val lang: Lang = JellySparqlLanguage.JELLY_SPARQL + override val supportsBaseIri: Boolean = false + override val supportsQuads: Boolean = true + private val rdfFormats: List[RdfFormat] = - List(NQuads, NTriples, JellyBinary, JellyText, Turtle, TriG, RdfProto, Thrift, RdfXml, JsonLd) + List( + NQuads, + NTriples, + JellyBinary, + JellyText, + JellySparql, + Turtle, + TriG, + RdfProto, + Thrift, + RdfXml, + JsonLd, + ) def all: List[RdfFormat] = rdfFormats @@ -139,5 +162,6 @@ object RdfFormat: formatGuessed match { case Some(f: RdfFormat.Jena) => formatGuessed case _ if fileName.endsWith(JellyText.extension) => Some(RdfFormat.JellyText) + case _ if guessType == JellySparql.lang.getContentType => Some(RdfFormat.JellySparql) case _ => None } 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 baf5ab8..f9a08da 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 @@ -2,7 +2,11 @@ package eu.neverblink.jelly.cli.command.sparql import caseapp.* import eu.neverblink.jelly.cli.* +import eu.neverblink.jelly.cli.command.rdf.util.RdfJellySerializationOptions import eu.neverblink.jelly.cli.command.sparql.util.SparqlFormat +import eu.neverblink.jelly.cli.util.jena.RdfSparqlConverter + +import java.io.{InputStream, OutputStream} object SparqlFromJellyPrint: val validFormats: List[SparqlFormat] = SparqlFormat.writeable @@ -49,3 +53,18 @@ object SparqlFromJelly extends SparqlSerDesCommand[SparqlFromJellyOptions]: val outputFormat = resolveFormat(options.outputFormat, options.outputFile) val (inputStream, outputStream) = getIoStreamsFromOptions(inputFile, options.outputFile) convert(SparqlFormat.JellySparql, outputFormat, inputStream, outputStream) + + /** Converts to Jelly-RDF with the default options. To set them, use rdf to-jelly --in-format + * jelly-sparql. + */ + override protected def jellySparqlToRdf( + inputStream: InputStream, + outputStream: OutputStream, + ): Unit = + RdfSparqlConverter.sparqlToRdf( + inputStream, + outputStream, + RdfJellySerializationOptions().asRdfStreamOptions, + rowsPerFrame = 256, + delimited = true, + ) 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 new file mode 100644 index 0000000..556b15e --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspect.scala @@ -0,0 +1,67 @@ +package eu.neverblink.jelly.cli.command.sparql + +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 scala.jdk.CollectionConverters.* + +@HelpMessage( + "Prints statistics about a Jelly-SPARQL stream.\n" + + "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" + + "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 – " + + "use sparql validate for that.", +) +@ArgsName("") +case class SparqlInspectOptions( + @Recurse + common: JellyCommandOptions = JellyCommandOptions(), + @HelpMessage( + "File to write the output statistics to. If not specified, the output is written to stdout.", + ) + @ExtraName("to") outputFile: Option[String] = None, + @HelpMessage( + "Whether to print the statistics per frame (default: false). " + + "If true, the statistics are computed and printed separately for each frame in the stream.", + ) + perFrame: Boolean = false, + @HelpMessage( + "Report the size (in bytes) of frames and other elements, rather than their counts.", + ) + size: Boolean = false, +) extends HasJellyCommandOptions + +object SparqlInspect extends JellyCommand[SparqlInspectOptions]: + + override def names: List[List[String]] = List( + List("sparql", "inspect"), + ) + + override final def group = "sparql" + + override def doRun(options: SparqlInspectOptions, remainingArgs: RemainingArgs): Unit = + val (inputStream, outputStream) = + getIoStreamsFromOptions(remainingArgs.remaining.headOption, options.outputFile) + given FrameInfo.StatisticCollector = + if options.size then FrameInfo.SizeStatistic else FrameInfo.CountStatistic + SparqlJellyUtil.translateErrors { + val frames = SparqlJellyUtil.iterateSparqlStream(inputStream).buffered + if !frames.hasNext then throw CriticalException("Empty input stream") + val firstFrame = frames.head + if firstFrame.getOptions == null then + throw CriticalException("First frame in the input stream does not contain stream options") + val frameInfos = frames.zipWithIndex.map { (frame, i) => + val metadata = frame.getMetadata.asScala.map(e => e.getKey -> e.getValue).toMap + val info = SparqlFrameInfo(i, metadata) + info.processFrame(frame) + info + } + if options.perFrame then + SparqlMetricsPrinter.printPerFrame(firstFrame, frameInfos, outputStream) + else SparqlMetricsPrinter.printAggregate(firstFrame, frameInfos, outputStream) + } 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 c21f2b5..a929182 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 @@ -1,15 +1,13 @@ package eu.neverblink.jelly.cli.command.sparql import caseapp.* -import com.google.protobuf.{InvalidProtocolBufferException, TextFormat} import eu.neverblink.jelly.cli.* -import eu.neverblink.jelly.cli.command.sparql.util.SparqlFormat +import eu.neverblink.jelly.cli.command.sparql.util.{SparqlFormat, SparqlJellyUtil} import eu.neverblink.jelly.cli.util.io.ProtoText -import eu.neverblink.jelly.core.{RdfProtoDeserializationError, RdfProtoSerializationError} import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsFrame -import eu.neverblink.jelly.core.sparql.JellySparqlIoUtils -import org.apache.jena.riot.{RIOT, RiotException} +import org.apache.jena.riot.RIOT import org.apache.jena.riot.resultset.{ResultSetReaderRegistry, ResultSetWriterRegistry} +import org.apache.jena.sparql.util.Context import java.io.{BufferedReader, InputStream, InputStreamReader, OutputStream} import java.nio.charset.StandardCharsets.UTF_8 @@ -45,6 +43,22 @@ abstract class SparqlSerDesCommand[T <: HasJellyCommandOptions: {Parser, Help}] .filter(validFormats.contains) .getOrElse(defaultFormat) + /** Context passed to Jena's result set reader and writer. Commands override this to configure the + * Jelly-SPARQL writer. + */ + protected def jenaContext: Context = RIOT.getContext.copy() + + /** Whether the Jelly-SPARQL text format should be converted to delimited Jelly-SPARQL. */ + protected def delimitedOutput: Boolean = true + + /** Converts Jelly-RDF to Jelly-SPARQL. Commands that read Jelly-RDF override this. */ + protected def jellyRdfToSparql(inputStream: InputStream, outputStream: OutputStream): Unit = + throw CriticalException("This command cannot read Jelly-RDF.") + + /** Converts Jelly-SPARQL to Jelly-RDF. Commands that write Jelly-RDF override this. */ + protected def jellySparqlToRdf(inputStream: InputStream, outputStream: OutputStream): Unit = + throw CriticalException("This command cannot write Jelly-RDF.") + /** 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. @@ -55,32 +69,22 @@ abstract class SparqlSerDesCommand[T <: HasJellyCommandOptions: {Parser, Help}] inputStream: InputStream, outputStream: OutputStream, ): Unit = - try { + SparqlJellyUtil.translateErrors { (from, to) match case (SparqlFormat.JellySparql, SparqlFormat.JellySparqlText) => jellyBinaryToText(inputStream, outputStream) case (SparqlFormat.JellySparqlText, SparqlFormat.JellySparql) => jellyTextToBinary(inputStream, outputStream) + case (SparqlFormat.JellyRdf, SparqlFormat.JellySparql) => + jellyRdfToSparql(inputStream, outputStream) + case (SparqlFormat.JellySparql, SparqlFormat.JellyRdf) => + jellySparqlToRdf(inputStream, outputStream) case (f: SparqlFormat.Jena, t: SparqlFormat.Jena) => jenaConvert(f, t, inputStream, outputStream) case _ => throw CriticalException(f"Conversion from $from to $to is not supported.") outputStream.flush() - } catch - // The Jelly RowSet reader wraps I/O errors (including protobuf ones) in a RiotException, - // so unwrap it to report a malformed Jelly file the same way the rdf commands do. - case e: RiotException => - e.getCause match - case cause: InvalidProtocolBufferException => throw InvalidJellyFile(cause) - case _ => throw JenaRiotException(e) - case e: InvalidProtocolBufferException => - throw InvalidJellyFile(e) - case e: RdfProtoDeserializationError => - throw JellyDeserializationError(e.getMessage) - case e: RdfProtoSerializationError => - throw JellySerializationError(e.getMessage) - case e: TextFormat.ParseException => - throw InvalidJellyFile(e) + } private def jenaConvert( from: SparqlFormat.Jena, @@ -88,7 +92,7 @@ abstract class SparqlSerDesCommand[T <: HasJellyCommandOptions: {Parser, Help}] inputStream: InputStream, outputStream: OutputStream, ): Unit = - val context = RIOT.getContext.copy() + 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) @@ -99,13 +103,7 @@ abstract class SparqlSerDesCommand[T <: HasJellyCommandOptions: {Parser, Help}] private val frameCommentPrefix = "# Frame" private def jellyBinaryToText(inputStream: InputStream, outputStream: OutputStream): Unit = - val response = JellySparqlIoUtils.autodetectDelimiting(inputStream) - val input = response.newInput() - val frames = - if response.isDelimited then - Iterator.continually(SparqlResultsFrame.parseDelimitedFrom(input)).takeWhile(_ != null) - else Iterator(SparqlResultsFrame.parseFrom(input)) - for (frame, i) <- frames.zipWithIndex do + for (frame, i) <- SparqlJellyUtil.iterateSparqlStream(inputStream).zipWithIndex do outputStream.write(f"$frameCommentPrefix $i\n".getBytes(UTF_8)) val text = ProtoText.print(SparqlResultsFrame.getDescriptor, frame.toByteArray) outputStream.write(text.getBytes(UTF_8)) @@ -119,8 +117,17 @@ abstract class SparqlSerDesCommand[T <: HasJellyCommandOptions: {Parser, Help}] "NEVER use it in production.\nUse --quiet to silence this warning.", true, ) + var frameCount = 0 def writeFrame(text: String): Unit = - ProtoText.parse(SparqlResultsFrame.getDescriptor, text).writeDelimitedTo(outputStream) + frameCount += 1 + val frame = ProtoText.parse(SparqlResultsFrame.getDescriptor, text) + if delimitedOutput then frame.writeDelimitedTo(outputStream) + else if frameCount == 1 then frame.writeTo(outputStream) + // Concatenated non-delimited frames would be read back as one frame with merged columns + else + throw CriticalException( + "The input has more than one frame, so it cannot be written as non-delimited output.", + ) Using.resource(BufferedReader(InputStreamReader(inputStream, UTF_8))) { reader => val buffer = StringBuilder() 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 1bb02d0..62fbc69 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 @@ -2,7 +2,15 @@ package eu.neverblink.jelly.cli.command.sparql import caseapp.* import eu.neverblink.jelly.cli.* -import eu.neverblink.jelly.cli.command.sparql.util.SparqlFormat +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.core.RdfProtoDeserializationError +import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsOptions +import eu.neverblink.jelly.core.sparql.{JellySparqlConstants, JellySparqlOptions} +import org.apache.jena.sparql.util.Context + +import java.io.{InputStream, OutputStream} object SparqlToJellyPrint: val validFormats: List[SparqlFormat] = SparqlFormat.readable @@ -15,6 +23,11 @@ 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 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. " + + "Unless --opt.rdf-version or --options-from is given, the RDF version of the output is taken " + + "from the input's options: 1.2 if it may contain triple terms, 1.1 otherwise.\n" + "If an error is detected, the program will exit with a non-zero code.\n" + "Otherwise, the program will exit with code 0.", ) @@ -32,6 +45,23 @@ case class SparqlToJellyOptions( SparqlToJellyPrint.helpMsg, ) @ExtraName("in-format") inputFormat: Option[String] = None, + @Recurse + jellySerializationOptions: SparqlJellySerializationOptions = SparqlJellySerializationOptions(), + @HelpMessage( + "Jelly-SPARQL file to copy serialization options from. Options can be overridden with " + + "command line --opt.* options. Default: (unset)", + ) + optionsFrom: Option[String] = None, + @HelpMessage( + "Maximum number of values (cells) in one frame. A frame may end earlier, when its lookup " + + "tables fill up. Default: " + JellySparqlConstants.DEFAULT_MAX_VALUES_PER_FRAME, + ) + valuesPerFrame: Int = JellySparqlConstants.DEFAULT_MAX_VALUES_PER_FRAME, + @HelpMessage( + "Whether the output should be delimited. Setting it to false will force the output to be " + + "a single frame – make sure you know what you are doing. Default: true", + ) + delimited: Boolean = true, ) extends HasJellyCommandOptions object SparqlToJelly extends SparqlSerDesCommand[SparqlToJellyOptions]: @@ -44,8 +74,66 @@ object SparqlToJelly extends SparqlSerDesCommand[SparqlToJellyOptions]: override val defaultFormat: SparqlFormat = SparqlToJellyPrint.defaultFormat + // Set in doRun, before the conversion starts + private var baseOptions: Option[SparqlResultsOptions] = None + private var streamOptions: SparqlResultsOptions = JellySparqlOptions.BIG + + override protected def jenaContext: Context = + super.jenaContext + .set(JellySparqlLanguage.SYMBOL_STREAM_OPTIONS, streamOptions) + .set(JellySparqlLanguage.SYMBOL_MAX_VALUES_PER_FRAME, getOptions.valuesPerFrame) + .set(JellySparqlLanguage.SYMBOL_DELIMITED_OUTPUT, getOptions.delimited) + + 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) + if options.valuesPerFrame < 1 then + throw InvalidArgument( + "--values-per-frame", + options.valuesPerFrame.toString, + Some("Must be at least 1"), + ) + 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) + + override protected def jellyRdfToSparql( + inputStream: InputStream, + outputStream: OutputStream, + ): Unit = + RdfSparqlConverter.rdfToSparql( + inputStream, + outputStream, + inputOptions => + getOptions.jellySerializationOptions.toSparqlResultsOptions( + Some(baseOptions.getOrElse(RdfSparqlConverter.defaultSparqlOptions(inputOptions))), + ), + getOptions.valuesPerFrame, + getOptions.delimited, + ) + + private def checkAndWarnOptions(inputFormat: SparqlFormat): Unit = + if inputFormat == SparqlFormat.JellySparqlText then + if getOptions.jellySerializationOptions.isAnySet || getOptions.optionsFrom.isDefined then + printLine( + "WARNING: Stream options are ignored for jelly-sparql-text input, because the frames " + + "are copied as they are. Use --quiet to silence this warning.", + true, + ) + else + try + JellySparqlOptions.checkCompatibility( + streamOptions, + JellySparqlOptions.DEFAULT_SUPPORTED_OPTIONS, + ) + catch + case e: RdfProtoDeserializationError => + printLine( + "WARNING: Readers with default settings will not accept these stream options: " + + s"${e.getMessage} Use --quiet to silence this warning.", + true, + ) 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 new file mode 100644 index 0000000..e52f403 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidate.scala @@ -0,0 +1,323 @@ +package eu.neverblink.jelly.cli.command.sparql + +import caseapp.* +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.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.sparql.core.Var +import org.apache.jena.sparql.engine.binding.Binding +import org.apache.jena.sparql.util.FmtUtils + +import java.io.InputStream +import scala.collection.mutable +import scala.jdk.CollectionConverters.* +import scala.util.Using + +object SparqlValidatePrint: + val validFormats: List[SparqlFormat] = List( + SparqlFormat.Json, + SparqlFormat.Xml, + SparqlFormat.Csv, + SparqlFormat.Tsv, + SparqlFormat.JellySparql, + ) + lazy val validFormatsString: String = SparqlFormat.validFormatsString(validFormats) + +@HelpMessage( + "Validates a Jelly-SPARQL stream.\nIf no additional options are specified, " + + "only basic validations are performed. You can also validate the stream against " + + "a reference SPARQL results file, check the stream options, its delimiting, and whether " + + "it ends with a trailer.\n" + + "A stream whose trailer reports an error (an incomplete result set) is always invalid.\n" + + "If an error is detected, the program will exit with a non-zero code.\n" + + "Otherwise, the program will exit with code 0.\n" + + "When comparing to a reference file, the solutions must come in the same order, and blank " + + "nodes are compared up to renaming.\n" + + "Note: this command works in a streaming manner, also when comparing: the solutions are " + + "compared one by one, as they are read.", +) +@ArgsName("") +case class SparqlValidateOptions( + @Recurse + common: JellyCommandOptions = JellyCommandOptions(), + @HelpMessage( + "SPARQL results file to compare the input stream to. If not specified, no comparison is done.", + ) + compareToFile: Option[String] = None, + @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: " + + SparqlValidatePrint.validFormatsString, + ) + compareToFormat: Option[String] = None, + @HelpMessage( + "Jelly-SPARQL file with the expected stream options. If not specified, the options are " + + "not checked.", + ) + optionsFile: Option[String] = None, + @HelpMessage( + "Whether the input stream should be checked to be delimited or undelimited. " + + "Possible values: 'either', 'true', 'false'. Default: 'either'.", + ) + delimited: String = "either", + @HelpMessage( + "Whether the stream must end with a trailer, which tells the reader that the result set " + + "is complete. Default: false", + ) + requireTrailer: Boolean = false, +) 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" + + override def doRun(options: SparqlValidateOptions, remainingArgs: RemainingArgs): Unit = + val delimiting = options.delimited match + case "" | "either" => Delimiting.Either + case "true" => Delimiting.Delimited + case "false" => Delimiting.Undelimited + case _ => + throw InvalidArgument( + "--delimited", + options.delimited, + Some("Valid values: true, false, either"), + ) + val expectedOptions = options.optionsFile.map(SparqlJellyUtil.loadOptionsFromFile) + val referenceLang = options.compareToFile.map(referenceFormat(_, 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) + } + + private def validateDelimiting(expected: Delimiting, delimited: Boolean): Unit = + expected match + case Delimiting.Either => () + case Delimiting.Delimited => + if !delimited then + throw CriticalException("Expected delimited input, but the file was not delimited") + case Delimiting.Undelimited => + if delimited then + throw CriticalException("Expected undelimited input, but the file was delimited") + + private def validateOptions( + streamOptions: SparqlResultsOptions, + expectedOptions: Option[SparqlResultsOptions], + ): Unit = + expectedOptions.foreach { expected => + if streamOptions != expected then + throw CriticalException( + s"Stream options do not match the expected options in ${getOptions.optionsFile.get}\n" + + s"Expected: ${printOptions(expected)}\n" + + s"Actual: ${printOptions(streamOptions)}", + ) + } + if streamOptions.getVersion <= 0 then + throw CriticalException( + "The version field in SparqlResultsOptions is <= 0. " + + "This field MUST be set to a positive value.", + ) + + private def printOptions(options: SparqlResultsOptions): String = + ProtoText.print(SparqlResultsOptions.getDescriptor, options.toByteArray) + + /** 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 + */ + private def validateContent( + frames: Iterator[SparqlResultsFrame], + expectedOptions: Option[SparqlResultsOptions], + reference: Option[Reference], + ): Unit = + if !frames.hasNext then throw CriticalException("Empty input stream") + val handler = ComparingHandler(reference) + // 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() + expectedOptions.foreach { e => + supported + .setMaxNameTableSize(e.getMaxNameTableSize.max(supported.getMaxNameTableSize)) + .setMaxPrefixTableSize(e.getMaxPrefixTableSize.max(supported.getMaxPrefixTableSize)) + .setMaxDatatypeTableSize(e.getMaxDatatypeTableSize.max(supported.getMaxDatatypeTableSize)) + } + 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. + 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 + throw CriticalException("The stream does not end with a trailer") + + for (frame, i) <- frames.zipWithIndex do + Option(frame.getOptions) match + case Some(o) if firstOptions.isEmpty => + validateOptions(o, expectedOptions) + firstOptions = Some(o) + case Some(o) => + if !firstOptions.contains(o) then + throw CriticalException( + s"Later occurrence of stream options in frame $i does not match the first", + ) + checkTrailer() + trailerSeen = false + case None if firstOptions.isEmpty => + throw CriticalException("First frame in the input stream does not contain stream options") + case None => () + decoder.ingestFrame(frame) + 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}", + ) + trailerSeen = true + } + 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 + private var variables: IndexedSeq[Var] = IndexedSeq() + // Number of solutions read so far + private var index = 0L + private val blankNodes = BlankNodeMapping() + + private def fileName = getOptions.compareToFile.getOrElse("") + + override def handleVariables(variables: java.util.List[String]): Unit = + 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 => + throw CriticalException( + s"Expected variables ${expected.mkString(" ")}, as in $fileName, " + + s"but got ${this.variables.mkString(" ")}", + ) + case _ => () + + override def createRowBuffer(size: Int): Array[Node] = new Array[Node](size) + + override def handleRow(row: Array[Node]): Unit = + reference match + case Some(Reference.Select(_, rows)) => + if !rows.hasNext then + throw CriticalException( + s"Expected $index solutions, as in $fileName, but the input stream has more", + ) + val expected = rows.next() + val matches = variables.indices.forall { i => + blankNodes.sameTerm(expected.get(variables(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"Actual: ${formatRow(row.toIndexedSeq)}", + ) + case _ => () + index += 1 + + override def handleAskResult(value: Boolean): Unit = + askResult = Some(value) + 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 _ => () + + /** Checks that the reference has no more solutions than the input. */ + def finish(): Unit = reference match + case Some(Reference.Select(_, rows)) if rows.hasNext => + throw CriticalException( + s"Expected ${index + rows.size} solutions, as in $fileName, but got $index", + ) + case _ => () + + private def formatRow(values: Seq[Node]): String = + variables.zip(values).map { (v, n) => + s"$v=${if n == null then "(unbound)" else FmtUtils.stringForNode(n)}" + }.mkString(" ") + + /** Compares terms, with blank nodes matched up to renaming. + * + * As the solutions are compared in order, the first time a blank node label is seen on one side, + * it is paired with the label on the other side. From then on, the pair must always come + * together. + */ + private class BlankNodeMapping: + private val expectedToActual = mutable.HashMap[Node, Node]() + private val actualToExpected = mutable.HashMap[Node, Node]() + + /** Whether the terms are the same. Nulls stand for unbound values. */ + def sameTerm(expected: Node, actual: Node): Boolean = + if expected == null || actual == null then expected eq actual + else if expected.isBlank && actual.isBlank then + expectedToActual.get(expected) match + case Some(paired) => paired == actual + case None if actualToExpected.contains(actual) => false + case None => + expectedToActual(expected) = actual + actualToExpected(actual) = expected + true + else if expected.isTripleTerm && actual.isTripleTerm then + val e = expected.getTriple + val a = actual.getTriple + sameTerm(e.getSubject, a.getSubject) && sameTerm(e.getPredicate, a.getPredicate) && + sameTerm(e.getObject, a.getObject) + else expected == actual + + /** Picks the format of the reference file. */ + private def referenceFormat(fileName: String, formatName: Option[String]): Lang = + val format = formatName match + case Some(name) => + SparqlFormat.find(name).filter(SparqlValidatePrint.validFormats.contains).getOrElse { + throw InvalidFormatSpecified(name, SparqlValidatePrint.validFormatsString) + } + case None => + SparqlFormat.inferFormat(fileName) + .filter(SparqlValidatePrint.validFormats.contains) + .getOrElse { + throw InvalidFormatSpecified("", SparqlValidatePrint.validFormatsString) + } + format match + case f: SparqlFormat.Jena => f.jenaLang + 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) + else + val rs = result.getResultSet + Reference.Select( + rs.getResultVars.asScala.map(Var.alloc).toSeq, + Iterator.continually(rs).takeWhile(_.hasNext).map(_.nextBinding()), + ) diff --git a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlFormat.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlFormat.scala index e4c70af..f1b498e 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlFormat.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlFormat.scala @@ -1,5 +1,6 @@ package eu.neverblink.jelly.cli.command.sparql.util +import eu.neverblink.jelly.convert.jena.riot.JellyLanguage import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage import org.apache.jena.riot.{Lang, RDFLanguages} import org.apache.jena.riot.resultset.ResultSetLang @@ -65,8 +66,16 @@ object SparqlFormat: override val cliOptions: List[String] = List("jelly-sparql-text") val extension = ".jellys.txt" + /** Jelly-RDF, as a result set of ?s ?p ?o (?g). See + * [[eu.neverblink.jelly.cli.util.jena.RdfSparqlConverter]]. + */ + case object JellyRdf extends SparqlFormat.Readable, SparqlFormat.Writeable: + override val fullName: String = "Jelly-RDF" + override val cliOptions: List[String] = List("jelly-rdf") + val lang: Lang = JellyLanguage.JELLY + private val sparqlFormats: List[SparqlFormat] = - List(Json, Xml, Csv, Tsv, Text, JellySparql, JellySparqlText) + List(Json, Xml, Csv, Tsv, Text, JellySparql, JellySparqlText, JellyRdf) def all: List[SparqlFormat] = sparqlFormats @@ -98,6 +107,8 @@ object SparqlFormat: if fileName.endsWith(JellySparqlText.extension) then Some(JellySparqlText) else val guessType = RDFLanguages.guessContentType(fileName) - sparqlFormats.collectFirst { - case f: SparqlFormat.Jena if f.jenaLang.getContentType == guessType => f - } + if guessType == JellyRdf.lang.getContentType then Some(JellyRdf) + else + sparqlFormats.collectFirst { + case f: SparqlFormat.Jena if f.jenaLang.getContentType == guessType => f + } 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 new file mode 100644 index 0000000..aeab8c0 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlJellySerializationOptions.scala @@ -0,0 +1,89 @@ +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.sparql.JellySparqlOptions +import eu.neverblink.jelly.core.utils.RdfVersionUtils + +private val defaultOptions: SparqlResultsOptions = JellySparqlOptions.BIG + +/** Options for serializing in Jelly-SPARQL. + * + * This is the Jelly-SPARQL counterpart of + * [[eu.neverblink.jelly.cli.command.rdf.util.RdfJellySerializationOptions]]. + */ +case class SparqlJellySerializationOptions( + @HelpMessage("Name of the output stream (in metadata). Default: (empty)") + `opt.streamName`: 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 " + + "terms), unspecified. Default: unspecified", + ) + `opt.rdfVersion`: Option[String] = None, + @HelpMessage( + "Maximum size of the name lookup table. Must be at least " + + JellySparqlOptions.MIN_NAME_TABLE_SIZE + ". Default: " + defaultOptions.getMaxNameTableSize, + ) + `opt.maxNameTableSize`: Option[Int] = None, + @HelpMessage( + "Maximum size of the prefix lookup table. Default: " + defaultOptions.getMaxPrefixTableSize, + ) + `opt.maxPrefixTableSize`: Option[Int] = None, + @HelpMessage( + "Maximum size of the datatype lookup table. Default: " + + defaultOptions.getMaxDatatypeTableSize, + ) + `opt.maxDatatypeTableSize`: Option[Int] = None, +): + + private def rdfVersion: Option[RdfVersion] = + `opt.rdfVersion`.map(_.trim.toLowerCase).map { + case "unspecified" => RdfVersion.RDF_VERSION_UNSPECIFIED + case label => + try RdfVersionUtils.rdfVersionFromLabel(label) + catch + case _: IllegalArgumentException => + throw InvalidArgument( + "--opt.rdf-version", + `opt.rdfVersion`.get, + Some("Must be one of: 1.1, 1.2-basic, 1.2, unspecified"), + ) + } + + private def checkTableSize(argument: String, size: Option[Int], minSize: Int): Unit = + size.filter(_ < minSize).foreach { s => + throw InvalidArgument(argument, s.toString, Some(f"Must be at least $minSize")) + } + + /** Builds the stream options, starting from `base` (or the default preset) and applying the + * options set on the command line on top. + * + * @param base + * options to start from, e.g., copied from another Jelly-SPARQL file + * @throws InvalidArgument + * if any of the options is invalid + */ + def toSparqlResultsOptions(base: Option[SparqlResultsOptions] = None): SparqlResultsOptions = + checkTableSize( + "--opt.max-name-table-size", + `opt.maxNameTableSize`, + JellySparqlOptions.MIN_NAME_TABLE_SIZE, + ) + checkTableSize("--opt.max-prefix-table-size", `opt.maxPrefixTableSize`, 0) + checkTableSize("--opt.max-datatype-table-size", `opt.maxDatatypeTableSize`, 0) + val options = base.getOrElse(defaultOptions).clone() + `opt.streamName`.foreach(options.setStreamName) + rdfVersion.foreach(options.setRdfVersion) + `opt.maxNameTableSize`.foreach(options.setMaxNameTableSize) + `opt.maxPrefixTableSize`.foreach(options.setMaxPrefixTableSize) + `opt.maxDatatypeTableSize`.foreach(options.setMaxDatatypeTableSize) + options + + /** Whether any of the --opt.* options were set. */ + def isAnySet: Boolean = + `opt.streamName`.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/SparqlJellyUtil.scala b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlJellyUtil.scala new file mode 100644 index 0000000..ee9b412 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlJellyUtil.scala @@ -0,0 +1,77 @@ +package eu.neverblink.jelly.cli.command.sparql.util + +import com.google.protobuf.{InvalidProtocolBufferException, TextFormat} +import eu.neverblink.jelly.cli.* +import eu.neverblink.jelly.cli.util.io.IoUtil +import eu.neverblink.jelly.core.{RdfProtoDeserializationError, RdfProtoSerializationError} +import eu.neverblink.jelly.core.proto.v1.sparql.{SparqlResultsFrame, SparqlResultsOptions} +import eu.neverblink.jelly.core.sparql.JellySparqlIoUtils +import org.apache.jena.riot.RiotException + +import java.io.InputStream +import scala.util.Using + +/** The Jelly-SPARQL counterpart of [[eu.neverblink.jelly.cli.command.rdf.util.JellyUtil]]. */ +object SparqlJellyUtil: + + /** Reads a Jelly-SPARQL stream and returns an iterator of its frames. + */ + def iterateSparqlStream(inputStream: InputStream): Iterator[SparqlResultsFrame] = + iterateSparqlStreamWithDelimitingInfo(inputStream)._2 + + /** Reads a Jelly-SPARQL stream and returns an iterator of its frames and a boolean saying whether + * the stream is delimited. + */ + def iterateSparqlStreamWithDelimitingInfo( + inputStream: InputStream, + ): (Boolean, Iterator[SparqlResultsFrame]) = + val response = JellySparqlIoUtils.autodetectDelimiting(inputStream) + val input = response.newInput() + if response.isDelimited then + ( + true, + Iterator.continually(SparqlResultsFrame.parseDelimitedFrom(input)).takeWhile(_ != null), + ) + else + val frame = SparqlResultsFrame.parseFrom(input) + // An empty input parses as an empty frame, but it's really an empty stream + (false, if frame.getSerializedSize == 0 then Iterator.empty else Iterator(frame)) + + /** Reads the stream options from the first frame of a Jelly-SPARQL file. + * + * @throws CriticalException + * if the file is empty or its first frame has no options + */ + def loadOptionsFromFile(fileName: String): SparqlResultsOptions = + translateErrors { + Using.resource(IoUtil.inputStream(fileName)) { is => + val frames = iterateSparqlStream(is) + if !frames.hasNext then throw CriticalException(s"The file $fileName is empty") + val options = frames.next().getOptions + if options == null then + throw CriticalException( + s"The first frame in $fileName does not contain stream options", + ) + options + } + } + + /** Runs `body` and turns the errors of Jelly, protobuf, and Jena into the CLI's own exceptions. + */ + def translateErrors[T](body: => T): T = + try body + catch + // The Jelly RowSet reader wraps I/O errors (including protobuf ones) in a RiotException, + // so unwrap it to report a malformed Jelly file the same way the rdf commands do. + case e: RiotException => + e.getCause match + case cause: InvalidProtocolBufferException => throw InvalidJellyFile(cause) + case _ => throw JenaRiotException(e) + case e: InvalidProtocolBufferException => + throw InvalidJellyFile(e) + case e: RdfProtoDeserializationError => + throw JellyDeserializationError(e.getMessage) + case e: RdfProtoSerializationError => + throw JellySerializationError(e.getMessage) + case e: TextFormat.ParseException => + throw InvalidJellyFile(e) 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 new file mode 100644 index 0000000..23f1614 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/command/sparql/util/SparqlMetricsPrinter.scala @@ -0,0 +1,198 @@ +package eu.neverblink.jelly.cli.command.sparql.util + +import com.google.protobuf.ByteString +import eu.neverblink.jelly.cli.command.rdf.util.FrameInfo +import eu.neverblink.jelly.cli.util.io.YamlDocBuilder +import eu.neverblink.jelly.cli.util.io.YamlDocBuilder.* +import eu.neverblink.jelly.core.proto.v1.RdfLookupEntryPacked +import eu.neverblink.jelly.core.proto.v1.sparql.* +import eu.neverblink.protoc.java.runtime.ProtoMessage + +import java.io.OutputStream +import scala.jdk.CollectionConverters.* + +/** Statistics of a single frame of a Jelly-SPARQL stream, or of several frames added together. + * + * 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 + statCollector: FrameInfo.StatisticCollector, +): + var frameCount: Long = 1 + // Error message of the last trailer that reported one + var trailerError: Option[String] = None + private object stat: + var frame: Long = 0 + var row: Long = 0 + var option: Long = 0 + var variable: Long = 0 + var name: Long = 0 + var prefix: Long = 0 + var datatype: Long = 0 + var iri: Long = 0 + var bnode: Long = 0 + var literal: Long = 0 + var poly: Long = 0 + var askResult: Long = 0 + var trailer: Long = 0 + + def +=(other: SparqlFrameInfo): SparqlFrameInfo = + this.frameCount += 1 + if other.trailerError.isDefined then this.trailerError = other.trailerError + this.stat.frame += other.stat.frame + this.stat.row += other.stat.row + this.stat.option += other.stat.option + this.stat.variable += other.stat.variable + this.stat.name += other.stat.name + this.stat.prefix += other.stat.prefix + this.stat.datatype += other.stat.datatype + this.stat.iri += other.stat.iri + this.stat.bnode += other.stat.bnode + this.stat.literal += other.stat.literal + this.stat.poly += other.stat.poly + this.stat.askResult += other.stat.askResult + this.stat.trailer += other.stat.trailer + this + + private def isCount: Boolean = statCollector == FrameInfo.CountStatistic + + private def measureAll(messages: Iterable[ProtoMessage[?]]): Long = + messages.iterator.map(statCollector.measure).sum + + private def measureLookup(entries: Iterable[RdfLookupEntryPacked]): Long = + if isCount then entries.iterator.map(_.getValues.size.toLong).sum + else measureAll(entries) + + /** Measures a list of columns: the number of values in them, or their size. */ + private def measureColumns[C <: ProtoMessage[?]]( + columns: Iterable[C], + valueCount: C => Long, + ): Long = + if isCount then columns.iterator.map(valueCount).sum + else measureAll(columns) + + def processFrame(frame: SparqlResultsFrame): Unit = + stat.frame += statCollector.measure(frame) + stat.row += frame.getRowCount + Option(frame.getOptions).foreach(o => stat.option += statCollector.measure(o)) + stat.variable += measureAll(frame.getVariables.asScala) + stat.name += measureLookup(frame.getNames.asScala) + stat.prefix += measureLookup(frame.getPrefixes.asScala) + stat.datatype += measureLookup(frame.getDatatypes.asScala) + stat.iri += measureColumns(frame.getIriColumns.asScala, iriValues) + stat.bnode += measureColumns(frame.getBnodeColumns.asScala, bnodeValues) + stat.literal += measureColumns(frame.getLiteralColumns.asScala, literalValues) + stat.poly += measureColumns(frame.getPolyColumns.asScala, polyValues) + Option(frame.getAskResult).foreach(r => stat.askResult += statCollector.measure(r)) + Option(frame.getTrailer).foreach { t => + stat.trailer += statCollector.measure(t) + if t.getError.nonEmpty then trailerError = Some(t.getError) + } + + private def iriValues(c: SparqlIriColumn): Long = c.getNameIds.size + private def bnodeValues(c: SparqlBnodeColumn): Long = c.getValues.size + private def literalValues(c: SparqlLiteralColumn): Long = c.getLexValues.size + private def polyValues(c: SparqlPolyColumn): Long = + Option(c.getIris).map(iriValues).getOrElse(0L) + + Option(c.getBnodes).map(bnodeValues).getOrElse(0L) + + Option(c.getLiterals).map(literalValues).getOrElse(0L) + + c.getTripleTerms.size + + def format(): Seq[(String, YamlValue)] = + val name = statCollector.name() + Seq( + "frame_" + name -> YamlLong(stat.frame), + // The number of solutions, whatever the statistic + "row_count" -> YamlLong(stat.row), + "option_" + name -> YamlLong(stat.option), + "variable_" + name -> YamlLong(stat.variable), + "name_" + name -> YamlLong(stat.name), + "prefix_" + name -> YamlLong(stat.prefix), + "datatype_" + name -> YamlLong(stat.datatype), + "iri_value_" + name -> YamlLong(stat.iri), + "bnode_value_" + name -> YamlLong(stat.bnode), + "literal_value_" + name -> YamlLong(stat.literal), + "poly_value_" + name -> YamlLong(stat.poly), + "ask_result_" + name -> YamlLong(stat.askResult), + "trailer_" + name -> YamlLong(stat.trailer), + ) ++ trailerError.map(e => "trailer_error" -> YamlString(e)) + +end SparqlFrameInfo + +/** Prints the statistics of a Jelly-SPARQL stream as YAML. + * + * This is the Jelly-SPARQL counterpart of + * [[eu.neverblink.jelly.cli.command.rdf.util.MetricsPrinter]]. + */ +object SparqlMetricsPrinter: + + /** Prints what the first frame says about the whole stream: the options, and the 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), + "rdf_version" -> YamlEnum( + Option(options.getRdfVersion).map(_.toString).getOrElse("UNKNOWN"), + options.getRdfVersionValue, + ), + "max_name_table_size" -> YamlInt(options.getMaxNameTableSize), + "max_prefix_table_size" -> YamlInt(options.getMaxPrefixTableSize), + "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))) + for (key, value) <- header do + o.write(YamlDocBuilder.build(YamlMap(key -> value)).getString.getBytes) + o.write(System.lineSeparator().getBytes) + + def printPerFrame( + firstFrame: SparqlResultsFrame, + iterator: Iterator[SparqlFrameInfo], + o: OutputStream, + ): Unit = + printHeader(firstFrame, o) + val builder = YamlDocBuilder.build(YamlMap("frames" -> YamlBlank())) + o.write(builder.getString.getBytes) + iterator.foreach { frame => + val yamlFrame = YamlListElem( + YamlMap( + Seq("frame_index" -> YamlLong(frame.frameIndex)) ++ + formatMetadata(frame.metadata).map("metadata" -> _) ++ + frame.format()*, + ), + ) + o.write(YamlDocBuilder.build(yamlFrame, builder.currIndent).getString.getBytes) + o.write(System.lineSeparator().getBytes) + } + + def printAggregate( + firstFrame: SparqlResultsFrame, + iterator: Iterator[SparqlFrameInfo], + o: OutputStream, + ): Unit = + printHeader(firstFrame, o) + 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()*) + o.write(YamlDocBuilder.build(YamlMap("frames" -> stats)).getString.getBytes) + + private def formatMetadata(metadata: Map[String, ByteString]): Option[YamlMap] = + if metadata.isEmpty then None + else + Some( + YamlMap( + metadata.map { case (k, v) => + k -> YamlString(v.toByteArray.map("%02x" format _).mkString) + }.toSeq*, + ), + ) diff --git a/src/main/scala/eu/neverblink/jelly/cli/util/io/YamlDocBuilder.scala b/src/main/scala/eu/neverblink/jelly/cli/util/io/YamlDocBuilder.scala index 3a122fd..bcb5dcf 100644 --- a/src/main/scala/eu/neverblink/jelly/cli/util/io/YamlDocBuilder.scala +++ b/src/main/scala/eu/neverblink/jelly/cli/util/io/YamlDocBuilder.scala @@ -21,11 +21,11 @@ class YamlDocBuilder(var currIndent: Int = 0): sb.append(v.toString) case YamlEnum(v, i) => sb.append(f"${v} (${i})") + case YamlList(v) if v.isEmpty => + sb.append("[]") case YamlList(v) => - v.zipWithIndex.foreach { (e, index) => - this.build(e, indent) - if e != v.last then sb.append(System.lineSeparator()) - } + // Each element starts on a new line on its own + v.foreach(e => this.build(e, indent)) case YamlListElem(v) => sb.append(System.lineSeparator()) sb.append(" " * indent).append("- ") 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 new file mode 100644 index 0000000..ae9ff8c --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverter.scala @@ -0,0 +1,241 @@ +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.convert.jena.JenaConverterFactory +import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage +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 +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.rowset.RowSetWriterRegistry +import org.apache.jena.sparql.core.{Quad, Var} +import org.apache.jena.sparql.engine.binding.Binding +import org.apache.jena.sparql.exec.RowSetStream + +import java.io.{InputStream, OutputStream} +import scala.collection.mutable.ArrayBuffer +import scala.jdk.CollectionConverters.* + +/** Converts between Jelly-RDF streams and Jelly-SPARQL result sets. + * + * A statement becomes a solution of ?s ?p ?o (triples) or ?s ?p ?o ?g (quads), where ?g is unbound + * for the default graph. The other way round, the variables are taken by position: subject, + * predicate, object, and, if there is a fourth one, graph. + */ +object RdfSparqlConverter: + + /** Default options of the Jelly-SPARQL output, for the given Jelly-RDF input options. + * + * This is the BIG preset, with the RDF version inferred from the input. + */ + def defaultSparqlOptions(inputOptions: RdfStreamOptions): SparqlResultsOptions = + JellySparqlOptions.BIG.clone().setRdfVersion( + if inputOptions.getRdfStar then RdfVersion.RDF_VERSION_1_2 else RdfVersion.RDF_VERSION_1_1, + ) + + /** Converts a Jelly-RDF stream to a Jelly-SPARQL result set, one solution per statement. + * + * Whether the result set has the ?g variable depends on the physical type of the input: TRIPLES + * streams give 3 variables, QUADS and GRAPHS streams give 4. Namespace declarations are dropped. + * + * @param outputOptions + * makes the stream options of the output from the options of the input + * @param valuesPerFrame + * maximum number of values (cells) in one output frame + * @param delimited + * whether the output should be delimited + */ + def rdfToSparql( + inputStream: InputStream, + outputStream: OutputStream, + outputOptions: RdfStreamOptions => SparqlResultsOptions, + valuesPerFrame: Int, + delimited: Boolean, + ): Unit = + SparqlJellyUtil.translateErrors { + val (inputOptions, vars, bindings) = readStatements(inputStream) + val context = RIOT.getContext.copy() + .set(JellySparqlLanguage.SYMBOL_STREAM_OPTIONS, outputOptions(inputOptions)) + .set(JellySparqlLanguage.SYMBOL_MAX_VALUES_PER_FRAME, valuesPerFrame) + .set(JellySparqlLanguage.SYMBOL_DELIMITED_OUTPUT, delimited) + RowSetWriterRegistry + .getFactory(JellySparqlLanguage.JELLY_SPARQL) + .create(JellySparqlLanguage.JELLY_SPARQL) + .write(outputStream, RowSetStream.create(vars.asJava, bindings.asJava), context) + outputStream.flush() + } + + /** Reads the Jelly-RDF stream frame by frame, turning every statement into a solution. + * + * @return + * the options of the input, the variables of the result set, and a lazy iterator of its + * solutions + */ + private def readStatements( + inputStream: InputStream, + ): (RdfStreamOptions, Seq[Var], Iterator[Binding]) = + val frames = JellyUtil.iterateRdfStream(inputStream).buffered + val inputOptions = frames.headOption.flatMap(_.getRows.asScala.headOption) match + case Some(row) if row.hasOptions => row.getOptions + case Some(_) => throw CriticalException("First input row is not an options row") + case None => throw CriticalException("Empty input stream") + val quads = inputOptions.getPhysicalType match + case PhysicalStreamType.TRIPLES => false + case PhysicalStreamType.QUADS | PhysicalStreamType.GRAPHS => true + case t => throw CriticalException(s"Unsupported physical stream type: $t") + val vars = if quads then StatementBinding.quadVars else StatementBinding.tripleVars + + // The decoder pushes the statements of one frame here, and the iterator takes them out + var buffer = ArrayBuffer[Binding]() + val handler = new AnyStatementHandler[Node] { + override def handleTriple(s: Node, p: Node, o: Node): Unit = + buffer += TripleBinding(s, p, o) + + override def handleQuad(s: Node, p: Node, o: Node, g: Node): Unit = + if !quads then + throw CriticalException( + "Found a quad in a stream that started as a TRIPLES stream. " + + "The result set only has the ?s ?p ?o variables.", + ) + buffer += (if Quad.isDefaultGraph(g) then TripleBinding(s, p, o) + else QuadBinding(s, p, o, g)) + } + val decoder = JenaConverterFactory.getInstance().anyStatementDecoder( + handler, + JellyOptions.DEFAULT_SUPPORTED_OPTIONS, + ) + val bindings = frames.flatMap { (frame: RdfStreamFrame) => + // A new buffer for every frame, so that the iterator can take the old one as it is + buffer = ArrayBuffer[Binding]() + frame.getRows.forEach(decoder.ingestRow) + buffer + } + (inputOptions, vars, bindings) + + /** Converts a Jelly-SPARQL result set to a Jelly-RDF stream, one statement per solution. + * + * @param outputOptions + * stream options of the output. The physical type may be unspecified. + * @param rowsPerFrame + * target number of rows per output frame + * @param delimited + * whether the output should be delimited + */ + def sparqlToRdf( + inputStream: InputStream, + outputStream: OutputStream, + outputOptions: RdfStreamOptions, + rowsPerFrame: Int, + delimited: Boolean, + ): Unit = + SparqlJellyUtil.translateErrors { + convertSparqlToRdf(inputStream, outputStream, outputOptions, rowsPerFrame, delimited) + outputStream.flush() + } + + private def convertSparqlToRdf( + inputStream: InputStream, + outputStream: OutputStream, + outputOptions: RdfStreamOptions, + rowsPerFrame: Int, + delimited: Boolean, + ): 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 + throw CriticalException( + s"The result set must have 3 or 4 variables, but it has ${vars.size}: " + + vars.mkString(", "), + ) + 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)) + def required(i: Int): Node = + val node = get(i) + if node == null then + throw CriticalException( + s"Solution $index 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 + checkStatement( + triple.toString, + StatementUtils.isGeneralized(triple), + StatementUtils.hasTripleTerms(triple), + index, + jellyOpt, + ) + writer.triple(triple) + else + val graph = (if quads then Option(get(3)) else None).getOrElse(Quad.defaultGraphIRI) + val quad = Quad.create(graph, triple) + checkStatement( + quad.toString, + StatementUtils.isGeneralized(quad), + StatementUtils.hasTripleTerms(quad), + index, + 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 = + val jellyOpt = options.clone() + jellyOpt.getPhysicalType match + case PhysicalStreamType.UNSPECIFIED => + jellyOpt.setPhysicalType( + if quads then PhysicalStreamType.QUADS else PhysicalStreamType.TRIPLES, + ) + case PhysicalStreamType.TRIPLES if quads => + throw InvalidArgument( + "--opt.physical-type", + "TRIPLES", + Some("The result set has 4 variables, so the output must be QUADS or GRAPHS"), + ) + case _ => jellyOpt + + private def checkStatement( + statement: String, + generalized: Boolean, + tripleTerms: Boolean, + index: Long, + opt: RdfStreamOptions, + ): Unit = + if !opt.getGeneralizedStatements && generalized then + throw CriticalException( + s"Solution $index 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. " + + "Use --opt.triple-terms=true to allow triple terms.", + ) diff --git a/src/main/scala/eu/neverblink/jelly/cli/util/jena/StatementBinding.scala b/src/main/scala/eu/neverblink/jelly/cli/util/jena/StatementBinding.scala new file mode 100644 index 0000000..2f64b99 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/util/jena/StatementBinding.scala @@ -0,0 +1,128 @@ +package eu.neverblink.jelly.cli.util.jena + +import org.apache.jena.graph.Node +import org.apache.jena.sparql.core.Var +import org.apache.jena.sparql.engine.binding.{Binding, BindingBase} +import org.apache.jena.sparql.util.FmtUtils + +import java.util.function.BiConsumer + +/** Optimized Binding implementations for converting between Jelly-RDF and Jelly-SPARQL. */ +object StatementBinding: + val subject: Var = Var.alloc("s") + val predicate: Var = Var.alloc("p") + val `object`: Var = Var.alloc("o") + val graph: Var = Var.alloc("g") + + val tripleVars: Seq[Var] = Seq(subject, predicate, `object`) + val quadVars: Seq[Var] = Seq(subject, predicate, `object`, graph) + + // Shared by all bindings, so that a binding only holds its terms + private[jena] val tripleVarList: java.util.List[Var] = + java.util.List.of(subject, predicate, `object`) + private[jena] val quadVarList: java.util.List[Var] = + java.util.List.of(subject, predicate, `object`, graph) + private[jena] val tripleVarSet: java.util.Set[Var] = java.util.Set.copyOf(tripleVarList) + private[jena] val quadVarSet: java.util.Set[Var] = java.util.Set.copyOf(quadVarList) + + /** Finds the variable a binding may hold, first by identity, which is the usual case, and then by + * name. Returns null if the variable is not one of ?s ?p ?o ?g. + */ + private[jena] def resolve(v: Var): Var = + if (v eq subject) || (v eq predicate) || (v eq `object`) || (v eq graph) then v + else + v.getVarName match + case "s" => subject + case "p" => predicate + case "o" => `object` + case "g" => graph + case _ => null + + /** Formats a binding the way Jena does: ( ?s = <...> ) ( ?p = <...> ) ... */ + private[jena] def format(b: Binding): String = + val sb = StringBuilder() + b.forEach { (v, n) => + if sb.nonEmpty then sb.append(' ') + sb.append("( ").append(v).append(" = ").append(FmtUtils.stringForNode(n)).append(" )") + } + sb.toString + +/** Binding of ?s ?p ?o. Also used for quads in the default graph, where ?g is unbound. */ +final class TripleBinding(s: Node, p: Node, o: Node) extends Binding: + import StatementBinding.* + + override def vars(): java.util.Iterator[Var] = tripleVarList.iterator() + + override def varsMentioned(): java.util.Set[Var] = tripleVarSet + + override def forEach(action: BiConsumer[Var, Node]): Unit = + action.accept(subject, s) + action.accept(predicate, p) + action.accept(`object`, o) + + override def contains(v: Var): Boolean = get(v) != null + + override def get(v: Var): Node = + val resolved = resolve(v) + if resolved eq subject then s + else if resolved eq predicate then p + else if resolved eq `object` then o + else null + + override def size(): Int = 3 + + override def isEmpty: Boolean = false + + override def detach(): Binding = this + + override def hashCode(): Int = BindingBase.hashCode(this) + + // Same as Jena's: the same variables, bound to the same terms + override def equals(other: Any): Boolean = other match + case b: Binding => + b.size() == 3 && s == b.get(subject) && p == b.get(predicate) && o == b.get(`object`) + case _ => false + + override def toString: String = format(this) + +/** Binding of ?s ?p ?o ?g, for quads in a named graph. All terms must be non-null. */ +final class QuadBinding(s: Node, p: Node, o: Node, g: Node) extends Binding: + import StatementBinding.* + + override def vars(): java.util.Iterator[Var] = quadVarList.iterator() + + override def varsMentioned(): java.util.Set[Var] = quadVarSet + + override def forEach(action: BiConsumer[Var, Node]): Unit = + action.accept(subject, s) + action.accept(predicate, p) + action.accept(`object`, o) + action.accept(graph, g) + + override def contains(v: Var): Boolean = get(v) != null + + override def get(v: Var): Node = + val resolved = resolve(v) + if resolved eq subject then s + else if resolved eq predicate then p + else if resolved eq `object` then o + else if resolved eq graph then g + else null + + override def size(): Int = 4 + + override def isEmpty: Boolean = false + + // There is no parent to detach from + override def detach(): Binding = this + + override def hashCode(): Int = BindingBase.hashCode(this) + + // Same as Jena's: the same variables, bound to the same terms + override def equals(other: Any): Boolean = other match + case b: Binding => + b.size() == 4 && s == b.get(subject) && p == b.get(predicate) && + o == b.get(`object`) && g == b.get(graph) + case _ => false + + override def toString: String = format(this) diff --git a/src/main/scala/eu/neverblink/jelly/cli/util/jena/riot/JellyWriterUtil.scala b/src/main/scala/eu/neverblink/jelly/cli/util/jena/riot/JellyWriterUtil.scala new file mode 100644 index 0000000..f025fb2 --- /dev/null +++ b/src/main/scala/eu/neverblink/jelly/cli/util/jena/riot/JellyWriterUtil.scala @@ -0,0 +1,70 @@ +package eu.neverblink.jelly.cli.util.jena.riot + +import eu.neverblink.jelly.convert.jena.JenaConverterFactory +import eu.neverblink.jelly.convert.jena.riot.{JellyFormatVariant, JellyLanguage, JellyStreamWriter} +import eu.neverblink.jelly.core.proto.v1.{LogicalStreamType, PhysicalStreamType, RdfStreamOptions} +import org.apache.jena.riot.RIOT +import org.apache.jena.riot.system.{StreamRDF, StreamRDFWriter} + +import java.io.OutputStream + +object JellyWriterUtil: + + /** Creates a StreamRDF that writes Jelly-RDF with the given options. + * + * If the physical type is unspecified, the writer picks TRIPLES or QUADS based on the first + * statement it gets. + * + * @param jellyOpt + * stream options of the output + * @param frameSize + * target number of rows per frame + * @param enableNamespaceDeclarations + * whether to write namespace declarations + * @param delimited + * whether the output should be delimited + * @param out + * where to write the output + */ + def createWriter( + jellyOpt: RdfStreamOptions, + frameSize: Int, + enableNamespaceDeclarations: Boolean, + delimited: Boolean, + out: OutputStream, + ): StreamRDF = + jellyOpt.getPhysicalType match + case PhysicalStreamType.GRAPHS => + JellyStreamWriterGraphs( + JellyFormatVariant + .builder() + .options( + jellyOpt.clone.setLogicalType( + if jellyOpt.getLogicalType == LogicalStreamType.UNSPECIFIED then + LogicalStreamType.FLAT_QUADS + else jellyOpt.getLogicalType, + ), + ) + .frameSize(frameSize) + .enableNamespaceDeclarations(enableNamespaceDeclarations) + .isDelimited(delimited) + .build(), + out = out, + ) + case PhysicalStreamType.UNSPECIFIED => + val writerContext = RIOT.getContext.copy() + .set(JellyLanguage.SYMBOL_STREAM_OPTIONS, jellyOpt) + .set(JellyLanguage.SYMBOL_FRAME_SIZE, frameSize) + .set(JellyLanguage.SYMBOL_ENABLE_NAMESPACE_DECLARATIONS, enableNamespaceDeclarations) + .set(JellyLanguage.SYMBOL_DELIMITED_OUTPUT, delimited) + StreamRDFWriter.getWriterStream(out, JellyLanguage.JELLY, writerContext) + case _ => + // TRIPLES or QUADS: if the physical type is specified, we can just construct the writer + val variant = JellyFormatVariant + .builder() + .options(jellyOpt) + .frameSize(frameSize) + .enableNamespaceDeclarations(enableNamespaceDeclarations) + .isDelimited(delimited) + .build() + JellyStreamWriter.create(JenaConverterFactory.getInstance(), variant, out) 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 new file mode 100644 index 0000000..d401be5 --- /dev/null +++ b/src/test/scala/eu/neverblink/jelly/cli/command/helpers/RdfSparqlTestData.scala @@ -0,0 +1,95 @@ +package eu.neverblink.jelly.cli.command.helpers + +import eu.neverblink.jelly.cli.command.rdf.util.JellyUtil +import eu.neverblink.jelly.cli.command.sparql.SparqlSerDesSpec +import eu.neverblink.jelly.cli.util.jena.riot.JellyWriterUtil +import eu.neverblink.jelly.convert.jena.riot.JellyLanguage +import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage +import eu.neverblink.jelly.core.JellyOptions +import eu.neverblink.jelly.core.proto.v1.{PhysicalStreamType, RdfStreamOptions} +import org.apache.jena.graph.{Node, NodeFactory, Triple} +import org.apache.jena.query.ResultSetFactory +import org.apache.jena.riot.RDFParser +import org.apache.jena.sparql.core.{DatasetGraphFactory, Quad} + +import java.io.{ByteArrayInputStream, ByteArrayOutputStream} +import scala.jdk.CollectionConverters.* + +/** Test data and helpers for converting between Jelly-RDF and Jelly-SPARQL. */ +object RdfSparqlTestData: + + def iri(s: String): Node = NodeFactory.createURI("http://example.org/" + s) + def lit(s: String): Node = NodeFactory.createLiteralString(s) + + val triples: Seq[Triple] = Seq( + Triple.create(iri("a"), iri("p"), iri("b")), + Triple.create(iri("a"), iri("q"), lit("hello")), + Triple.create(iri("b"), iri("p"), NodeFactory.createLiteralLang("cześć", "pl")), + ) + + val quads: Seq[Quad] = Seq( + Quad.create(iri("g1"), triples(0)), + Quad.create(iri("g2"), triples(1)), + Quad.create(Quad.defaultGraphIRI, triples(2)), + ) + + /** The triples as quads in the default graph. */ + val triplesAsQuads: Set[Quad] = triples.map(Quad.create(Quad.defaultGraphIRI, _)).toSet + + /** Writes the statements as Jelly-RDF with the given physical type. */ + def writeRdf( + physicalType: PhysicalStreamType, + triples: Seq[Triple] = Nil, + quads: Seq[Quad] = Nil, + rdfStar: Boolean = true, + rowsPerFrame: Int = 256, + ): Array[Byte] = + val out = ByteArrayOutputStream() + val writer = JellyWriterUtil.createWriter( + JellyOptions.SMALL_ALL_FEATURES.clone().setPhysicalType(physicalType).setRdfStar(rdfStar), + rowsPerFrame, + false, + true, + out, + ) + writer.start() + triples.foreach(writer.triple) + quads.foreach(writer.quad) + writer.finish() + out.toByteArray + + /** Reads Jelly-RDF into a set of quads. Triples end up in the default graph. */ + def readRdf(jelly: Array[Byte]): Set[Quad] = + val dataset = DatasetGraphFactory.create() + RDFParser.source(ByteArrayInputStream(jelly)).lang(JellyLanguage.JELLY).parse(dataset) + dataset.find().asScala.map { q => + if q.isDefaultGraph then Quad.create(Quad.defaultGraphIRI, q.asTriple) else q + }.toSet + + /** The stream options of a Jelly-RDF stream. */ + def rdfStreamOptions(jelly: Array[Byte]): RdfStreamOptions = + JellyUtil.iterateRdfStream(ByteArrayInputStream(jelly)).next().getRows.iterator().next() + .getOptions + + /** Reads Jelly-SPARQL into the variables and rows, with each row as a map of the bound values. */ + def readSparql(jelly: Array[Byte]): (Seq[String], Seq[Map[String, Node]]) = + val rs = ResultSetFactory.makeRewindable( + SparqlSerDesSpec.parse(jelly, JellySparqlLanguage.JELLY_SPARQL), + ) + val vars = rs.getResultVars.asScala.toSeq + val rows = Iterator.continually(rs).takeWhile(_.hasNext).map(_.nextBinding()).map { b => + b.vars().asScala.map(v => v.getVarName -> b.get(v)).toMap + }.toSeq + (vars, rows) + + /** 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] = + 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) + + 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/rdf/RdfFromJellySpec.scala b/src/test/scala/eu/neverblink/jelly/cli/command/rdf/RdfFromJellySpec.scala index c5c7b99..954bcb7 100644 --- a/src/test/scala/eu/neverblink/jelly/cli/command/rdf/RdfFromJellySpec.scala +++ b/src/test/scala/eu/neverblink/jelly/cli/command/rdf/RdfFromJellySpec.scala @@ -473,3 +473,38 @@ class RdfFromJellySpec extends AnyWordSpec with Matchers with TestFixtureHelper: } } } + + "rdf from-jelly command with the jelly-sparql format" should { + import eu.neverblink.jelly.cli.command.helpers.RdfSparqlTestData.* + import eu.neverblink.jelly.cli.command.sparql.SparqlSerDesSpec.readFrames + import eu.neverblink.jelly.core.proto.v1.RdfVersion + import eu.neverblink.jelly.core.sparql.JellySparqlOptions + + "write a result set of ?s ?p ?o ?g" in { + RdfFromJelly.setStdIn(ByteArrayInputStream(writeRdf(PhysicalStreamType.QUADS, quads = quads))) + RdfFromJelly.runTestCommand(List("rdf", "from-jelly", "--out-format", "jelly-sparql")) + val (vars, rows) = readSparql(RdfFromJelly.getOutBytes) + vars shouldBe Seq("s", "p", "o", "g") + rows.size shouldBe quads.size + } + + "infer the format from the .jellys extension" in { + withEmptyJellyFile { j => + val target = j + "s" + try + RdfFromJelly.setStdIn(ByteArrayInputStream(writeRdf(PhysicalStreamType.TRIPLES, triples))) + RdfFromJelly.runTestCommand(List("rdf", "from-jelly", "--to", target)) + readSparql(Files.readAllBytes(Paths.get(target)))._1 shouldBe Seq("s", "p", "o") + finally Files.deleteIfExists(Paths.get(target)) + } + } + + "use the default options, with the RDF version inferred from the input" in { + val rdf = writeRdf(PhysicalStreamType.TRIPLES, triples, rdfStar = false) + RdfFromJelly.setStdIn(ByteArrayInputStream(rdf)) + RdfFromJelly.runTestCommand(List("rdf", "from-jelly", "--out-format", "jelly-sparql")) + val options = readFrames(RdfFromJelly.getOutBytes).head.getOptions + options.getMaxNameTableSize shouldBe JellySparqlOptions.BIG.getMaxNameTableSize + options.getRdfVersion shouldBe RdfVersion.RDF_VERSION_1_1 + } + } diff --git a/src/test/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJellySpec.scala b/src/test/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJellySpec.scala index ed7bfb7..c1c116a 100644 --- a/src/test/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJellySpec.scala +++ b/src/test/scala/eu/neverblink/jelly/cli/command/rdf/RdfToJellySpec.scala @@ -970,3 +970,68 @@ class RdfToJellySpec extends AnyWordSpec with TestFixtureHelper with Matchers: } } } + + "rdf to-jelly command with the jelly-sparql format" should { + import eu.neverblink.jelly.cli.command.helpers.RdfSparqlTestData.* + import java.io.ByteArrayOutputStream + import java.nio.file.Files + + /** Jelly-SPARQL with the statements, made without going through any command. */ + def sparqlOf(rdf: Array[Byte]): Array[Byte] = + val out = ByteArrayOutputStream() + eu.neverblink.jelly.cli.util.jena.RdfSparqlConverter.rdfToSparql( + ByteArrayInputStream(rdf), + out, + eu.neverblink.jelly.cli.util.jena.RdfSparqlConverter.defaultSparqlOptions, + 4096, + true, + ) + out.toByteArray + + "read a result set of ?s ?p ?o ?g" in { + RdfToJelly.setStdIn( + ByteArrayInputStream(sparqlOf(writeRdf(PhysicalStreamType.QUADS, quads = quads))), + ) + RdfToJelly.runTestCommand( + List("rdf", "to-jelly", "--in-format", "jelly-sparql", "--opt.stream-name=z"), + ) + val jelly = RdfToJelly.getOutBytes + rdfStreamOptions(jelly).getPhysicalType shouldBe PhysicalStreamType.QUADS + rdfStreamOptions(jelly).getStreamName shouldBe "z" + readRdf(jelly) shouldBe quads.toSet + } + + "infer the format from the .jellys extension" in { + val file = Files.createTempFile("jelly-cli-rdf", ".jellys") + try + Files.write(file, sparqlOf(writeRdf(PhysicalStreamType.TRIPLES, triples))) + RdfToJelly.runTestCommand(List("rdf", "to-jelly", file.toString)) + readRdf(RdfToJelly.getOutBytes) shouldBe triplesAsQuads + finally Files.deleteIfExists(file) + } + + "write a GRAPHS stream with --opt.physical-type=GRAPHS" in { + RdfToJelly.setStdIn( + ByteArrayInputStream(sparqlOf(writeRdf(PhysicalStreamType.QUADS, quads = quads))), + ) + RdfToJelly.runTestCommand( + List("rdf", "to-jelly", "--in-format", "jelly-sparql", "--opt.physical-type=GRAPHS"), + ) + rdfStreamOptions(RdfToJelly.getOutBytes).getPhysicalType shouldBe PhysicalStreamType.GRAPHS + } + + "report a result set that cannot be written as RDF" in { + RdfToJelly.setStdIn( + ByteArrayInputStream( + writeSparql( + Seq("s", "p", "o"), + Seq(Map("s" -> litJson("lit"), "p" -> iriJson("p"), "o" -> litJson("v"))), + ), + ), + ) + val e = intercept[ExitException] { + RdfToJelly.runTestCommand(List("rdf", "to-jelly", "--in-format", "jelly-sparql")) + } + e.getCause.getMessage should include("--opt.generalized-statements=true") + } + } 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 new file mode 100644 index 0000000..e33a1a6 --- /dev/null +++ b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlInspectSpec.scala @@ -0,0 +1,147 @@ +package eu.neverblink.jelly.cli.command.sparql + +import com.google.protobuf.ByteString +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.sparql.JellySparqlOptions +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec +import org.yaml.snakeyaml.Yaml + +import java.io.ByteArrayInputStream +import java.nio.charset.StandardCharsets.UTF_8 +import java.util +import scala.jdk.CollectionConverters.* + +class SparqlInspectSpec extends AnyWordSpec with TestFixtureHelper with Matchers: + import SparqlSerDesSpec.* + + // Not used by these tests – the SPARQL fixtures are written out by hand. + protected val testCardinality: Int = 0 + + SparqlInspect.testMode(true) + + private type YamlMap = util.Map[String, Any] + + private def inspect(jelly: Array[Byte], args: String*): YamlMap = + SparqlInspect.setStdIn(ByteArrayInputStream(jelly)) + val (out, _) = SparqlInspect.runTestCommand(List("sparql", "inspect") ++ args) + Yaml().load(out).asInstanceOf[YamlMap] + + private def perFrame(parsed: YamlMap): Seq[YamlMap] = + parsed.get("frames").asInstanceOf[util.List[YamlMap]].asScala.toSeq + + private def aggregate(parsed: YamlMap): YamlMap = + parsed.get("frames").asInstanceOf[YamlMap] + + "sparql inspect command" should { + "print the stream options, the variables, and aggregate statistics" in { + val parsed = inspect(writeJelly(selectJson)) + val options = parsed.get("stream_options").asInstanceOf[YamlMap] + options.get("stream_name") shouldBe "" + options.get("max_name_table_size") shouldBe JellySparqlOptions.BIG.getMaxNameTableSize + options.get("version") shouldBe 1 + options.get("rdf_version") shouldBe "RDF_VERSION_UNSPECIFIED (0)" + parsed.get("variables").asInstanceOf[util.List[String]].asScala shouldBe + Seq("s", "label", "num", "bn") + parsed.containsKey("ask_result") shouldBe false + val frames = aggregate(parsed) + frames.get("frame_count") shouldBe 1 + frames.get("row_count") shouldBe 3 + frames.get("option_count") shouldBe 1 + frames.get("variable_count") shouldBe 4 + frames.get("iri_value_count") shouldBe 3 + frames.get("bnode_value_count") shouldBe 1 + frames.get("trailer_count") shouldBe 1 + frames.get("ask_result_count") shouldBe 0 + frames.containsKey("trailer_error") shouldBe false + } + + "print the variables as a plain YAML list" in { + SparqlInspect.setStdIn(ByteArrayInputStream(writeJelly(selectJson))) + val (out, _) = SparqlInspect.runTestCommand(List("sparql", "inspect")) + val nl = System.lineSeparator() + out should include( + Seq("variables: ", " - \"s\"", " - \"label\"", " - \"num\"", " - \"bn\"").mkString(nl), + ) + } + + "print statistics per frame" in { + val parsed = inspect(toSmallFrames(selectJson, 4), "--per-frame") + val frames = perFrame(parsed) + frames.size should be > 1 + frames.map(_.get("frame_index")) shouldBe frames.indices + frames.map(_.get("row_count").asInstanceOf[Int]).sum shouldBe 3 + frames.head.get("option_count") shouldBe 1 + frames.tail.map(_.get("option_count")).toSet shouldBe Set(0) + frames.last.get("trailer_count") shouldBe 1 + } + + "aggregate statistics over several frames" in { + val jelly = toSmallFrames(selectJson, 4) + val frames = aggregate(inspect(jelly)) + frames.get("frame_count") shouldBe readFrames(jelly).size + frames.get("row_count") shouldBe 3 + } + + "report sizes instead of counts with --size" in { + val jelly = writeJelly(selectJson) + val frames = aggregate(inspect(jelly, "--size")) + frames.get("frame_size") shouldBe readFrames(jelly).head.getSerializedSize + frames.get("row_count") shouldBe 3 + frames.get("iri_value_size").asInstanceOf[Int] should be > 0 + frames.containsKey("iri_value_count") shouldBe false + } + + "print the result of an ASK query" in { + for value <- Seq(true, false) do + val parsed = inspect(writeJelly(askJson(value))) + parsed.get("ask_result") shouldBe value + parsed.containsKey("variables") shouldBe false + aggregate(parsed).get("ask_result_count") shouldBe 1 + } + + "print frame metadata in --per-frame" in { + val frame = readFrames(writeJelly(selectJson)).head.clone() + frame.addMetadata( + SparqlResultsFrame.MetadataEntry.newInstance() + .setKey("key").setValue(ByteString.copyFromUtf8("abc")), + ) + frame.resetCachedSize() + val frames = perFrame(inspect(writeFrames(Seq(frame)), "--per-frame")) + frames.head.get("metadata").asInstanceOf[YamlMap].get("key") shouldBe "616263" + } + + "print the error reported by the trailer" in { + val frame = readFrames(writeJelly(selectJson)).head.clone() + .setTrailer(SparqlResultsTrailer.newInstance().setError("query timeout")) + frame.resetCachedSize() + val jelly = writeFrames(Seq(frame)) + aggregate(inspect(jelly)).get("trailer_error") shouldBe "query timeout" + perFrame(inspect(jelly, "--per-frame")).head.get("trailer_error") shouldBe "query timeout" + } + + "print an empty list of variables for a zero-variable result set" in { + val json = """{ "head": { "vars": [] }, "results": { "bindings": [ {}, {} ] } }""" + val parsed = inspect(writeJelly(json)) + parsed.get("variables").asInstanceOf[util.List[String]].asScala shouldBe empty + aggregate(parsed).get("row_count") shouldBe 2 + } + + "complain about empty input" in { + val e = intercept[ExitException](inspect(Array())) + e.getCause.getMessage should include("Empty input stream") + } + + "complain about malformed input" in { + val e = intercept[ExitException](inspect("definitely not Jelly".getBytes(UTF_8))) + e.getCause shouldBe a[InvalidJellyFile] + } + + "complain about a first frame without stream options" in { + val frames = readFrames(toSmallFrames(selectJson, 4)) + val e = intercept[ExitException](inspect(writeFrames(frames.tail))) + e.getCause.getMessage should include("does not contain stream options") + } + } 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 cd0954b..d9189cb 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 @@ -2,11 +2,18 @@ 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 +import eu.neverblink.jelly.cli.command.sparql.util.{SparqlFormat, SparqlJellyUtil} import eu.neverblink.jelly.convert.jena.sparql.JellySparqlLanguage +import eu.neverblink.jelly.core.proto.v1.RdfVersion +import eu.neverblink.jelly.core.proto.v1.sparql.SparqlResultsFrame +import eu.neverblink.jelly.core.sparql.{JellySparqlConstants, JellySparqlOptions} import org.apache.jena.query.{ResultSet, ResultSetFactory} import org.apache.jena.riot.{Lang, RIOT, ResultSetMgr} -import org.apache.jena.riot.resultset.{ResultSetLang, ResultSetWriterRegistry} +import org.apache.jena.riot.resultset.{ + ResultSetLang, + ResultSetReaderRegistry, + ResultSetWriterRegistry, +} import org.apache.jena.sparql.resultset.ResultsCompare import org.scalatest.matchers.should.Matchers import org.scalatest.wordspec.AnyWordSpec @@ -44,13 +51,37 @@ object SparqlSerDesSpec: /** Writes the SELECT results to Jelly-SPARQL, with at most `valuesPerFrame` cells per frame. */ def toSmallFrames(json: String, valuesPerFrame: Int): Array[Byte] = + writeJelly(json, valuesPerFrame = valuesPerFrame) + + /** Writes SPARQL JSON results (SELECT or ASK) to Jelly-SPARQL. */ + def writeJelly( + json: String, + valuesPerFrame: Int = JellySparqlConstants.DEFAULT_MAX_VALUES_PER_FRAME, + delimited: Boolean = true, + ): Array[Byte] = val out = ByteArrayOutputStream() val context = RIOT.getContext.copy() .set(JellySparqlLanguage.SYMBOL_MAX_VALUES_PER_FRAME, valuesPerFrame) - ResultSetWriterRegistry + .set(JellySparqlLanguage.SYMBOL_DELIMITED_OUTPUT, delimited) + val writer = ResultSetWriterRegistry .getFactory(JellySparqlLanguage.JELLY_SPARQL) .create(JellySparqlLanguage.JELLY_SPARQL) - .write(out, parse(json, ResultSetLang.RS_JSON), context) + val result = ResultSetReaderRegistry + .getFactory(ResultSetLang.RS_JSON) + .create(ResultSetLang.RS_JSON) + .readAny(ByteArrayInputStream(json.getBytes(UTF_8)), context) + if result.isBoolean then writer.write(out, result.getBooleanResult.booleanValue, context) + else writer.write(out, result.getResultSet, context) + out.toByteArray + + /** Reads the frames of a delimited Jelly-SPARQL stream. */ + def readFrames(bytes: Array[Byte]): Seq[SparqlResultsFrame] = + SparqlJellyUtil.iterateSparqlStream(ByteArrayInputStream(bytes)).toSeq + + /** Writes the frames as a delimited Jelly-SPARQL stream. */ + def writeFrames(frames: Seq[SparqlResultsFrame]): Array[Byte] = + val out = ByteArrayOutputStream() + frames.foreach(_.writeDelimitedTo(out)) out.toByteArray class SparqlSerDesSpec extends AnyWordSpec with TestFixtureHelper with Matchers: @@ -145,6 +176,126 @@ class SparqlSerDesSpec extends AnyWordSpec with TestFixtureHelper with Matchers: } e.getCause shouldBe a[InvalidFormatSpecified] } + + "use the BIG preset options by default" in { + val options = readFrames(toJelly(selectJson, ".srj")).head.getOptions + options.getMaxNameTableSize shouldBe JellySparqlOptions.BIG.getMaxNameTableSize + options.getMaxPrefixTableSize shouldBe JellySparqlOptions.BIG.getMaxPrefixTableSize + options.getMaxDatatypeTableSize shouldBe JellySparqlOptions.BIG.getMaxDatatypeTableSize + options.getRdfVersion shouldBe RdfVersion.RDF_VERSION_UNSPECIFIED + options.getStreamName shouldBe "" + } + + "set the stream options from --opt.* options" in { + val jelly = toJelly( + selectJson, + ".srj", + List( + "--opt.stream-name=my-stream", + "--opt.rdf-version=1.2-basic", + "--opt.max-name-table-size=200", + "--opt.max-prefix-table-size=0", + "--opt.max-datatype-table-size=10", + ), + ) + val options = readFrames(jelly).head.getOptions + options.getStreamName shouldBe "my-stream" + options.getRdfVersion shouldBe RdfVersion.RDF_VERSION_1_2_BASIC + options.getMaxNameTableSize shouldBe 200 + options.getMaxPrefixTableSize shouldBe 0 + options.getMaxDatatypeTableSize shouldBe 10 + options.getVersion shouldBe JellySparqlConstants.PROTO_VERSION + ResultsCompare.equalsByTermAndOrder( + parse(jelly, JellySparqlLanguage.JELLY_SPARQL), + parse(selectJson, ResultSetLang.RS_JSON), + ) shouldBe true + } + + "accept every RDF version label" in { + for (label, version) <- Seq( + "1.1" -> RdfVersion.RDF_VERSION_1_1, + "1.2-basic" -> RdfVersion.RDF_VERSION_1_2_BASIC, + "1.2" -> RdfVersion.RDF_VERSION_1_2, + "unspecified" -> RdfVersion.RDF_VERSION_UNSPECIFIED, + ) + do + val jelly = toJelly(askJson(true), ".srj", List("--opt.rdf-version", label)) + readFrames(jelly).head.getOptions.getRdfVersion shouldBe version + } + + "copy the stream options from another file with --options-from, with overrides" in { + val source = toJelly( + selectJson, + ".srj", + List("--opt.stream-name=source", "--opt.max-name-table-size=300"), + ) + withEmptyFile(".jellys") { sourceFile => + Files.write(Path.of(sourceFile), source) + val jelly = toJelly( + selectJson, + ".srj", + List("--options-from", sourceFile, "--opt.stream-name=override"), + ) + val options = readFrames(jelly).head.getOptions + options.getStreamName shouldBe "override" + options.getMaxNameTableSize shouldBe 300 + } + } + + "reject an unknown RDF version" in { + val e = intercept[ExitException] { + toJelly(selectJson, ".srj", List("--opt.rdf-version=1.3")) + } + e.getCause shouldBe a[InvalidArgument] + e.getCause.getMessage should include("--opt.rdf-version") + } + + "reject a name table smaller than the minimum" in { + val e = intercept[ExitException] { + toJelly(selectJson, ".srj", List("--opt.max-name-table-size=64")) + } + e.getCause shouldBe a[InvalidArgument] + e.getCause.getMessage should include("--opt.max-name-table-size") + } + + "warn about lookup tables that default readers do not accept, unless --quiet is set" in { + val tooBig = (JellySparqlOptions.MAX_NAME_TABLE_SIZE + 1).toString + withFile(selectJson, ".srj") { f => + val (_, err) = SparqlToJelly.runTestCommand( + List("sparql", "to-jelly", f, "--opt.max-name-table-size", tooBig), + ) + err should include("WARNING") + err should include("name table") + val (_, quietErr) = SparqlToJelly.runTestCommand( + List("sparql", "to-jelly", f, "--opt.max-name-table-size", tooBig, "--quiet"), + ) + quietErr shouldBe empty + } + } + + "split the output into frames with --values-per-frame" in { + val frames = readFrames(toJelly(selectJson, ".srj", List("--values-per-frame=4"))) + // 4 variables and 4 values per frame – one row per frame, plus the trailer + frames.map(_.getRowCount).sum shouldBe 3 + frames.size should be > 1 + } + + "reject --values-per-frame below 1" in { + val e = intercept[ExitException] { + toJelly(selectJson, ".srj", List("--values-per-frame=0")) + } + e.getCause shouldBe a[InvalidArgument] + } + + "write a single non-delimited frame with --delimited=false" in { + val jelly = toJelly(selectJson, ".srj", List("--delimited=false")) + SparqlJellyUtil.iterateSparqlStreamWithDelimitingInfo(ByteArrayInputStream(jelly))._1 shouldBe + false + ResultsCompare.equalsByTermAndOrder( + parse(jelly, JellySparqlLanguage.JELLY_SPARQL), + parse(selectJson, ResultSetLang.RS_JSON), + ) shouldBe true + } } "sparql from-jelly command" should { @@ -216,6 +367,85 @@ class SparqlSerDesSpec extends AnyWordSpec with TestFixtureHelper with Matchers: } } + 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) + try testCode(file.toString) + finally Files.deleteIfExists(file) + + "Jelly-RDF format" should { + import eu.neverblink.jelly.cli.command.helpers.RdfSparqlTestData.* + import eu.neverblink.jelly.core.proto.v1.PhysicalStreamType + + "be read by to-jelly, with the variables ?s ?p ?o (?g)" in { + SparqlToJelly.setStdIn( + ByteArrayInputStream(writeRdf(PhysicalStreamType.QUADS, quads = quads)), + ) + SparqlToJelly.runTestCommand(List("sparql", "to-jelly", "--in-format", "jelly-rdf")) + val (vars, rows) = readSparql(SparqlToJelly.getOutBytes) + vars shouldBe Seq("s", "p", "o", "g") + rows.size shouldBe quads.size + } + + "be inferred from the .jelly extension in to-jelly" in { + withBytesFile(writeRdf(PhysicalStreamType.TRIPLES, triples), ".jelly") { f => + SparqlToJelly.runTestCommand(List("sparql", "to-jelly", f)) + readSparql(SparqlToJelly.getOutBytes)._1 shouldBe Seq("s", "p", "o") + } + } + + "get the RDF version from the input in to-jelly, unless --opt.rdf-version is set" in { + val rdf = writeRdf(PhysicalStreamType.TRIPLES, triples, rdfStar = false) + SparqlToJelly.setStdIn(ByteArrayInputStream(rdf)) + SparqlToJelly.runTestCommand( + List("sparql", "to-jelly", "--in-format", "jelly-rdf", "--opt.stream-name=x"), + ) + val options = readFrames(SparqlToJelly.getOutBytes).head.getOptions + options.getRdfVersion shouldBe RdfVersion.RDF_VERSION_1_1 + options.getStreamName shouldBe "x" + SparqlToJelly.setStdIn(ByteArrayInputStream(rdf)) + SparqlToJelly.runTestCommand( + List("sparql", "to-jelly", "--in-format", "jelly-rdf", "--opt.rdf-version=1.2"), + ) + readFrames(SparqlToJelly.getOutBytes).head.getOptions.getRdfVersion shouldBe + RdfVersion.RDF_VERSION_1_2 + } + + "be written by from-jelly" in { + val sparql = toJellyFromRdf(writeRdf(PhysicalStreamType.QUADS, quads = quads)) + SparqlFromJelly.setStdIn(ByteArrayInputStream(sparql)) + SparqlFromJelly.runTestCommand( + List("sparql", "from-jelly", "--out-format", "jelly-rdf"), + ) + val jelly = SparqlFromJelly.getOutBytes + rdfStreamOptions(jelly).getPhysicalType shouldBe PhysicalStreamType.QUADS + readRdf(jelly) shouldBe quads.toSet + } + + "be inferred from the .jelly extension in from-jelly" in { + val sparql = toJellyFromRdf(writeRdf(PhysicalStreamType.TRIPLES, triples)) + withEmptyFile(".jelly") { target => + SparqlFromJelly.setStdIn(ByteArrayInputStream(sparql)) + SparqlFromJelly.runTestCommand(List("sparql", "from-jelly", "--to", target)) + readRdf(Files.readAllBytes(Path.of(target))) shouldBe triplesAsQuads + } + } + + "report a result set that cannot be written as RDF in from-jelly" in { + SparqlFromJelly.setStdIn(ByteArrayInputStream(writeSparql(Seq("a", "b"), Nil))) + val e = intercept[ExitException] { + SparqlFromJelly.runTestCommand(List("sparql", "from-jelly", "--out-format", "jelly-rdf")) + } + e.getCause.getMessage should include("must have 3 or 4 variables") + } + } + + /** Runs `sparql to-jelly` over Jelly-RDF and returns the Jelly-SPARQL bytes. */ + private def toJellyFromRdf(rdf: Array[Byte]): Array[Byte] = + SparqlToJelly.setStdIn(ByteArrayInputStream(rdf)) + SparqlToJelly.runTestCommand(List("sparql", "to-jelly", "--in-format", "jelly-rdf")) + SparqlToJelly.getOutBytes + /** Runs `sparql from-jelly` to the text format and returns the text. */ private def toText(jelly: Array[Byte]): String = SparqlFromJelly.setStdIn(ByteArrayInputStream(jelly)) @@ -311,6 +541,55 @@ class SparqlSerDesSpec extends AnyWordSpec with TestFixtureHelper with Matchers: quietErr shouldBe empty } + "write non-delimited output from text with --delimited=false" in { + val text = toText(toJelly(selectJson, ".srj")) + SparqlToJelly.setStdIn(ByteArrayInputStream(text.getBytes(UTF_8))) + SparqlToJelly.runTestCommand( + List( + "sparql", + "to-jelly", + "--in-format", + "jelly-sparql-text", + "--quiet", + "--delimited=false", + ), + ) + val jelly = SparqlToJelly.getOutBytes + SparqlJellyUtil.iterateSparqlStreamWithDelimitingInfo(ByteArrayInputStream(jelly))._1 shouldBe + false + ResultsCompare.equalsByTermAndOrder( + parse(jelly, JellySparqlLanguage.JELLY_SPARQL), + parse(selectJson, ResultSetLang.RS_JSON), + ) shouldBe true + } + + "refuse to write several frames from text as non-delimited output" in { + val text = toText(toSmallFrames(selectJson, 4)) + SparqlToJelly.setStdIn(ByteArrayInputStream(text.getBytes(UTF_8))) + val e = intercept[ExitException] { + SparqlToJelly.runTestCommand( + List( + "sparql", + "to-jelly", + "--in-format", + "jelly-sparql-text", + "--quiet", + "--delimited=false", + ), + ) + } + e.getCause.getMessage should include("more than one frame") + } + + "warn that stream options are ignored for text input" in { + val text = toText(toJelly(selectJson, ".srj")) + SparqlToJelly.setStdIn(ByteArrayInputStream(text.getBytes(UTF_8))) + val (_, err) = SparqlToJelly.runTestCommand( + List("sparql", "to-jelly", "--in-format", "jelly-sparql-text", "--opt.stream-name=x"), + ) + err should include("Stream options are ignored") + } + "report malformed text" in { val e = intercept[ExitException] { fromText("this is { definitely not a frame") 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 new file mode 100644 index 0000000..67fc69f --- /dev/null +++ b/src/test/scala/eu/neverblink/jelly/cli/command/sparql/SparqlValidateSpec.scala @@ -0,0 +1,312 @@ +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 org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import java.io.ByteArrayInputStream +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.{Files, Path} +import java.util.UUID.randomUUID + +class SparqlValidateSpec extends AnyWordSpec with TestFixtureHelper with Matchers: + import SparqlSerDesSpec.* + + // Not used by these tests – the SPARQL fixtures are written out by hand. + protected val testCardinality: Int = 0 + + SparqlValidate.testMode(true) + + private val tmpDir: Path = Files.createTempDirectory("jelly-cli-sparql-validate") + + private def withFile[T](content: Array[Byte], extension: String)(testCode: String => T): T = + val file = Files.createTempFile(tmpDir, randomUUID.toString, extension) + Files.write(file, content) + try testCode(file.toString) + finally Files.deleteIfExists(file) + + private def validate(jelly: Array[Byte], args: String*): Unit = + SparqlValidate.setStdIn(ByteArrayInputStream(jelly)) + SparqlValidate.runTestCommand(List("sparql", "validate") ++ args) + + /** Runs the validation, expecting it to fail, and returns the error. */ + private def validateFails(jelly: Array[Byte], args: String*): Throwable = + intercept[ExitException](validate(jelly, args*)).getCause + + /** The same result set as selectJson, with the rows in reverse order and the blank node renamed. + */ + private val selectJsonReordered: String = + """{ "head": { "vars": [ "s", "label", "num", "bn" ] }, + | "results": { "bindings": [ + | { "s": {"type":"uri","value":"http://example.org/c"}, + | "num": {"type":"literal","value":"-1", + | "datatype":"http://www.w3.org/2001/XMLSchema#integer"} }, + | { "s": {"type":"uri","value":"http://example.org/b"}, + | "label": {"type":"literal","value":"cześć","xml:lang":"pl"} }, + | { "s": {"type":"uri","value":"http://example.org/a"}, + | "label": {"type":"literal","value":"hello"}, + | "num": {"type":"literal","value":"42", + | "datatype":"http://www.w3.org/2001/XMLSchema#integer"}, + | "bn": {"type":"bnode","value":"other"} } + | ] } }""".stripMargin + + private def withTrailer(frame: SparqlResultsFrame, error: Option[String]): SparqlResultsFrame = + val f = frame.clone() + .setTrailer(error.map(e => SparqlResultsTrailer.newInstance().setError(e)).orNull) + f.resetCachedSize() + f + + "sparql validate command" should { + "accept a valid SELECT result stream" in { + validate(writeJelly(selectJson)) + validate(toSmallFrames(selectJson, 4)) + } + + "accept a valid ASK result stream" in { + validate(writeJelly(askJson(true))) + validate(writeJelly(askJson(false)), "--require-trailer") + } + + "complain about empty input" in { + validateFails(Array()).getMessage should include("Empty input stream") + } + + "complain about malformed input" in { + validateFails("definitely not Jelly".getBytes(UTF_8)) shouldBe a[InvalidJellyFile] + } + + "check the delimiting" when { + val delimited = writeJelly(selectJson) + val undelimited = writeJelly(selectJson, delimited = false) + + "input is delimited" in { + validate(delimited) + validate(delimited, "--delimited=either") + validate(delimited, "--delimited=true") + validateFails(delimited, "--delimited=false").getMessage should include( + "Expected undelimited input", + ) + } + + "input is undelimited" in { + validate(undelimited) + validate(undelimited, "--delimited=false") + validateFails(undelimited, "--delimited=true").getMessage should include( + "Expected delimited input", + ) + } + + "the argument is invalid" in { + validateFails(delimited, "--delimited=maybe") shouldBe a[InvalidArgument] + } + } + + "check the stream structure" when { + "the first frame has no stream options" in { + val frames = readFrames(toSmallFrames(selectJson, 4)) + val broken = writeFrames(frames.tail) + validateFails(broken).getMessage should include("does not contain stream options") + } + + "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") + } + + "the stream options are repeated and differ" in { + 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", + ) + } + + "the version in the options is 0" in { + val frames = readFrames(writeJelly(selectJson)).map { f => + val c = f.clone().setOptions(f.getOptions.clone().setVersion(0)) + c.resetCachedSize() + c + } + validateFails(writeFrames(frames)).getMessage should include("version") + } + } + + "check the trailer" when { + "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( + "does not end with a trailer", + ) + } + + "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( + "does not end with a trailer", + ) + } + + "the trailer reports an error" in { + val frames = readFrames(writeJelly(selectJson)).map(withTrailer(_, Some("query timeout"))) + validateFails(writeFrames(frames)).getMessage should include("query timeout") + } + } + + "check the stream options against an options file" when { + "the options match" in { + val jelly = writeJelly(selectJson) + withFile(writeJelly(askJson(true)), ".jellys") { optionsFile => + validate(jelly, "--options-file", optionsFile) + } + } + + "the options differ" in { + val optionsFrames = readFrames(writeJelly(askJson(true))).map { f => + val c = f.clone().setOptions(f.getOptions.clone().setMaxNameTableSize(200)) + c.resetCachedSize() + c + } + withFile(writeFrames(optionsFrames), ".jellys") { optionsFile => + val e = validateFails(writeJelly(selectJson), "--options-file", optionsFile) + e.getMessage should include("Stream options do not match") + e.getMessage should include("max_name_table_size: 200") + } + } + } + + "compare the results to a reference file" when { + "the results match (JSON)" in { + withFile(selectJson.getBytes(UTF_8), ".srj") { ref => + validate(writeJelly(selectJson), "--compare-to-file", ref) + validate(toSmallFrames(selectJson, 4), "--compare-to-file", ref) + } + } + + "the results match (Jelly-SPARQL)" in { + withFile(toSmallFrames(selectJson, 4), ".jellys") { ref => + validate(writeJelly(selectJson), "--compare-to-file", ref) + } + } + + "the format is given explicitly" in { + withFile(selectJson.getBytes(UTF_8), ".txt") { ref => + validate(writeJelly(selectJson), "--compare-to-file", ref, "--compare-to-format=json") + } + } + + "the format cannot be inferred" in { + withFile(selectJson.getBytes(UTF_8), ".txt") { ref => + validateFails(writeJelly(selectJson), "--compare-to-file", ref) shouldBe + a[InvalidFormatSpecified] + } + } + + "the rows are in a different order" in { + withFile(selectJsonReordered.getBytes(UTF_8), ".srj") { ref => + val e = validateFails(writeJelly(selectJson), "--compare-to-file", ref) + e.getMessage should include("Solution 0 does not match") + e.getMessage should include("Expected: ?s=") + e.getMessage should include("Actual: ?s=") + } + } + + "the blank nodes are renamed consistently" in { + withFile(selectJson.replace("b0", "renamed").getBytes(UTF_8), ".srj") { ref => + validate(writeJelly(selectJson), "--compare-to-file", ref) + } + } + + "the blank nodes are not renamed consistently" in { + def bnodes(labels: String*) = + val rows = labels.map(l => s"""{ "x": {"type":"bnode","value":"$l"} }""") + s"""{ "head": { "vars": [ "x" ] }, "results": { "bindings": [ ${rows.mkString( + ", ", + )} ] } }""" + // One label on one side, two labels on the other – both ways round + for (expected, actual) <- Seq( + bnodes("a", "a") -> bnodes("b", "c"), + bnodes("a", "b") -> bnodes("c", "c"), + ) + do + withFile(expected.getBytes(UTF_8), ".srj") { ref => + validateFails(writeJelly(actual), "--compare-to-file", ref) + .getMessage should include("Solution 1 does not match") + } + validate( + writeJelly(bnodes("c", "d", "c")), + "--compare-to-file", { + val f = Files.createTempFile(tmpDir, randomUUID.toString, ".srj"); + Files.write(f, bnodes("a", "b", "a").getBytes(UTF_8)); f.toString + }, + ) + } + + "the variables are in a different order" in { + val reordered = selectJson.replace( + """"vars": [ "s", "label", "num", "bn" ]""", + """"vars": [ "label", "s", "num", "bn" ]""", + ) + withFile(reordered.getBytes(UTF_8), ".srj") { ref => + validateFails(writeJelly(selectJson), "--compare-to-file", ref) + .getMessage should include("Expected variables ?label ?s ?num ?bn") + } + } + + "the results differ" in { + val different = selectJson.replace("hello", "goodbye") + withFile(different.getBytes(UTF_8), ".srj") { ref => + validateFails(writeJelly(selectJson), "--compare-to-file", ref) + .getMessage should include("Solution 0 does not match") + } + } + + "the number of solutions differs" in { + // The first solution of selectJson + val fewer = + """{ "head": { "vars": [ "s", "label", "num", "bn" ] }, + | "results": { "bindings": [ + | { "s": {"type":"uri","value":"http://example.org/a"}, + | "label": {"type":"literal","value":"hello"}, + | "num": {"type":"literal","value":"42", + | "datatype":"http://www.w3.org/2001/XMLSchema#integer"}, + | "bn": {"type":"bnode","value":"b0"} } ] } }""".stripMargin + withFile(fewer.getBytes(UTF_8), ".srj") { ref => + validateFails(writeJelly(selectJson), "--compare-to-file", ref) + .getMessage should include("Expected 1 solutions, as in") + } + withFile(selectJson.getBytes(UTF_8), ".srj") { ref => + validateFails(writeJelly(fewer), "--compare-to-file", ref) + .getMessage should include("Expected 3 solutions, as in") + } + } + + "the variables differ" in { + val otherVars = selectJson.replace("\"bn\" ]", "\"other\" ]") + withFile(otherVars.getBytes(UTF_8), ".srj") { ref => + validateFails(writeJelly(selectJson), "--compare-to-file", ref) + .getMessage should include("Expected variables") + } + } + + "the ASK results match or differ" in { + withFile(askJson(true).getBytes(UTF_8), ".srj") { ref => + validate(writeJelly(askJson(true)), "--compare-to-file", ref) + validateFails(writeJelly(askJson(false)), "--compare-to-file", ref) + .getMessage should include("Expected the ASK result to be true") + validateFails(writeJelly(selectJson), "--compare-to-file", ref) + .getMessage should include("Expected an ASK result") + } + } + } + } 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 new file mode 100644 index 0000000..c87042e --- /dev/null +++ b/src/test/scala/eu/neverblink/jelly/cli/util/jena/RdfSparqlConverterSpec.scala @@ -0,0 +1,255 @@ +package eu.neverblink.jelly.cli.util.jena + +import eu.neverblink.jelly.cli.* +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.core.sparql.{JellySparqlConstants, JellySparqlOptions} +import org.apache.jena.graph.{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 + +class RdfSparqlConverterSpec extends AnyWordSpec with TestFixtureHelper with Matchers: + + // Not used by these tests – the fixtures are written out by hand. + protected val testCardinality: Int = 0 + + private def toSparql( + jelly: Array[Byte], + outputOptions: RdfStreamOptions => SparqlResultsOptions = + RdfSparqlConverter.defaultSparqlOptions, + valuesPerFrame: Int = JellySparqlConstants.DEFAULT_MAX_VALUES_PER_FRAME, + ): Array[Byte] = + val out = ByteArrayOutputStream() + RdfSparqlConverter.rdfToSparql( + ByteArrayInputStream(jelly), + out, + outputOptions, + valuesPerFrame, + delimited = true, + ) + out.toByteArray + + /** The options the CLI uses when no --opt.* option is given. */ + private def defaultRdfOptions = RdfJellySerializationOptions().asRdfStreamOptions.clone() + + private def toRdf( + jelly: Array[Byte], + options: RdfStreamOptions = defaultRdfOptions, + ): Array[Byte] = + val out = ByteArrayOutputStream() + RdfSparqlConverter.sparqlToRdf(ByteArrayInputStream(jelly), out, options, 256, true) + out.toByteArray + + "rdfToSparql" should { + "turn a TRIPLES stream into ?s ?p ?o solutions" in { + val (vars, rows) = readSparql(toSparql(writeRdf(PhysicalStreamType.TRIPLES, triples))) + vars shouldBe Seq("s", "p", "o") + rows shouldBe triples.map { t => + Map("s" -> t.getSubject, "p" -> t.getPredicate, "o" -> t.getObject) + } + } + + "turn a QUADS stream into ?s ?p ?o ?g solutions, with ?g unbound for the default graph" in { + val (vars, rows) = readSparql(toSparql(writeRdf(PhysicalStreamType.QUADS, quads = quads))) + vars shouldBe Seq("s", "p", "o", "g") + rows shouldBe quads.map { q => + Map("s" -> q.getSubject, "p" -> q.getPredicate, "o" -> q.getObject) ++ + (if q.isDefaultGraph then Map() else Map("g" -> q.getGraph)) + } + } + + "leave ?g unbound only for the default graph, wherever it is in the stream" in { + // The default graph between named graphs, twice, and a named graph repeated after it + val mixed = Seq( + Quad.create(iri("g1"), triples(0)), + Quad.create(Quad.defaultGraphIRI, triples(1)), + Quad.create(iri("g2"), triples(2)), + Quad.create(Quad.defaultGraphIRI, triples(0)), + Quad.create(iri("g1"), triples(1)), + ) + val expected = mixed.map { q => + Map("s" -> q.getSubject, "p" -> q.getPredicate, "o" -> q.getObject) ++ + (if q.isDefaultGraph then Map() else Map("g" -> q.getGraph)) + } + for + physicalType <- Seq(PhysicalStreamType.QUADS, PhysicalStreamType.GRAPHS) + // With 1 row per frame, the graph changes also happen across frame boundaries + rowsPerFrame <- Seq(256, 1) + do + withClue(s"$physicalType, $rowsPerFrame rows per frame: ") { + val rdf = writeRdf(physicalType, quads = mixed, rowsPerFrame = rowsPerFrame) + if rowsPerFrame == 1 then + JellyUtil.iterateRdfStream(ByteArrayInputStream(rdf)).size should be > 1 + val sparql = toSparql(rdf) + val (vars, rows) = readSparql(sparql) + vars shouldBe Seq("s", "p", "o", "g") + rows shouldBe expected + // And back: the unbound ?g must go to the default graph again + readRdf(toRdf(sparql)) shouldBe mixed.toSet + } + } + + "turn a GRAPHS stream into ?s ?p ?o ?g solutions" in { + val (vars, rows) = readSparql(toSparql(writeRdf(PhysicalStreamType.GRAPHS, quads = quads))) + vars shouldBe Seq("s", "p", "o", "g") + rows.size shouldBe quads.size + rows.flatMap(_.get("g")).toSet shouldBe Set(iri("g1"), iri("g2")) + } + + "announce the RDF version based on the input options by default" in { + for (tripleTerms, version) <- Seq( + false -> RdfVersion.RDF_VERSION_1_1, + true -> RdfVersion.RDF_VERSION_1_2, + ) + do + val jelly = toSparql(writeRdf(PhysicalStreamType.TRIPLES, triples, rdfStar = tripleTerms)) + val options = readFrames(jelly).head.getOptions + options.getRdfVersion shouldBe version + options.getMaxNameTableSize shouldBe JellySparqlOptions.BIG.getMaxNameTableSize + } + + "use the output options it is given" in { + val jelly = toSparql( + writeRdf(PhysicalStreamType.TRIPLES, triples), + _ => JellySparqlOptions.SMALL.clone().setStreamName("given"), + ) + val options = readFrames(jelly).head.getOptions + options.getStreamName shouldBe "given" + options.getMaxNameTableSize shouldBe JellySparqlOptions.SMALL.getMaxNameTableSize + } + + "keep triple terms" in { + val tripleTerm = NodeFactory.createTripleTerm(triples(0)) + val jelly = toSparql( + writeRdf(PhysicalStreamType.TRIPLES, Seq(Triple.create(iri("a"), iri("says"), tripleTerm))), + ) + readSparql(jelly)._2.head("o") shouldBe tripleTerm + } + + "split the output into frames" in { + val frames = readFrames( + toSparql(writeRdf(PhysicalStreamType.TRIPLES, triples), valuesPerFrame = 3), + ) + frames.map(_.getRowCount).sum shouldBe triples.size + frames.size should be > 1 + } + + "complain about empty input" in { + // The Jelly writer writes nothing at all when there are no statements + writeRdf(PhysicalStreamType.TRIPLES) shouldBe empty + val e = intercept[CriticalException](toSparql(Array())) + e.getMessage should include("Empty input stream") + } + + "complain about malformed input" in { + intercept[InvalidJellyFile](toSparql("not Jelly".getBytes(UTF_8))) + } + } + + "sparqlToRdf" should { + "turn 3 variables into a TRIPLES stream" in { + val jelly = toRdf(toSparql(writeRdf(PhysicalStreamType.TRIPLES, triples))) + rdfStreamOptions(jelly).getPhysicalType shouldBe PhysicalStreamType.TRIPLES + readRdf(jelly) shouldBe triplesAsQuads + } + + "turn 4 variables into a QUADS stream, with unbound graphs in the default graph" in { + val jelly = toRdf(toSparql(writeRdf(PhysicalStreamType.QUADS, quads = quads))) + rdfStreamOptions(jelly).getPhysicalType shouldBe PhysicalStreamType.QUADS + readRdf(jelly) shouldBe quads.toSet + } + + "take the variables by position, not by name" in { + val jelly = toRdf( + writeSparql( + Seq("x", "y", "z"), + Seq(Map("x" -> iriJson("a"), "y" -> iriJson("p"), "z" -> litJson("v"))), + ), + ) + readRdf(jelly) shouldBe Set(Quad.create(Quad.defaultGraphIRI, iri("a"), iri("p"), lit("v"))) + } + + "keep the physical type it is given" in { + val quadsSparql = toSparql(writeRdf(PhysicalStreamType.QUADS, quads = quads)) + val graphs = + toRdf(quadsSparql, defaultRdfOptions.setPhysicalType(PhysicalStreamType.GRAPHS)) + rdfStreamOptions(graphs).getPhysicalType shouldBe PhysicalStreamType.GRAPHS + readRdf(graphs) shouldBe quads.toSet + // 3 variables in a QUADS stream end up in the default graph + val triplesSparql = toSparql(writeRdf(PhysicalStreamType.TRIPLES, triples)) + val quadsRdf = + toRdf(triplesSparql, defaultRdfOptions.setPhysicalType(PhysicalStreamType.QUADS)) + rdfStreamOptions(quadsRdf).getPhysicalType shouldBe PhysicalStreamType.QUADS + readRdf(quadsRdf) shouldBe triplesAsQuads + } + + "refuse to write 4 variables as a TRIPLES stream" in { + val sparql = toSparql(writeRdf(PhysicalStreamType.QUADS, quads = quads)) + intercept[InvalidArgument]( + toRdf(sparql, defaultRdfOptions.setPhysicalType(PhysicalStreamType.TRIPLES)), + ) + } + + "reject a result set without 3 or 4 variables" in { + for vars <- Seq(Seq("a", "b"), Seq("a", "b", "c", "d", "e")) do + val e = intercept[CriticalException](toRdf(writeSparql(vars, Nil))) + e.getMessage should include("must have 3 or 4 variables") + } + + "reject an ASK result" in { + val e = intercept[CriticalException](toRdf(writeJelly(askJson(true)))) + e.getMessage should include("ASK result") + } + + "reject a solution with an unbound subject, predicate, or object" in { + val sparql = writeSparql( + Seq("s", "p", "o"), + Seq(Map("s" -> iriJson("a"), "o" -> litJson("v"))), + ) + val e = intercept[CriticalException](toRdf(sparql)) + e.getMessage should include("Solution 0 has no value for ?p") + } + + "reject generalized statements, unless they are enabled" in { + val sparql = writeSparql( + Seq("s", "p", "o"), + Seq(Map("s" -> litJson("lit"), "p" -> iriJson("p"), "o" -> litJson("v"))), + ) + val e = intercept[CriticalException](toRdf(sparql)) + e.getMessage should include("--opt.generalized-statements=true") + val jelly = toRdf(sparql, defaultRdfOptions.setGeneralizedStatements(true)) + rdfStreamOptions(jelly).getGeneralizedStatements shouldBe true + } + + "reject a literal graph name, unless generalized statements are enabled" in { + val sparql = writeSparql( + Seq("s", "p", "o", "g"), + Seq( + Map("s" -> iriJson("a"), "p" -> iriJson("p"), "o" -> litJson("v"), "g" -> litJson("g")), + ), + ) + val e = intercept[CriticalException](toRdf(sparql)) + e.getMessage should include("--opt.generalized-statements=true") + } + + "reject triple terms, unless they are enabled" in { + val tripleTerm = NodeFactory.createTripleTerm(triples(0)) + val sparql = toSparql( + writeRdf(PhysicalStreamType.TRIPLES, Seq(Triple.create(iri("a"), iri("says"), tripleTerm))), + ) + val e = intercept[CriticalException](toRdf(sparql, defaultRdfOptions.setRdfStar(false))) + e.getMessage should include("--opt.triple-terms=true") + } + + "reject a malformed input" in { + intercept[InvalidJellyFile](toRdf("not Jelly".getBytes(UTF_8))) + } + } diff --git a/src/test/scala/eu/neverblink/jelly/cli/util/jena/StatementBindingSpec.scala b/src/test/scala/eu/neverblink/jelly/cli/util/jena/StatementBindingSpec.scala new file mode 100644 index 0000000..4a4a855 --- /dev/null +++ b/src/test/scala/eu/neverblink/jelly/cli/util/jena/StatementBindingSpec.scala @@ -0,0 +1,86 @@ +package eu.neverblink.jelly.cli.util.jena + +import eu.neverblink.jelly.cli.command.helpers.TestFixtureHelper +import eu.neverblink.jelly.cli.util.jena.StatementBinding.* +import org.apache.jena.graph.NodeFactory +import org.apache.jena.sparql.core.Var +import org.apache.jena.sparql.engine.binding.{Binding, BindingFactory} +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +import scala.collection.mutable.ArrayBuffer +import scala.jdk.CollectionConverters.* + +class StatementBindingSpec extends AnyWordSpec with TestFixtureHelper with Matchers: + // Not used by these tests + protected val testCardinality: Int = 0 + + private val s = NodeFactory.createURI("http://example.org/s") + private val p = NodeFactory.createURI("http://example.org/p") + private val o = NodeFactory.createLiteralString("o") + private val g = NodeFactory.createURI("http://example.org/g") + + private def forEachPairs(b: Binding) = + val pairs = ArrayBuffer[(Var, AnyRef)]() + b.forEach((v, n) => pairs += (v -> n)) + pairs.toSeq + + "TripleBinding" should { + val binding = TripleBinding(s, p, o) + + "hold ?s ?p ?o, in that order" in { + binding.vars().asScala.toSeq shouldBe tripleVars + forEachPairs(binding) shouldBe Seq(subject -> s, predicate -> p, `object` -> o) + binding.size() shouldBe 3 + binding.isEmpty shouldBe false + } + + "find the values by variables that are not the same objects" in { + binding.get(Var.alloc("s")) shouldBe s + binding.get(Var.alloc("p")) shouldBe p + binding.get(Var.alloc("o")) shouldBe o + } + + "not hold ?g or any other variable" in { + binding.get(graph) shouldBe null + binding.contains(graph) shouldBe false + binding.get(Var.alloc("x")) shouldBe null + } + + "be equal to Jena's own binding" in { + val jena = BindingFactory.binding(subject, s, predicate, p, `object`, o) + binding shouldBe jena + jena shouldBe binding + binding.hashCode shouldBe jena.hashCode + binding.detach() shouldBe jena + binding.varsMentioned().asScala shouldBe jena.varsMentioned().asScala + binding.toString shouldBe jena.toString + } + } + + "QuadBinding" should { + val binding = QuadBinding(s, p, o, g) + + "hold ?s ?p ?o ?g, in that order" in { + binding.vars().asScala.toSeq shouldBe quadVars + forEachPairs(binding) shouldBe + Seq(subject -> s, predicate -> p, `object` -> o, graph -> g) + binding.size() shouldBe 4 + } + + "find the values by variables that are not the same objects" in { + binding.get(Var.alloc("g")) shouldBe g + binding.contains(Var.alloc("g")) shouldBe true + binding.get(Var.alloc("x")) shouldBe null + } + + "be equal to Jena's own binding" in { + val jena = BindingFactory.binding(subject, s, predicate, p, `object`, o, graph, g) + binding shouldBe jena + jena shouldBe binding + binding.hashCode shouldBe jena.hashCode + binding.detach() shouldBe jena + binding.varsMentioned().asScala shouldBe jena.varsMentioned().asScala + binding.toString shouldBe jena.toString + } + }