Skip to content

Use RDF Messages for flat distributions #23

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Ostrzyciel merged 1 commit into from
Jun 20, 2026
Merged

Use RDF Messages for flat distributions #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@ CI worker for processing RiverBench datasets and other repositories.

Please file issues, bugs, and other requests in the **[main issue tracker](https://github.com/RiverBench/RiverBench/issues)**.

This project uses Scala Next (3.8.4) and JDK 25.

## Usage
In the container, use `ci-worker <args>` to run the worker.

Expand Down
3 changes: 2 additions & 1 deletion src/main/scala/commands/DatasetDocGenCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,8 @@ object DatasetDocGenCommand extends Command:
|### Download links
|
|The dataset is published in a few size variants, each containing a specific number of stream elements.
|For each size, there are three distribution types available: flat (just an N-Triples/N-Quads file),
|For each size, there are three distribution types available: flat
|(an N-Triples/N-Quads file in the [RDF Message Log format](https://w3c-cg.github.io/rsp/spec/messages)),
|streaming (a .tar.gz archive with Turtle/TriG files, one file per stream element),
|and [Jelly](https://w3id.org/jelly) (a native binary format for streaming RDF).
|See the [documentation](../../documentation/dataset-release-format.md) for more details.
Expand Down
10 changes: 10 additions & 0 deletions src/main/scala/commands/PackageCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -328,6 +328,12 @@ object PackageCommand extends Command:
}
StreamUtil.broadcastSink(sinks)

/**
* RDF Messages delimiter.
* https://w3c-cg.github.io/rsp/spec/messages#message-delimiter
*/
private val messageDelimiter = "MESSAGE\n".getBytes

private def packageFlatSerializeFlow(metadata: MetadataInfo): Flow[(DatasetGraph, Long), ByteString, NotUsed] =
Flow[(DatasetGraph, Long)]
.map((ds, _) => ds.find().asScala.toSeq.sorted)
Expand All @@ -341,9 +347,13 @@ object PackageCommand extends Command:
val writer = StreamRDFWriter.getWriterStream(os, Lang.NQUADS)
quads.foreach(q => writer.quad(q))
writer.finish()
os.write(messageDelimiter)
os.toByteArray
})
.map(ByteString.fromArrayUnsafe)
// RDF Messages version announcement
// https://w3c-cg.github.io/rsp/spec/messages#version-announcement
.prepend(Source.single(ByteString("VERSION \"1.1-messages\"\n")))

/**
* Creates a sink that writes the data to flat files
Expand Down
2 changes: 2 additions & 0 deletions src/main/scala/util/io/SaveResult.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@ case class SaveResult(io: IOResult, name: String, size: Long, md5: String, sha1:

dType match
case DistType.Flat =>
// Flat files using the RDF Message Log format:
// https://w3c-cg.github.io/rsp/spec/messages#turtle
if mi.streamTypes.exists(_.elementType == ElementType.Triple) then
distRes.addProperty(RdfUtil.dcatMediaType, "application/n-triples")
else
Expand Down