Implement more Jelly-SPARQL commands

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.

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")

src/main/scala/eu/neverblink/jelly/cli/App.scala

@@ -35,4 +35,6 @@ object App extends CommandsEntryPoint:
     RdfValidate,
     SparqlFromJelly,
     SparqlToJelly,
+    SparqlInspect,
+    SparqlValidate,
   )

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

@@ -8,13 +8,15 @@ 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,
 }
 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

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("<file-to-convert>")
@@ -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,

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
+    }

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
     }

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,
+    )

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("<file-to-inspect>")
+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)
+    }