From 0b5e9eaffd112f62b01273d4c665959f78120f8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Thu, 15 Oct 2026 09:42:00 +0200 Subject: [PATCH 1/3] feat(io): add progressive line-based RDF parser for N-Triples and N-Quads --- .../impl/io/parser/nquads/NQuadsListener.java | 12 -- .../impl/io/parser/nquads/NQuadsParser.java | 14 +-- .../io/parser/ntriples/NTriplesParser.java | 23 +--- .../impl/io/parser/support/LineRDFParser.java | 63 ++++++++++ .../impl/io/parser/LineRDFParserTest.java | 110 ++++++++++++++++++ 5 files changed, 181 insertions(+), 41 deletions(-) create mode 100644 src/main/java/fr/inria/corese/core/next/data/impl/io/parser/support/LineRDFParser.java create mode 100644 src/test/java/fr/inria/corese/core/next/data/impl/io/parser/LineRDFParserTest.java diff --git a/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsListener.java b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsListener.java index 7db86a9f9..3317f1c28 100644 --- a/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsListener.java +++ b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsListener.java @@ -41,7 +41,6 @@ public NQuadsListener(Model model, ValueFactory factory, IOOptions options) { @Override public void enterStatement(NQuadsParser.StatementContext ctx) { - checkForInvalidDirectives(ctx); currentSubject = extractSubject(ctx.subject()); @@ -49,17 +48,6 @@ public void enterStatement(NQuadsParser.StatementContext ctx) { currentGraph = (ctx.graphLabel() != null) ? extractGraph(ctx.graphLabel()) : null; } - /** - * Checks for invalid directives in N-Quads format. - * N-Quads does not support @base or @prefix directives. - */ - private void checkForInvalidDirectives(NQuadsParser.StatementContext ctx) { - String text = ctx.getText(); - if (text != null && (text.contains("@base") || text.contains("@prefix"))) { - throw new ParsingException( - "Directives (@base, @prefix) are not allowed in N-Quads format"); - } - } @Override public void exitStatement(NQuadsParser.StatementContext ctx) { Value object = extractObject(ctx.object()); diff --git a/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsParser.java b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsParser.java index 31b301d6b..76dcee7fc 100644 --- a/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsParser.java +++ b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/nquads/NQuadsParser.java @@ -6,10 +6,9 @@ import fr.inria.corese.core.next.data.spi.io.parser.AbstractRDFParser; import fr.inria.corese.core.next.data.api.io.option.RDFParsingOptions; import fr.inria.corese.core.next.data.api.exception.ParsingException; +import fr.inria.corese.core.next.data.impl.io.parser.support.LineRDFParser; import fr.inria.corese.core.next.generated.antlr.NQuadsLexer; import org.antlr.v4.runtime.*; -import org.antlr.v4.runtime.tree.ParseTree; -import org.antlr.v4.runtime.tree.ParseTreeWalker; import java.io.IOException; import java.io.InputStream; @@ -20,7 +19,7 @@ /** * An ANTLR4-based parser for N-Quads format. * This parser uses an ANTLR grammar to tokenize and parse N-Quads documents, - * then a listener to build the RDF model. + * then a listener to populate the RDF model one physical line at a time. */ public class NQuadsParser extends AbstractRDFParser { @@ -65,20 +64,15 @@ public void parse(InputStream in, String baseURI) throws ParsingException { @Override public void parse(Reader reader, String baseURI) throws ParsingException { try { - CharStream charStream = CharStreams.fromReader(reader); - - NQuadsLexer lexer = new NQuadsLexer(charStream); + NQuadsLexer lexer = new NQuadsLexer(CharStreams.fromString("")); configureErrorHandling(lexer); CommonTokenStream tokens = new CommonTokenStream(lexer); fr.inria.corese.core.next.generated.antlr.NQuadsParser parser = new fr.inria.corese.core.next.generated.antlr.NQuadsParser(tokens); configureErrorHandling(parser); - ParseTree tree = parser.nquadsDoc(); - - ParseTreeWalker walker = new ParseTreeWalker(); NQuadsListener listener = new NQuadsListener(getModel(), getValueFactory(), getConfig()); - walker.walk(listener, tree); + LineRDFParser.parse(reader, lexer, parser, parser::nquadsDoc, listener); } catch (ParsingException e) { throw e; diff --git a/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/ntriples/NTriplesParser.java b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/ntriples/NTriplesParser.java index 96684a22d..37f7f1fbd 100644 --- a/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/ntriples/NTriplesParser.java +++ b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/ntriples/NTriplesParser.java @@ -6,12 +6,10 @@ import fr.inria.corese.core.next.data.spi.io.parser.AbstractRDFParser; import fr.inria.corese.core.next.data.api.io.option.RDFParsingOptions; import fr.inria.corese.core.next.data.api.exception.ParsingException; +import fr.inria.corese.core.next.data.impl.io.parser.support.LineRDFParser; import fr.inria.corese.core.next.generated.antlr.NTriplesLexer; -import org.antlr.v4.runtime.CharStream; import org.antlr.v4.runtime.CharStreams; import org.antlr.v4.runtime.CommonTokenStream; -import org.antlr.v4.runtime.tree.ParseTree; -import org.antlr.v4.runtime.tree.ParseTreeWalker; import org.antlr.v4.runtime.ConsoleErrorListener; import org.antlr.v4.runtime.misc.ParseCancellationException; import org.antlr.v4.runtime.BaseErrorListener; @@ -28,7 +26,7 @@ /** * An ANTLR4-based parser for the N-Triples format. * This parser uses an ANTLR grammar to tokenize and parse N-Triples documents, - * then a listener to build the RDF model. + * then a listener to populate the RDF model one physical line at a time. */ public class NTriplesParser extends AbstractRDFParser { @@ -73,16 +71,7 @@ public void parse(InputStream in, String baseURI) throws ParsingException { @Override public void parse(Reader reader, String baseURI) throws ParsingException { try { - CharStream charStream = CharStreams.fromReader(reader); - String input = charStream.toString(); - if (input.contains("@prefix")) { - throw new ParsingException("@prefix directives are not allowed in N-Triples"); - } - if (input.contains("@base")) { - throw new ParsingException("@base directives are not allowed in N-Triples"); - } - charStream = CharStreams.fromString(input); - NTriplesLexer lexer = new NTriplesLexer(charStream); + NTriplesLexer lexer = new NTriplesLexer(CharStreams.fromString("")); lexer.removeErrorListener(ConsoleErrorListener.INSTANCE); lexer.addErrorListener(new NTriplesErrorListener()); @@ -95,12 +84,8 @@ public void parse(Reader reader, String baseURI) throws ParsingException { antlrParser.setErrorHandler(new BailErrorStrategy()); antlrParser.addErrorListener(new NTriplesErrorListener()); - ParseTreeWalker walker = new ParseTreeWalker(); - ParseTree tree = antlrParser.ntriplesDoc(); - NTriplesListener listener = new NTriplesListener(getModel(), getValueFactory(), getConfig()); - - walker.walk(listener, tree); + LineRDFParser.parse(reader, lexer, antlrParser, antlrParser::ntriplesDoc, listener); } catch (ParseCancellationException pce) { if (pce.getCause() instanceof ParsingException cause) { diff --git a/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/support/LineRDFParser.java b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/support/LineRDFParser.java new file mode 100644 index 000000000..9cddc0482 --- /dev/null +++ b/src/main/java/fr/inria/corese/core/next/data/impl/io/parser/support/LineRDFParser.java @@ -0,0 +1,63 @@ +package fr.inria.corese.core.next.data.impl.io.parser.support; + +import fr.inria.corese.core.next.data.api.exception.ParsingException; +import org.antlr.v4.runtime.CharStreams; +import org.antlr.v4.runtime.CommonTokenStream; +import org.antlr.v4.runtime.Lexer; +import org.antlr.v4.runtime.Parser; +import org.antlr.v4.runtime.Token; +import org.antlr.v4.runtime.tree.ParseTree; +import org.antlr.v4.runtime.tree.ParseTreeListener; +import org.antlr.v4.runtime.tree.ParseTreeWalker; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.Reader; +import java.util.function.Supplier; + +/** + * Runs a line-oriented RDF grammar without retaining the complete input or its + * parse tree. Parser instances are confined to one call; buffers and syntax + * trees are bounded by the largest physical line, plus the reader buffer. + */ +public final class LineRDFParser { + + private LineRDFParser() { } + + /** + * Parses and emits each complete physical line before reading the next one. + * Valid prefixes may already have been added to the target when parsing fails; + * callers requiring atomicity must supply a transactional target. + * + * @param source caller-owned reader, never closed here + * @param lexer configured lexer, reused for each line + * @param parser configured parser, reused for each line + * @param document grammar entry point on {@code parser} + * @param listener statement-emitting listener, shared across lines + * @throws IOException if reading fails + * @throws ParsingException if a line is invalid or the thread is interrupted + */ + public static void parse(Reader source, Lexer lexer, Parser parser, + Supplier document, ParseTreeListener listener) throws IOException { + BufferedReader reader = source instanceof BufferedReader buffered ? buffered : new BufferedReader(source); + CommonTokenStream tokens = new CommonTokenStream(lexer); + ParseTreeWalker walker = new ParseTreeWalker(); + String line; + int lineNumber = 0; + while ((line = reader.readLine()) != null) { + if (Thread.currentThread().isInterrupted()) { + throw new ParsingException("RDF parsing was interrupted"); + } + lexer.setInputStream(CharStreams.fromString(line)); + lexer.setLine(++lineNumber); + tokens.setTokenSource(lexer); + parser.setTokenStream(tokens); + ParseTree tree = document.get(); + if (tokens.LA(1) != Token.EOF) { + throw new ParsingException("Unexpected RDF content at line " + lineNumber + + ":" + tokens.LT(1).getCharPositionInLine()); + } + walker.walk(listener, tree); + } + } +} diff --git a/src/test/java/fr/inria/corese/core/next/data/impl/io/parser/LineRDFParserTest.java b/src/test/java/fr/inria/corese/core/next/data/impl/io/parser/LineRDFParserTest.java new file mode 100644 index 000000000..5e73bd5eb --- /dev/null +++ b/src/test/java/fr/inria/corese/core/next/data/impl/io/parser/LineRDFParserTest.java @@ -0,0 +1,110 @@ +package fr.inria.corese.core.next.data.impl.io.parser; + +import fr.inria.corese.core.next.data.Values; +import fr.inria.corese.core.next.data.api.exception.ParsingException; +import fr.inria.corese.core.next.data.api.io.format.RDFFormat; +import fr.inria.corese.core.next.data.api.model.Model; +import fr.inria.corese.core.next.data.impl.model.LinkedHashModel; +import fr.inria.corese.core.next.io.CoreseIO; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import java.io.IOException; +import java.io.Reader; +import java.io.StringReader; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +class LineRDFParserTest { + static Stream formats() { + return Stream.of(RDFFormat.NTRIPLES, RDFFormat.NQUADS); + } + + @ParameterizedTest + @MethodSource("formats") + void emitsStatementsBeforeReadingTheWholeInput(RDFFormat format) { + GeneratedReader reader = new GeneratedReader(20_000); + long[] counts = { 0, 0 }; + Model model = mock(Model.class, withSettings().stubOnly().defaultAnswer(invocation -> { + if (invocation.getMethod().getName().equals("add")) { + if (counts[0]++ == 0) { + counts[1] = reader.position; + } + return true; + } + return RETURNS_DEFAULTS.answer(invocation); + })); + CoreseIO.parser(format, model, Values.factory()).parse(reader); + assertEquals(20_000, counts[0]); + assertTrue(counts[1] < reader.totalCharacters / 2); + assertFalse(reader.closed); + } + + @ParameterizedTest + @MethodSource("formats") + void acceptsBlankLinesCommentsLineEndingsAndDirectiveTextInsideLiterals(RDFFormat format) { + String input = "# comment\r\n\r\n \"@prefix @base\" .\r_:a _:a .\n \"escaped\\nvalue\" ."; + var model = CoreseIO.read(new StringReader(input), format); + assertEquals(3, model.size()); + assertTrue(model.contains(Values.factory().createIRI("urn:s"), Values.factory().createIRI("urn:p"), Values.factory().createLiteral("@prefix @base"))); + } + + @ParameterizedTest + @MethodSource("formats") + void rejectsTrailingContentAndReportsThePhysicalLine(RDFFormat format) { + var parser = CoreseIO.parser(format, new LinkedHashModel(Values.factory()), Values.factory()); + ParsingException failure = assertThrows(ParsingException.class, () -> parser.parse( + new StringReader("# header\n\n . garbage"))); + assertTrue(failure.getMessage().contains("3:"), failure.getMessage()); + assertThrows(ParsingException.class, () -> parser.parse(new StringReader("@prefix x: ."))); + } + + @ParameterizedTest + @MethodSource("formats") + void interruptedParsingFailsWithoutConsumingTheInterrupt(RDFFormat format) { + try { + Thread.currentThread().interrupt(); + assertThrows(ParsingException.class, () -> CoreseIO.read( + new StringReader(" .\n"), format)); + assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + } + + private static final class GeneratedReader extends Reader { + private static final String ROW = " .\n"; + private final long totalCharacters; + private long position; + private boolean closed; + + private GeneratedReader(long rows) { + totalCharacters = rows * ROW.length(); + } + + @Override + public int read(char[] buffer, int offset, int length) throws IOException { + if (closed) { + throw new IOException("Reader closed"); + } + if (length == 0) { + return 0; + } + if (position == totalCharacters) { + return -1; + } + int count = (int) Math.min(length, totalCharacters - position); + for (int index = 0; index < count; index++) { + buffer[offset + index] = ROW.charAt((int) (position++ % ROW.length())); + } + return count; + } + + @Override + public void close() { + closed = true; + } + } +} From 9a99445ccd67221a42414cbaf89b19f59b224296 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Thu, 15 Oct 2026 09:42:25 +0200 Subject: [PATCH 2/3] feat(repository): add direct load, export, and functional transaction helpers --- .../next/data/api/io/format/RDFFormat.java | 39 +++ .../next/query/api/repository/Repository.java | 167 ++++++++++ .../api/repository/RepositoryConnection.java | 202 ++++++++++++ .../CoreseRepositoryConnection.java | 92 ++++++ .../repository/RepositoryImportModel.java | 112 +++++++ .../data/api/io/format/RDFFormatTest.java | 17 + .../impl/repository/RepositoryIOTest.java | 295 ++++++++++++++++++ 7 files changed, 924 insertions(+) create mode 100644 src/main/java/fr/inria/corese/core/next/query/impl/repository/RepositoryImportModel.java create mode 100644 src/test/java/fr/inria/corese/core/next/query/impl/repository/RepositoryIOTest.java diff --git a/src/main/java/fr/inria/corese/core/next/data/api/io/format/RDFFormat.java b/src/main/java/fr/inria/corese/core/next/data/api/io/format/RDFFormat.java index 3f32d3c50..298d8b09f 100644 --- a/src/main/java/fr/inria/corese/core/next/data/api/io/format/RDFFormat.java +++ b/src/main/java/fr/inria/corese/core/next/data/api/io/format/RDFFormat.java @@ -4,6 +4,7 @@ import java.util.Locale; import java.util.Objects; import java.util.Optional; +import java.nio.file.Path; /** * Describes a semantic RDF serialization format, extending file-level metadata @@ -148,6 +149,44 @@ public static Optional byExtension(String extension) { .findFirst(); } + /** + * Looks up a format from the final extension of a file name, ignoring case. + * This is a name-based hint; the file is not opened or inspected. Compression + * suffixes are not interpreted. Ambiguous extensions require an explicit + * format. Canonical N-Quads is treated as N-Quads input. + * + * @param fileName file name, or {@code null} + * @return the matching input format, or empty for an ambiguous, unknown, or missing extension + */ + public static Optional forFileName(String fileName) { + if (fileName == null) { + return Optional.empty(); + } + int separator = Math.max(fileName.lastIndexOf('/'), fileName.lastIndexOf('\\')); + int dot = fileName.lastIndexOf('.'); + if (dot <= separator + 1) { + return Optional.empty(); + } + String extension = fileName.substring(dot + 1); + List candidates = all().stream() + .filter(format -> format != RDFC_1_0) + .filter(format -> format.getExtensions().stream().anyMatch(extension::equalsIgnoreCase)) + .limit(2) + .toList(); + return candidates.size() == 1 ? Optional.of(candidates.getFirst()) : Optional.empty(); + } + + /** + * Looks up a format from a path's file name without accessing the filesystem. + * + * @param path file path, or {@code null} + * @return the matching format, or empty when no supported extension is present + */ + public static Optional forPath(Path path) { + return path == null || path.getFileName() == null + ? Optional.empty() : forFileName(path.getFileName().toString()); + } + /** * Finds a known RDF format by MIME type (case-insensitive). * diff --git a/src/main/java/fr/inria/corese/core/next/query/api/repository/Repository.java b/src/main/java/fr/inria/corese/core/next/query/api/repository/Repository.java index 0842ab44c..412e03686 100644 --- a/src/main/java/fr/inria/corese/core/next/query/api/repository/Repository.java +++ b/src/main/java/fr/inria/corese/core/next/query/api/repository/Repository.java @@ -11,6 +11,15 @@ import java.util.ArrayList; import java.util.List; +import java.io.InputStream; +import java.io.Reader; +import java.io.OutputStream; +import java.io.Writer; +import java.nio.file.Path; +import java.util.function.Consumer; +import java.util.function.Function; +import fr.inria.corese.core.next.data.api.io.format.RDFFormat; +import fr.inria.corese.core.next.data.api.term.Resource; /** * A Corese repository that contains RDF data that can be queried and updated. @@ -142,4 +151,162 @@ default void update(String sparqlUpdate) throws QuerySyntaxException, Repository conn.prepareUpdate(sparqlUpdate).execute(); } } + + /** + * Loads a file using its extension, owning the connection and import transaction. + * + * @param source RDF file + * @see RepositoryConnection#load(Path) + */ + default void load(Path source) { + try (RepositoryConnection connection = getConnection()) { + connection.load(source); + } + } + + /** + * Loads RDF through a connection owned and closed by this call. + * Caller-owned streams and readers remain open. + * + * @param source RDF input + * @param format explicit RDF format + * @param contexts target graphs, or none to preserve source contexts + * @see RepositoryConnection#load(Path, RDFFormat, Resource...) + */ + default void load(Path source, RDFFormat format, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.load(source, format, contexts); + } + } + + /** + * Loads RDF through a connection owned and closed by this call. + * Caller-owned streams and readers remain open. + * + * @param source RDF input + * @param format explicit RDF format + * @param contexts target graphs, or none to preserve source contexts + * @see RepositoryConnection#load(InputStream, RDFFormat, Resource...) + */ + default void load(InputStream source, RDFFormat format, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.load(source, format, contexts); + } + } + + /** + * Loads RDF with an external base IRI in a transaction owned by this call. + * + * @param source caller-owned input, never closed here + * @param format explicit RDF format + * @param baseIRI external base IRI, or null + * @param contexts target graphs, or none to preserve source contexts + * @see RepositoryConnection#load(InputStream, RDFFormat, String, Resource...) + */ + default void load(InputStream source, RDFFormat format, String baseIRI, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.load(source, format, baseIRI, contexts); + } + } + + /** + * Loads RDF through a connection owned and closed by this call. + * Caller-owned streams and readers remain open. + * + * @param source RDF input + * @param format explicit RDF format + * @param contexts target graphs, or none to preserve source contexts + * @see RepositoryConnection#load(Reader, RDFFormat, Resource...) + */ + default void load(Reader source, RDFFormat format, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.load(source, format, contexts); + } + } + + /** + * Loads RDF with an external base IRI in a transaction owned by this call. + * + * @param source caller-owned input, never closed here + * @param format explicit RDF format + * @param baseIRI external base IRI, or null + * @param contexts target graphs, or none to preserve source contexts + * @see RepositoryConnection#load(Reader, RDFFormat, String, Resource...) + */ + default void load(Reader source, RDFFormat format, String baseIRI, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.load(source, format, baseIRI, contexts); + } + } + + /** + * Exports RDF through a connection owned and closed by this call. + * Caller-owned streams and writers remain open. + * + * @param destination RDF output + * @param format RDF format + * @param contexts selected graphs, or none for every graph + * @see RepositoryConnection#export(Path, RDFFormat, Resource...) + */ + default void export(Path destination, RDFFormat format, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.export(destination, format, contexts); + } + } + + /** + * Exports RDF through a connection owned and closed by this call. + * Caller-owned streams and writers remain open. + * + * @param destination RDF output + * @param format RDF format + * @param contexts selected graphs, or none for every graph + * @see RepositoryConnection#export(OutputStream, RDFFormat, Resource...) + */ + default void export(OutputStream destination, RDFFormat format, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.export(destination, format, contexts); + } + } + + /** + * Exports RDF through a connection owned and closed by this call. + * Caller-owned streams and writers remain open. + * + * @param destination RDF output + * @param format RDF format + * @param contexts selected graphs, or none for every graph + * @see RepositoryConnection#export(Writer, RDFFormat, Resource...) + */ + default void export(Writer destination, RDFFormat format, Resource... contexts) { + try (RepositoryConnection connection = getConnection()) { + connection.export(destination, format, contexts); + } + } + + /** + * Runs work in an owned connection and transaction, then closes the connection. + * + * @param action work that must not manually end the transaction or close the connection + * @see RepositoryConnection#inTransaction(Consumer) + */ + default void inTransaction(Consumer action) { + try (RepositoryConnection connection = getConnection()) { + connection.inTransaction(action); + } + } + + /** + * Runs work in an owned connection and transaction and returns its detached value. + * + * @param action work that must not manually end the transaction or close the connection + * @param return value type + * @return the callback's value after commit and connection closure + * @see RepositoryConnection#computeInTransaction(Function) + */ + default T computeInTransaction(Function action) { + try (RepositoryConnection connection = getConnection()) { + return connection.computeInTransaction(action); + } + } } diff --git a/src/main/java/fr/inria/corese/core/next/query/api/repository/RepositoryConnection.java b/src/main/java/fr/inria/corese/core/next/query/api/repository/RepositoryConnection.java index 0817c1124..d3645d4a7 100644 --- a/src/main/java/fr/inria/corese/core/next/query/api/repository/RepositoryConnection.java +++ b/src/main/java/fr/inria/corese/core/next/query/api/repository/RepositoryConnection.java @@ -2,6 +2,7 @@ import fr.inria.corese.core.next.data.api.factory.ValueFactory; import fr.inria.corese.core.next.data.api.model.Statement; +import fr.inria.corese.core.next.data.api.io.format.RDFFormat; import fr.inria.corese.core.next.data.api.term.IRI; import fr.inria.corese.core.next.data.api.term.Resource; import fr.inria.corese.core.next.data.api.term.Value; @@ -14,6 +15,17 @@ import fr.inria.corese.core.next.query.api.exception.RepositoryException; import fr.inria.corese.core.next.query.api.result.StatementResult; +import java.io.InputStream; +import java.io.Reader; +import java.io.OutputStream; +import java.io.Writer; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Objects; +import java.util.function.Consumer; +import java.util.function.Function; + /** * A connection to a Corese {@link Repository}, providing access to query * evaluation and data updates. @@ -74,6 +86,144 @@ public interface RepositoryConnection extends AutoCloseable { */ void add(Iterable statements) throws RepositoryException; + /** + * Loads a file using its extension as the format hint and its URI as the base + * IRI. The file opened by this method is closed before returning. + * + * @param source RDF file + * @throws IllegalArgumentException if the extension is ambiguous or not recognized + * @throws RepositoryException if the file cannot be read + * @throws fr.inria.corese.core.next.data.api.exception.ParsingException if parsing fails + */ + default void load(Path source) { + Objects.requireNonNull(source, "source"); + load(source, RDFFormat.forPath(source).orElseThrow( + () -> new IllegalArgumentException("Unknown or ambiguous RDF file extension: " + source))); + } + + /** + * Loads a file with an explicit format, using its URI as the base IRI. + * Context and transaction semantics are those of + * {@link #load(Reader, RDFFormat, String, Resource...)}. + * + * @param source RDF file, opened and closed by this method + * @param format explicit format, independent of the file extension + * @param contexts target graphs, or none to preserve source contexts + * @throws RepositoryException if the file cannot be read + * @throws fr.inria.corese.core.next.data.api.exception.ParsingException if parsing fails + */ + default void load(Path source, RDFFormat format, Resource... contexts) { + Objects.requireNonNull(source, "source"); + Objects.requireNonNull(format, "format"); + try (InputStream input = Files.newInputStream(source)) { + load(input, format, source.toAbsolutePath().normalize().toUri().toString(), contexts); + } catch (IOException failure) { + throw new RepositoryException("Could not read RDF from " + source, failure); + } + } + + /** + * Loads a caller-owned byte stream without an external base IRI. + * + * @param source input, never closed by this method + * @param format RDF format + * @param contexts target graphs, or none to preserve source contexts + * @see #load(Reader, RDFFormat, String, Resource...) + */ + default void load(InputStream source, RDFFormat format, Resource... contexts) { + load(source, format, null, contexts); + } + + /** + * Loads a caller-owned byte stream with an external base IRI. Context and + * transaction semantics are those of the reader overload. + * + * @param source input, never closed by this method + * @param format RDF format + * @param baseIRI external base IRI, or {@code null} + * @param contexts target graphs, or none to preserve source contexts + * @see #load(Reader, RDFFormat, String, Resource...) + */ + void load(InputStream source, RDFFormat format, String baseIRI, Resource... contexts); + + /** + * Loads a caller-owned reader without an external base IRI. + * + * @param source input, never closed by this method + * @param format RDF format + * @param contexts target graphs, or none to preserve source contexts + * @see #load(Reader, RDFFormat, String, Resource...) + */ + default void load(Reader source, RDFFormat format, Resource... contexts) { + load(source, format, null, contexts); + } + + /** + * Parses directly into this connection without an intermediate statement + * collection. N-Triples and N-Quads retain at most one physical line of parser + * input; other formats may buffer the document. The stored data still consumes + * memory proportional to its size. + * + *

With no active transaction, this method owns a transaction and rolls it + * back on failure. Otherwise it participates in the caller's transaction, + * without committing or rolling it back; a failure can leave an imported + * prefix in that transaction, which the caller must roll back. Transactional + * storage is required when this method needs to start a transaction.

+ * + *

No contexts preserves source graph names; explicit contexts replace them + * and replicate statements into each target. A {@code null} element selects + * the default graph. Blank nodes are scoped to this invocation, including + * blank graph names from the document. Prefix declarations are used during + * parsing but are not persisted as repository metadata.

+ * + * @param source input, never closed by this method + * @param format RDF format + * @param baseIRI external base IRI, or {@code null} + * @param contexts target graphs, or none to preserve source contexts + * @throws NullPointerException if the input, format, or contexts array is null + * @throws RepositoryException if storage access or transaction handling fails + * @throws fr.inria.corese.core.next.data.api.exception.ParsingException if parsing fails + */ + void load(Reader source, RDFFormat format, String baseIRI, Resource... contexts); + + /** + * Exports selected graphs to a UTF-8 file, closing the file opened here. + * A failed write may leave a partial file. + * + * @param destination output file + * @param format RDF format + * @param contexts selected graphs, or none for every graph + * @see #export(Writer, RDFFormat, Resource...) + */ + void export(Path destination, RDFFormat format, Resource... contexts); + + /** + * Exports selected graphs to a caller-owned byte stream without closing it. + * + * @param destination output stream + * @param format RDF format + * @param contexts selected graphs, or none for every graph + * @see #export(Writer, RDFFormat, Resource...) + */ + void export(OutputStream destination, RDFFormat format, Resource... contexts); + + /** + * Serializes a captured statement traversal and always closes its storage + * cursor, including on failure. N-Triples and N-Quads consume it progressively; + * other formats may materialize the selected data. A format without named graph + * support rejects named statements instead of silently discarding contexts. + * Failures may leave partial output; caller-owned outputs are never closed. + * + * @param destination output writer + * @param format RDF format + * @param contexts selected graphs; none means every graph, a null element the default graph + * @throws NullPointerException if a required argument or contexts array is null + * @throws fr.inria.corese.core.next.data.api.exception.SerializationException if serialization + * fails or the format cannot preserve the selected contexts + * @throws RepositoryException if storage access fails + */ + void export(Writer destination, RDFFormat format, Resource... contexts); + /** * Adds a statement assembled from RDF terms. With no context, the statement * is added to the default graph. With several contexts, one statement is @@ -309,6 +459,58 @@ Update prepareUpdate(String updateString) */ void rollback() throws RepositoryException; + /** + * Executes a callback in a transaction owned by this method. Nested use is + * rejected before invoking the callback. The callback must not close the + * connection or manually end the transaction, and runs on the calling thread. + * Callback and commit failures trigger rollback while the transaction is active; + * a rollback failure is suppressed on the original exception. + * + * @param action work to execute + * @throws NullPointerException if the callback is null + * @throws RepositoryException if a transaction is active or transaction handling fails + */ + default void inTransaction(Consumer action) { + Objects.requireNonNull(action, "action"); + computeInTransaction(connection -> { + action.accept(connection); + return null; + }); + } + + /** + * Executes work and returns its value after a successful commit. The lifecycle + * contract is the same as {@link #inTransaction(Consumer)}. Returned values must + * not depend on a cursor or transaction ended by this method. + * + * @param action work to execute + * @param return value type + * @return the callback's value after commit + * @throws NullPointerException if the callback is null + * @throws RepositoryException if a transaction is active or transaction handling fails + */ + @SuppressWarnings("java:S1181") // Roll back on callback Errors too, then rethrow the original failure unchanged. + default T computeInTransaction(Function action) { + Objects.requireNonNull(action, "action"); + begin(); + try { + T result = action.apply(this); + commit(); + return result; + } catch (RuntimeException | Error failure) { + try { + if (isActive()) { + rollback(); + } + } catch (RuntimeException | Error rollbackFailure) { + if (failure != rollbackFailure) { + failure.addSuppressed(rollbackFailure); + } + } + throw failure; + } + } + /** * Closes this connection and releases any resources it holds. Repeated calls * have no effect. An active transaction is rolled back before closing. diff --git a/src/main/java/fr/inria/corese/core/next/query/impl/repository/CoreseRepositoryConnection.java b/src/main/java/fr/inria/corese/core/next/query/impl/repository/CoreseRepositoryConnection.java index eff080a46..55177a450 100644 --- a/src/main/java/fr/inria/corese/core/next/query/impl/repository/CoreseRepositoryConnection.java +++ b/src/main/java/fr/inria/corese/core/next/query/impl/repository/CoreseRepositoryConnection.java @@ -2,6 +2,12 @@ import fr.inria.corese.core.next.data.api.factory.ValueFactory; import fr.inria.corese.core.next.data.api.model.Statement; +import fr.inria.corese.core.next.data.api.io.format.RDFFormat; +import fr.inria.corese.core.next.data.api.io.parser.RDFParser; +import fr.inria.corese.core.next.data.api.io.serializer.RDFSerializer; +import fr.inria.corese.core.next.data.api.exception.SerializationException; +import fr.inria.corese.core.next.io.CoreseIO; +import fr.inria.corese.core.next.storage.StorageModels; import fr.inria.corese.core.next.data.api.term.IRI; import fr.inria.corese.core.next.data.api.term.Resource; import fr.inria.corese.core.next.data.api.term.Value; @@ -37,6 +43,14 @@ import fr.inria.corese.core.next.storage.api.transaction.TransactionManager; import java.util.Objects; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.function.Consumer; +import java.io.InputStream; +import java.io.Reader; +import java.io.OutputStream; +import java.io.Writer; +import java.nio.file.Path; /** * Connection to a Corese repository. @@ -48,6 +62,8 @@ */ public final class CoreseRepositoryConnection implements RepositoryConnection { + private static final String PARAM_DESTINATION = "destination"; + private final Repository repository; private final StorageManager storage; private final NextSparqlPipelineExecutor executor; @@ -94,6 +110,82 @@ public void add(Iterable statements) throws RepositoryExcep } } + @Override + public void load(InputStream source, RDFFormat format, String baseIRI, Resource... contexts) { + Objects.requireNonNull(source, "source"); + importData(format, contexts, rdfParser -> rdfParser.parse(source, baseIRI)); + } + + @Override + public void load(Reader source, RDFFormat format, String baseIRI, Resource... contexts) { + Objects.requireNonNull(source, "source"); + importData(format, contexts, rdfParser -> rdfParser.parse(source, baseIRI)); + } + + private void importData(RDFFormat format, Resource[] contexts, Consumer action) { + checkOpen(); + Objects.requireNonNull(format, "format"); + Objects.requireNonNull(contexts, "contexts"); + RepositoryImportModel target = new RepositoryImportModel( + StorageModels.create(storage), getValueFactory(), contexts); + RDFParser rdfParser = CoreseIO.parser(format, target, getValueFactory()); + if (isActive()) { + action.accept(rdfParser); + } else { + inTransaction(connection -> action.accept(rdfParser)); + } + } + + @Override + public void export(Path destination, RDFFormat format, Resource... contexts) { + Objects.requireNonNull(destination, PARAM_DESTINATION); + exportData(format, contexts, serializer -> serializer.write(destination)); + } + + @Override + public void export(OutputStream destination, RDFFormat format, Resource... contexts) { + Objects.requireNonNull(destination, PARAM_DESTINATION); + exportData(format, contexts, serializer -> serializer.write(destination)); + } + + @Override + public void export(Writer destination, RDFFormat format, Resource... contexts) { + Objects.requireNonNull(destination, PARAM_DESTINATION); + exportData(format, contexts, serializer -> serializer.write(destination)); + } + + private void exportData(RDFFormat format, Resource[] contexts, Consumer action) { + checkOpen(); + Objects.requireNonNull(format, "format"); + Objects.requireNonNull(contexts, "contexts"); + try (StatementResult statements = getStatements(null, null, null, contexts)) { + Iterable source = format.supportsNamedGraphs() + ? statements : requireDefaultGraph(statements, format); + action.accept(CoreseIO.serializer(source, format)); + } + } + + private static Iterable requireDefaultGraph(StatementResult statements, RDFFormat format) { + return () -> new Iterator<>() { + @Override + public boolean hasNext() { + return statements.hasNext(); + } + + @Override + public Statement next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + Statement statement = statements.next(); + if (statement.getContext() != null) { + throw new SerializationException("Cannot preserve named graphs in " + format.getName(), format); + } + return statement; + } + }; + } + @Override public void add(Resource subject, IRI predicate, Value object, Resource... contexts) throws RepositoryException { diff --git a/src/main/java/fr/inria/corese/core/next/query/impl/repository/RepositoryImportModel.java b/src/main/java/fr/inria/corese/core/next/query/impl/repository/RepositoryImportModel.java new file mode 100644 index 000000000..c56562f71 --- /dev/null +++ b/src/main/java/fr/inria/corese/core/next/query/impl/repository/RepositoryImportModel.java @@ -0,0 +1,112 @@ +package fr.inria.corese.core.next.query.impl.repository; + +import fr.inria.corese.core.next.data.api.factory.ValueFactory; +import fr.inria.corese.core.next.data.api.model.Model; +import fr.inria.corese.core.next.data.api.model.Statement; +import fr.inria.corese.core.next.data.api.namespace.Namespace; +import fr.inria.corese.core.next.data.api.term.BNode; +import fr.inria.corese.core.next.data.api.term.IRI; +import fr.inria.corese.core.next.data.api.term.Resource; +import fr.inria.corese.core.next.data.api.term.Value; +import fr.inria.corese.core.next.data.spi.model.AbstractModel; + +import java.util.Iterator; +import java.util.Optional; +import java.util.Set; + +/** + * Parser-local model view that scopes source blank nodes and redirects writes to + * optional target contexts. No statement collection or blank-node dictionary is + * retained. Reads and namespace operations delegate to the live backing model. + * Instances are confined to the thread and lifetime of one import. + */ +@SuppressWarnings("java:S2160") // Inherit RDF set equality; adapter configuration is not part of statement equality. +final class RepositoryImportModel extends AbstractModel { + private final Model target; + private final ValueFactory factory; + private final Resource[] contexts; + private final String blankNodePrefix; + + /** + * Creates a view for one import. + * + * @param target live model over the importing connection's storage + * @param factory RDF value factory + * @param contexts non-null target contexts; empty preserves document contexts + */ + RepositoryImportModel(Model target, ValueFactory factory, Resource[] contexts) { + this.target = target; + this.factory = factory; + this.contexts = contexts.clone(); + String scope = factory.createBNode().getID(); + this.blankNodePrefix = scope.length() + "-" + scope + "-"; + } + + @Override + public boolean add(Resource subject, IRI predicate, Value object, Resource... sourceContexts) { + Resource scopedSubject = scope(subject); + Value scopedObject = object instanceof BNode blank ? scope(blank) : object; + if (contexts.length != 0) { + return target.add(scopedSubject, predicate, scopedObject, contexts); + } + if (sourceContexts == null || sourceContexts.length == 0) { + return target.add(scopedSubject, predicate, scopedObject); + } + boolean changed = false; + for (Resource context : sourceContexts) { + changed |= target.add(scopedSubject, predicate, scopedObject, scope(context)); + } + return changed; + } + + private Resource scope(Resource resource) { + return resource instanceof BNode blank + ? factory.createBNode(blankNodePrefix + blank.getID()) : resource; + } + + @Override + public boolean contains(Resource subject, IRI predicate, Value object, Resource... selected) { + return target.contains(subject, predicate, object, selected); + } + + @Override + public boolean remove(Resource subject, IRI predicate, Value object, Resource... selected) { + return target.remove(subject, predicate, object, selected); + } + + @Override + public Model filter(Resource subject, IRI predicate, Value object, Resource... selected) { + return target.filter(subject, predicate, object, selected); + } + + @Override + public void removeTermIteration(Iterator iterator, Resource subject, IRI predicate, + Value object, Resource... selected) { + target.remove(subject, predicate, object, selected); + } + + @Override + public Iterator iterator() { + return target.iterator(); + } + + @Override + public int size() { + return target.size(); + } + + @Override + public Set getNamespaces() { + return target.getNamespaces(); + } + + @Override + public void setNamespace(Namespace namespace) { + target.setNamespace(namespace); + } + + @Override + public Optional removeNamespace(String prefix) { + return target.removeNamespace(prefix); + } +} diff --git a/src/test/java/fr/inria/corese/core/next/data/api/io/format/RDFFormatTest.java b/src/test/java/fr/inria/corese/core/next/data/api/io/format/RDFFormatTest.java index 7b1426121..72ebe0731 100644 --- a/src/test/java/fr/inria/corese/core/next/data/api/io/format/RDFFormatTest.java +++ b/src/test/java/fr/inria/corese/core/next/data/api/io/format/RDFFormatTest.java @@ -5,11 +5,28 @@ import java.util.List; import java.util.Optional; +import java.nio.file.Path; import static org.junit.jupiter.api.Assertions.*; class RDFFormatTest { + @Test + void fileNameHintsUseOnlyTheFinalExtensionWithoutInspectingFiles() { + assertEquals(Optional.of(RDFFormat.TURTLE), RDFFormat.forFileName("data.TTL")); + assertEquals(Optional.of(RDFFormat.NQUADS), RDFFormat.forPath(Path.of("missing", "data.nq"))); + assertEquals(Optional.of(RDFFormat.NTRIPLES), RDFFormat.forFileName("C:\\data\\file.nt")); + assertTrue(RDFFormat.forFileName("directory.ttl/file").isEmpty()); + assertTrue(RDFFormat.forFileName("data.ttl.gz").isEmpty()); + assertEquals(Optional.of(RDFFormat.RDFXML), RDFFormat.forFileName("data.xml")); + assertEquals(Optional.of(RDFFormat.RDFXML), RDFFormat.forFileName("data.rdf")); + assertEquals(Optional.of(RDFFormat.RDFA), RDFFormat.forFileName("data.xhtml")); + assertTrue(RDFFormat.forFileName(".ttl").isEmpty()); + assertTrue(RDFFormat.forFileName("data.").isEmpty()); + assertTrue(RDFFormat.forFileName(null).isEmpty()); + assertTrue(RDFFormat.forPath(null).isEmpty()); + } + @Test @DisplayName("Constructor should correctly initialize fields and getters should return correct values") diff --git a/src/test/java/fr/inria/corese/core/next/query/impl/repository/RepositoryIOTest.java b/src/test/java/fr/inria/corese/core/next/query/impl/repository/RepositoryIOTest.java new file mode 100644 index 000000000..a8e6dabbf --- /dev/null +++ b/src/test/java/fr/inria/corese/core/next/query/impl/repository/RepositoryIOTest.java @@ -0,0 +1,295 @@ +package fr.inria.corese.core.next.query.impl.repository; + +import fr.inria.corese.core.next.data.api.exception.ParsingException; +import fr.inria.corese.core.next.data.api.exception.SerializationException; +import fr.inria.corese.core.next.data.api.io.format.RDFFormat; +import fr.inria.corese.core.next.data.api.term.BNode; +import fr.inria.corese.core.next.data.api.term.Resource; +import fr.inria.corese.core.next.data.Values; +import fr.inria.corese.core.next.query.Repositories; +import fr.inria.corese.core.next.query.api.exception.RepositoryException; +import fr.inria.corese.core.next.query.api.repository.RepositoryConnection; +import fr.inria.corese.core.next.storage.Storages; +import fr.inria.corese.core.next.storage.api.StorageManager; +import fr.inria.corese.core.next.storage.api.operations.QueryOperations; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.StringReader; +import java.io.StringWriter; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; +import static org.mockito.AdditionalAnswers.delegatesTo; + +class RepositoryIOTest { + private static final String TRIPLE = " .\n"; + private static final String QUAD = " .\n"; + + @TempDir + Path directory; + + @Test + void fileImportsDetectFormatsAndRespectExplicitOverrides() throws IOException { + Path file = directory.resolve("input.NT"); + Files.writeString(file, TRIPLE); + try (var repository = Repositories.create()) { + repository.load(file); + assertTrue(repository.ask("ASK { }")); + Path unknown = directory.resolve("input.unknown"); + Files.writeString(unknown, QUAD); + assertThrows(IllegalArgumentException.class, () -> repository.load(unknown)); + repository.load(unknown, RDFFormat.NQUADS); + assertTrue(repository.ask("ASK { GRAPH { } }")); + } + } + + @Test + void fileUriAndExplicitBaseResolveRelativeTurtleIris() throws IOException { + Path file = directory.resolve("input.ttl"); + Files.writeString(file, " ."); + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + connection.load(file); + assertTrue(connection.hasStatement(Values.factory().createIRI(directory.resolve("subject").toUri().toString()), + Values.factory().createIRI("urn:p"), Values.factory().createIRI(directory.resolve("object").toUri().toString()))); + connection.load(new StringReader(" ."), RDFFormat.TURTLE, "https://example.org/base/"); + assertTrue(connection.hasStatement(Values.factory().createIRI("https://example.org/base/s"), + Values.factory().createIRI("urn:p"), Values.factory().createIRI("https://example.org/base/o"))); + } + } + + @Test + void ownedImportRollsBackValidPrefixOnMalformedInput() { + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + connection.load(new StringReader(TRIPLE), RDFFormat.NTRIPLES); + assertThrows(ParsingException.class, () -> connection.load( + new StringReader(" .\ninvalid"), RDFFormat.NTRIPLES)); + assertEquals(1, connection.size()); + assertFalse(connection.isActive()); + connection.load(new StringReader(QUAD), RDFFormat.NQUADS); + assertEquals(2, connection.size()); + } + } + + @Test + void importJoinsActiveTransactionAndLeavesItsFailureToTheCaller() { + try (var repository = Repositories.create(); var writer = repository.getConnection(); + var reader = repository.getConnection()) { + writer.begin(); + writer.load(new StringReader(TRIPLE), RDFFormat.NTRIPLES); + assertEquals(0, reader.size()); + assertEquals(1, writer.size()); + assertThrows(ParsingException.class, () -> writer.load(new StringReader(QUAD + "invalid"), RDFFormat.NQUADS)); + assertTrue(writer.isActive()); + assertEquals(2, writer.size()); + writer.rollback(); + assertEquals(0, reader.size()); + writer.inTransaction(connection -> connection.load(new StringReader(TRIPLE), RDFFormat.NTRIPLES)); + assertEquals(1, reader.size()); + } + } + + @Test + void importRetargetsContextsIncludingExplicitDefaultGraph() { + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + connection.load(new StringReader(QUAD), RDFFormat.NQUADS, + Values.factory().createIRI("urn:target"), (Resource) null); + assertEquals(2, connection.size()); + assertEquals(0, connection.size(Values.factory().createIRI("urn:g"))); + assertEquals(1, connection.size(Values.factory().createIRI("urn:target"))); + assertEquals(1, connection.size((Resource) null)); + } + } + + @Test + void blankNodeLabelsAreSharedWithinOneImportAndSeparatedBetweenImports() { + String input = "_:a _:b _:a .\n_:b _:a _:a .\n"; + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + connection.load(new StringReader(input), RDFFormat.NQUADS); + connection.load(new StringReader(input), RDFFormat.NQUADS); + assertEquals(4, connection.size()); + try (var statements = connection.getStatements(null, null, null)) { + var first = statements.next(); + assertInstanceOf(BNode.class, first.getSubject()); + assertEquals(first.getSubject(), first.getContext()); + assertEquals(first.getObject(), statements.next().getSubject()); + } + } + } + + @Test + void borrowedInputsAndOutputsRemainOpen() { + try (var repository = Repositories.create()) { + var input = new ByteArrayInputStream(TRIPLE.getBytes(StandardCharsets.UTF_8)) { + @Override + public void close() { fail("Caller-owned input was closed"); } + }; + repository.load(input, RDFFormat.NTRIPLES); + var reader = new StringReader(TRIPLE) { + @Override + public void close() { fail("Caller-owned reader was closed"); } + }; + repository.load(reader, RDFFormat.NTRIPLES); + var output = new ByteArrayOutputStream() { + @Override + public void close() { fail("Caller-owned output was closed"); } + }; + repository.export(output, RDFFormat.NTRIPLES); + assertTrue(output.toString(StandardCharsets.UTF_8).contains("urn:s")); + var writer = new StringWriter() { + @Override + public void close() { fail("Caller-owned writer was closed"); } + }; + repository.export(writer, RDFFormat.NTRIPLES); + assertTrue(writer.toString().contains("urn:s")); + } + } + + @Test + void failedReadsRollBackAndDoNotCloseBorrowedInput() { + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + var reader = new StringReader(TRIPLE) { + @Override + public int read(char[] buffer, int offset, int length) throws IOException { + throw new IOException("input unavailable"); + } + + @Override + public void close() { fail("Caller-owned input was closed"); } + }; + assertThrows(ParsingException.class, () -> connection.load(reader, RDFFormat.NTRIPLES)); + assertFalse(connection.isActive()); + assertEquals(0, connection.size()); + } + } + + @Test + void exportRoundTripsDatasetAndSupportsContextSelection() throws IOException { + Path file = directory.resolve("backup.nq"); + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + connection.load(new StringReader(TRIPLE + QUAD), RDFFormat.NQUADS); + repository.export(file, RDFFormat.NQUADS); + connection.clear(); + repository.load(file); + assertEquals(2, connection.size()); + StringWriter writer = new StringWriter(); + connection.export(writer, RDFFormat.NTRIPLES, (Resource) null); + assertEquals(TRIPLE.trim(), writer.toString().trim()); + assertThrows(SerializationException.class, () -> connection.export(new StringWriter(), RDFFormat.TURTLE)); + assertThrows(SerializationException.class, () -> connection.export(new StringWriter(), RDFFormat.NTRIPLES)); + } + } + + @Test + void failedExportClosesTheCapturedCursor() { + var storage = Storages.create(); + var session = storage.openSession(); + var closed = new AtomicBoolean(); + var queries = mock(QueryOperations.class, delegatesTo(session.queries())); + doAnswer(call -> session.queries().find(call.getArgument(0)).onClose(() -> closed.set(true))) + .when(queries).find(any()); + var wrapped = mock(StorageManager.class, delegatesTo(session)); + when(wrapped.queries()).thenReturn(queries); + try (var repository = Repositories.create(storage); + var real = new CoreseRepositoryConnection(repository, wrapped)) { + real.load(new StringReader(TRIPLE), RDFFormat.NTRIPLES); + var writer = new StringWriter() { + @Override + public void write(String text) { throw new IllegalStateException("output unavailable"); } + + @Override + public void write(char[] buffer, int offset, int length) { + throw new IllegalStateException("output unavailable"); + } + }; + assertThrows(RuntimeException.class, () -> real.export(writer, RDFFormat.NTRIPLES)); + assertTrue(closed.get(), "Export failure must close the captured storage stream"); + real.clear(); + assertEquals(0, real.size()); + } + } + + @Test + void functionalTransactionsCommitResultsAndRollBackCallbackFailures() { + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + long count = repository.computeInTransaction(current -> { + current.load(new StringReader(TRIPLE), RDFFormat.NTRIPLES); + return current.size(); + }); + assertEquals(1, count); + IllegalStateException original = new IllegalStateException("callback failed"); + assertSame(original, assertThrows(IllegalStateException.class, () -> connection.inTransaction(current -> { + current.clear(); + throw original; + }))); + assertEquals(1, connection.size()); + assertFalse(connection.isActive()); + connection.begin(); + assertThrows(RepositoryException.class, () -> connection.inTransaction(current -> fail("Nested callback ran"))); + assertTrue(connection.isActive()); + connection.rollback(); + } + } + + @Test + void exportRetainsItsReadVersionDuringWriterMutations() { + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + StringBuilder input = new StringBuilder(); + for (int index = 0; index < 1_000; index++) { + input.append(" .\n"); + } + connection.load(new StringReader(input.toString()), RDFFormat.NTRIPLES); + var mutated = new AtomicBoolean(); + var writer = new StringWriter() { + @Override + public void write(char[] buffer, int offset, int length) { + if (mutated.compareAndSet(false, true)) { + connection.add(Values.factory().createIRI("urn:late"), + Values.factory().createIRI("urn:p"), Values.factory().createIRI("urn:o")); + } + super.write(buffer, offset, length); + } + }; + connection.export(writer, RDFFormat.NTRIPLES); + assertTrue(mutated.get()); + assertEquals(1_000, writer.toString().lines().count()); + assertFalse(writer.toString().contains("urn:late")); + assertEquals(1_001, connection.size()); + } + } + + @Test + void callbackErrorsAlsoRollBackAndReleaseTheOwnedTransaction() { + try (var repository = Repositories.create(); var connection = repository.getConnection()) { + AssertionError original = new AssertionError("callback failed"); + assertSame(original, assertThrows(AssertionError.class, () -> connection.inTransaction(current -> { + current.load(new StringReader(TRIPLE), RDFFormat.NTRIPLES); + throw original; + }))); + assertFalse(connection.isActive()); + assertEquals(0, connection.size()); + } + } + + @Test + void failedCommitKeepsOriginalFailureAndSuppressesRollbackFailure() { + var connection = mock(RepositoryConnection.class, CALLS_REAL_METHODS); + var commitFailure = new RepositoryException("commit failed"); + var rollbackFailure = new RepositoryException("rollback failed"); + when(connection.isActive()).thenReturn(true); + doThrow(commitFailure).when(connection).commit(); + doThrow(rollbackFailure).when(connection).rollback(); + assertSame(commitFailure, assertThrows(RepositoryException.class, + () -> connection.computeInTransaction(current -> 42))); + assertArrayEquals(new Throwable[] { rollbackFailure }, commitFailure.getSuppressed()); + verify(connection).begin(); + verify(connection).rollback(); + } +} From ba763bdfb1387395f78ce4057595b0bbe2314eeb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A9mi=20C=C3=A9r=C3=A8s?= Date: Thu, 15 Oct 2026 09:47:48 +0200 Subject: [PATCH 3/3] style(docs): remove consecutive blank line in README.md --- README.md | 1 - 1 file changed, 1 deletion(-) diff --git a/README.md b/README.md index dce9889fb..35c5d1d98 100644 --- a/README.md +++ b/README.md @@ -79,7 +79,6 @@ or use the default native backend (`memory`). **- [Storage Backend Plugin Guide](docs/STORAGE_PLUGIN_GUIDE.md)** - ## Contributing We welcome contributions! Here’s how to get involved: