This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 481fd774a4 TIKA-4889: TEXT input for inference bindings (#3159)
481fd774a4 is described below
commit 481fd774a40e5d2c6ed989b1656a84dbc1e91266
Author: Tim Allison <[email protected]>
AuthorDate: Mon Sep 14 17:47:58 2026 -0400
TIKA-4889: TEXT input for inference bindings (#3159)
---
CHANGES.txt | 9 ++
.../ROOT/pages/configuration/inference.adoc | 47 +++++++-
.../ROOT/pages/configuration/metadata-filters.adoc | 6 +-
.../src/main/java/org/apache/tika/cli/TikaCLI.java | 5 +
.../src/main/java/org/apache/tika/gui/TikaGUI.java | 5 +
.../apache/tika/cli/TikaCLITextVectorsTest.java | 134 +++++++++++++++++++++
.../apache/tika/metadata/TikaCoreProperties.java | 8 ++
.../tika/parser/inference/InferenceBinding.java | 13 ++
.../tika/parser/inference/InferenceDispatcher.java | 83 ++++++++++++-
.../tika/parser/inference/InferenceUnit.java | 26 +++-
.../apache/tika/parser/inference/TextChunker.java | 23 ++--
.../parser/inference/InferenceTextStageTest.java | 127 +++++++++++++++++++
.../apache/tika/metadata/metadata-key-fields.json | 1 +
.../org/apache/tika/metadata/metadata-keys.json | 1 +
.../main/java/org/apache/tika/inference/Chunk.java | 10 ++
.../org/apache/tika/inference/ChunkSerializer.java | 6 +
.../java/org/apache/tika/inference/EmbedTask.java | 120 ++++++++++++++++--
.../org/apache/tika/inference/EmbeddingEngine.java | 4 +
.../org/apache/tika/inference/MarkdownChunker.java | 54 ++++++++-
.../tika/inference/OpenAIEmbeddingEngine.java | 35 ++++--
.../apache/tika/inference/ChunkSerializerTest.java | 10 ++
.../org/apache/tika/inference/EmbedTaskTest.java | 56 +++++++++
.../apache/tika/inference/MarkdownChunkerTest.java | 21 ++++
.../tika/inference/OpenAIEmbeddingEngineTest.java | 13 ++
.../apache/tika/parser/pdf/AbstractPDF2XHTML.java | 6 +
.../org/apache/tika/parser/pdf/PDFParserTest.java | 3 +
.../tika/pipes/core/server/ConnectionHandler.java | 3 +-
.../apache/tika/pipes/core/server/EmitHandler.java | 29 ++++-
.../apache/tika/pipes/core/server/PipesServer.java | 6 +-
.../pipes/core/server/SharedServerResources.java | 13 +-
.../core/server/EmitHandlerUserMetadataTest.java | 2 +-
.../apache/tika/pipes/core/PipesClientTest.java | 20 +++
.../configs/tika-config-inference-bindings.json | 5 +-
.../apache/tika/config/loader/InferenceLoader.java | 33 ++++-
.../tika/config/loader/InferenceLoaderTest.java | 53 ++++++++
.../org/apache/tika/config/loader/TestChunker.java | 35 ++++--
.../server/standard/InferenceBindingsWireTest.java | 30 +++++
.../configs/tika-config-inference-bindings.json | 3 +-
38 files changed, 993 insertions(+), 65 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 6c214ece66..3f0127b135 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,10 +1,19 @@
Release 4.1.0 - unreleased
+ * TEXT input for inference bindings: a binding on TEXT with a chunker
+ (markdown-chunker) embeds the text of every document in a tree in
+ batched requests, after the parse and before the metadata filters; each
+ chunk lands on its document with a text locator. Chunks now name the
+ binding that wrote them as "producer". A PDF released as PAGES records
+ tk:inference-released and is not embedded twice. The embedding filters
+ keep working (TIKA-4889).
+
* "pdf-parser": {"maxRenderedPages": N} bounds how many pages the
RENDER_PAGES_BEFORE_PARSE and RENDER_PAGES_AT_PAGE_END image strategies
render, counted from the first page and independent of maxPages: text
extraction can cover the whole document while only the first page is
rendered, as for a thumbnail (TIKA-4856).
+
* PDF: a bookmark outline is now walked without recursion, and lists nest at
most 50 deep with deeper items joining the deepest list (TIKA-4894).
diff --git a/docs/modules/ROOT/pages/configuration/inference.adoc
b/docs/modules/ROOT/pages/configuration/inference.adoc
index c2069243e6..a4439cde7a 100644
--- a/docs/modules/ROOT/pages/configuration/inference.adoc
+++ b/docs/modules/ROOT/pages/configuration/inference.adoc
@@ -68,11 +68,14 @@ A list of bindings. Each names an engine, says what it is
fed, and what to ask:
picture, an image attachment, a top-level image file; a page render emitted
as an embedded
document is not one. `PAGES` is one image per rendered page of a document
whose parser
renders pages for inference: the PDF parser, when its `inference.input` says
so (see
- <<pages>>). (`TEXT` and `MEDIA` follow.)
+ <<pages>>). `TEXT` is the extracted text of every document in the tree (see
<<text>>).
+ (`MEDIA` follows.)
`tasks`:: One call per task on the same engine and input. `embed` writes one
vector chunk per
unit into `tk:chunks`, with the page locator when the unit has a page.
`maxChunks`:: How many units one document tree may send; for `PAGES`, how many
pages of one
- document; -1 for no limit.
+ document; for `TEXT`, how many chunks across the tree; -1 for no limit.
+`chunker`:: `TEXT` only: how a document's text is cut, `{ "markdown-chunker":
{ "maxChunkChars":
+ 1500, "overlapChars": 200 } }`. Without one the whole text is one chunk.
`maxBytes`:: Largest unit the binding accepts (20 MB unless set; -1 for no
limit); larger
ones are skipped and counted.
`enabled`:: `false` keeps the binding configured but idle until a request
selects it.
@@ -151,6 +154,46 @@ Per request, like every PDF setting:
"inference": { "bindings": ["page-vectors"] } } }
----
+[#text]
+== Text
+
+A `TEXT` binding embeds the extracted text of every document in the tree, the
container and
+its attachments alike:
+
+[source,json]
+----
+{
+ "engines": {
+ "embedder": { "openai-embedding-engine": { "baseUrl": "http://embed:8000";,
"model": "bge-m3" } }
+ },
+ "inference": [
+ { "id": "text-vectors", "engine": "embedder", "input": "TEXT", "tasks":
["embed"],
+ "chunker": { "markdown-chunker": { "maxChunkChars": 1500,
"overlapChars": 200 } },
+ "maxChunks": 500 }
+ ]
+}
+----
+
+It runs after the parse, over the finished metadata list and before the
metadata filters
+(so a filter shapes what it wrote and cannot starve it): every document's
`tk:content` is chunked, the chunks of the whole tree are sent in
+`maxBatchSize` requests in document order (a request may hold a container's
last chunks and
+an attachment's first; the endpoint treats inputs independently and answers by
index), and
+each chunk lands on the document its text came from, with a `text` locator and
the binding
+as its `producer`. A rejected request is retried one document at a time.
`maxChunks` caps the
+chunks of the tree, `maxBytes` the text of one document, and
`_mime-include`/`_mime-exclude`
+narrow by document type. Use a `MARKDOWN` content handler for the chunker to
see headings.
+
+A document that released other inputs to inference is not embedded twice: a
PDF parsed with
+`inference.input: ["PAGES"]` records `tk:inference-released: PAGES` and the
TEXT stage skips
+it; list `["TEXT", "PAGES"]` to get both. Because this stage runs over the
metadata list, it
+covers `-J`, `/rmeta` and pipes; the streaming `/tika` endpoint has no list
and no text
+vectors. It is not a metadata filter: a per-request `metadata-filters` list
does not touch it,
+and the per-request `inference` block below is the way to switch it off.
+
+The 4.0 `openai-embedding-filter` and `jina-embedding-filter` keep working and
embed one
+document per request; the binding batches the tree and shares the engine with
the other
+bindings.
+
== Per request
[source,json]
diff --git a/docs/modules/ROOT/pages/configuration/metadata-filters.adoc
b/docs/modules/ROOT/pages/configuration/metadata-filters.adoc
index 262fee13a9..dc19f1fd8b 100644
--- a/docs/modules/ROOT/pages/configuration/metadata-filters.adoc
+++ b/docs/modules/ROOT/pages/configuration/metadata-filters.adoc
@@ -115,8 +115,10 @@ Adding the module to the classpath is enough to make the
name resolvable.
|`openai-embedding-filter`, `jina-embedding-filter`
|`tika-inference`
-|Chunks `tk:content` and calls an embeddings endpoint, putting the vectors on
`tk:chunks`. See
-xref:migration-to-4x/chunk-strategies.adoc[Chunk Strategies].
+|Chunks `tk:content` and calls an embeddings endpoint, putting the vectors on
`tk:chunks`, one
+document per request. Since 4.1.0 a `TEXT`
xref:configuration/inference.adoc#text[inference
+binding] does the same for the whole tree in batched requests; it runs before
this chain and
+is not part of it. See xref:migration-to-4x/chunk-strategies.adoc[Chunk
Strategies].
|`charsoup-metadata-filter`, `optimaize-metadata-filter`,
`open-nlp-metadata-filter`
|`tika-langdetect-charsoup`, `-optimaize`, `-opennlp`
diff --git a/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java
b/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java
index 0a8284ff16..97989e2c11 100644
--- a/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java
+++ b/tika-app/src/main/java/org/apache/tika/cli/TikaCLI.java
@@ -96,6 +96,7 @@ import org.apache.tika.parser.ParserDecorator;
import org.apache.tika.parser.PasswordProvider;
import org.apache.tika.parser.RecursiveParserWrapper;
import org.apache.tika.parser.digestutils.CommonsDigesterFactory;
+import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.pipes.api.ParseMode;
import org.apache.tika.pipes.fork.PipesForkParser;
import org.apache.tika.pipes.fork.PipesForkParserConfig;
@@ -740,6 +741,10 @@ public class TikaCLI {
JsonMetadataList.setPrettyPrinting(prettyPrint);
try (Writer writer = getOutputWriter(output, encoding)) {
List<Metadata> metadataList = handler.getMetadataList();
+ InferenceDispatcher inference =
tikaLoader.get(InferenceDispatcher.class);
+ if (inference != null) {
+ inference.text(metadataList, context);
+ }
tikaLoader.loadMetadataFilters().filter(metadataList);
JsonMetadataList.toJson(metadataList, writer);
}
diff --git a/tika-app/src/main/java/org/apache/tika/gui/TikaGUI.java
b/tika-app/src/main/java/org/apache/tika/gui/TikaGUI.java
index a1634a75f0..7eeeff2c41 100644
--- a/tika-app/src/main/java/org/apache/tika/gui/TikaGUI.java
+++ b/tika-app/src/main/java/org/apache/tika/gui/TikaGUI.java
@@ -85,6 +85,7 @@ import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.Parser;
import org.apache.tika.parser.RecursiveParserWrapper;
+import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.sax.BasicContentHandlerFactory;
import org.apache.tika.sax.BodyContentHandler;
import org.apache.tika.sax.ContentHandlerDecorator;
@@ -390,6 +391,10 @@ public class TikaGUI extends JFrame implements
ActionListener, HyperlinkListener
StringWriter jsonBuffer = new StringWriter();
JsonMetadataList.setPrettyPrinting(true);
List<Metadata> metadataList =
recursiveParserWrapperHandler.getMetadataList();
+ InferenceDispatcher inference =
tikaConfig.get(InferenceDispatcher.class);
+ if (inference != null) {
+ inference.text(metadataList, rpwContext);
+ }
tikaConfig.loadMetadataFilters().filter(metadataList);
JsonMetadataList.toJson(metadataList, jsonBuffer);
setText(json, jsonBuffer.toString());
diff --git
a/tika-app/src/test/java/org/apache/tika/cli/TikaCLITextVectorsTest.java
b/tika-app/src/test/java/org/apache/tika/cli/TikaCLITextVectorsTest.java
new file mode 100644
index 0000000000..899e7ac1c4
--- /dev/null
+++ b/tika-app/src/test/java/org/apache/tika/cli/TikaCLITextVectorsTest.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.cli;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.http.TikaTestHttpServer;
+import org.apache.tika.inference.Chunk;
+import org.apache.tika.inference.ChunkSerializer;
+
+/**
+ * The consumer's path for text: a TEXT binding with a chunker, and -J shows
every document of
+ * a zip with its own text vectors, all from one request that mixed the
documents' chunks.
+ */
+public class TikaCLITextVectorsTest {
+
+ /** Vector[0] is the input index, so placement is checkable. */
+ private static final String EMBEDDING_RESPONSE = "{\"data\":["
+ + "{\"index\":3,\"embedding\":[3.0,0.5]},"
+ + "{\"index\":0,\"embedding\":[0.0,0.5]},"
+ + "{\"index\":2,\"embedding\":[2.0,0.5]},"
+ + "{\"index\":1,\"embedding\":[1.0,0.5]}]}";
+
+ @TempDir
+ Path tmp;
+
+ @Test
+ public void testOneRequestForTheTreeVectorsOnEachDocument() throws
Exception {
+ Path zip = tmp.resolve("two-notes.zip");
+ try (ZipOutputStream out = new
ZipOutputStream(Files.newOutputStream(zip))) {
+ entry(out, "alpha.txt", "alpha is the first note");
+ entry(out, "beta.txt", "beta is the second note");
+ }
+ try (TikaTestHttpServer server = new TikaTestHttpServer()) {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200,
EMBEDDING_RESPONSE));
+ Path config = tmp.resolve("tika-config.json");
+ Files.writeString(config, """
+ {
+ "engines": {
+ "embedder": { "openai-embedding-engine": { "baseUrl":
"BASE_URL", "model": "bge" } }
+ },
+ "inference": [
+ { "id": "text-vectors", "engine": "embedder", "input":
"TEXT", "tasks": ["embed"],
+ "chunker": { "markdown-chunker": { "maxChunkChars":
2000, "overlapChars": 0 } } }
+ ]
+ }
+ """.replace("BASE_URL", server.url()));
+
+ JsonNode output = new ObjectMapper().readTree(
+ run("--config=" + config, "-J", zip.toUri().toString()));
+
+ assertEquals(3, output.size(), "the zip and its two notes");
+ assertEquals(1, server.getRequestCount(), "one request for the
whole tree");
+ JsonNode request = new
ObjectMapper().readTree(server.takeRequest().body());
+ // the zip's own markdown content is two headings, so two chunks;
one per note
+ assertEquals(4, request.get("input").size(), "chunks of all three
documents in one request");
+ assertEquals("# alpha.txt", request.get("input").get(0).asText());
+ assertEquals("alpha is the first note",
request.get("input").get(2).asText());
+
+ for (JsonNode document : output) {
+ String content = text(document, "tk:content");
+ List<Chunk> chunks = ChunkSerializer.fromJson(text(document,
"tk:chunks"));
+ boolean container = content.startsWith("# alpha.txt");
+ assertEquals(container ? 2 : 1, chunks.size(), content);
+ for (Chunk chunk : chunks) {
+ assertEquals("text-vectors", chunk.getProducer());
+ assertEquals(2, chunk.getVector().length);
+ assertTrue(content.contains(chunk.getText()), "the chunk
is this document's own text");
+ int start =
chunk.getLocators().getText().get(0).getStartOffset();
+ assertEquals(chunk.getText(), content.substring(start,
start + chunk.getText().length()));
+ }
+ if (content.startsWith("beta")) {
+ assertEquals(3.0f, chunks.get(0).getVector()[0], "fourth
input, placed by index");
+ }
+ }
+ }
+ }
+
+ private static void entry(ZipOutputStream zip, String name, String text)
throws Exception {
+ zip.putNextEntry(new ZipEntry(name));
+ zip.write(text.getBytes(UTF_8));
+ zip.closeEntry();
+ }
+
+ private static String text(JsonNode node, String field) {
+ JsonNode value = node.get(field);
+ assertNotNull(value, field);
+ return value.isArray() ? value.get(0).asText() : value.asText();
+ }
+
+ private static String run(String... args) throws Exception {
+ PrintStream stdout = System.out;
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try {
+ System.setOut(new PrintStream(out, true, UTF_8.name()));
+ TikaCLI.main(args);
+ } finally {
+ System.setOut(stdout);
+ }
+ String json = out.toString(UTF_8);
+ assertTrue(json.startsWith("["), json);
+ return json;
+ }
+}
diff --git
a/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java
b/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java
index 7470a06cad..c528f3f34d 100644
--- a/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java
+++ b/tika-core/src/main/java/org/apache/tika/metadata/TikaCoreProperties.java
@@ -152,6 +152,14 @@ public interface TikaCoreProperties {
* of document content for downstream indexing and semantic search.
*/
Property TIKA_CHUNKS = Property.reservedInternalText("tk:chunks");
+
+ /**
+ * The inference inputs a parser released for this document (TEXT, PAGES,
...), written
+ * only when they differ from the default of its text. The TEXT stage
skips a document
+ * whose list lacks TEXT, so a PDF released as pages is not embedded twice.
+ */
+ Property INFERENCE_RELEASED =
+ Property.reservedInternalTextBag(TIKA_META_PREFIX +
"inference-released");
/**
* Use this to store parse exception information in the Metadata object.
*/
diff --git
a/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceBinding.java
b/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceBinding.java
index 90e8d86efb..a9ca185223 100644
---
a/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceBinding.java
+++
b/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceBinding.java
@@ -34,10 +34,18 @@ public final class InferenceBinding {
private final int maxChunks;
private final long maxBytes;
private final boolean enabled;
+ private final TextChunker chunker;
public InferenceBinding(String id, String engine, InputKind input,
List<String> tasks,
Set<MediaType> include, Set<MediaType> exclude,
int maxChunks,
long maxBytes, boolean enabled) {
+ this(id, engine, input, tasks, include, exclude, maxChunks, maxBytes,
enabled, null);
+ }
+
+ public InferenceBinding(String id, String engine, InputKind input,
List<String> tasks,
+ Set<MediaType> include, Set<MediaType> exclude,
int maxChunks,
+ long maxBytes, boolean enabled, TextChunker
chunker) {
+ this.chunker = chunker;
this.id = id;
this.engine = engine;
this.input = input;
@@ -79,6 +87,11 @@ public final class InferenceBinding {
return enabled;
}
+ /** How a TEXT binding cuts a document's text; null means the whole text
is one chunk. */
+ public TextChunker getChunker() {
+ return chunker;
+ }
+
public boolean accepts(InputKind kind, MediaType type) {
if (kind != input || type == null) {
return false;
diff --git
a/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceDispatcher.java
b/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceDispatcher.java
index c5a7e5c5a9..6db6bd0d26 100644
---
a/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceDispatcher.java
+++
b/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceDispatcher.java
@@ -17,6 +17,7 @@
package org.apache.tika.parser.inference;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
@@ -36,6 +37,7 @@ import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.extractor.EmbeddedMetadataLookup;
import org.apache.tika.io.TemporaryResources;
+import org.apache.tika.metadata.HttpHeaders;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.mime.MediaType;
@@ -51,7 +53,8 @@ import org.apache.tika.parser.hook.ParseHook;
* dispatcher owns until the flush. At the flush every unit is re-aimed at the
metadata the
* recursive wrapper kept for its document, since the parser's own copy is no
longer read. A
* request narrows the bindings through {@link InferenceSelection}. Failures
mark the document
- * and never fail the parse.
+ * and never fail the parse. TEXT is not offered during the parse: whoever
holds the finished
+ * metadata list runs {@link #text} over it.
*
* @since Apache Tika 4.1
*/
@@ -248,6 +251,84 @@ public final class InferenceDispatcher implements
ParseHook, TransientParseState
return kept;
}
+ /**
+ * The {@link InputKind#TEXT} stage. Runs the TEXT bindings over the
finished metadata list of
+ * a document tree, so every document's text goes to the engine in one
pass and a batch may
+ * hold chunks of the container and of its attachments alike. Call it
after the parse and
+ * before any metadata filters. A document that released other inputs to
inference
+ * ({@link TikaCoreProperties#INFERENCE_RELEASED} without TEXT) is
skipped. Failures mark
+ * the first document of the list.
+ */
+ public void text(List<Metadata> metadataList, ParseContext context) throws
TikaException {
+ if (metadataList.isEmpty()) {
+ return;
+ }
+ Metadata root = metadataList.get(0);
+ for (Bound b : running(InputKind.TEXT, context)) {
+ InferenceBinding binding = b.binding();
+ List<InferenceUnit> units = new ArrayList<>();
+ int skipped = 0;
+ for (Metadata metadata : metadataList) {
+ String text = metadata.get(TikaCoreProperties.TIKA_CONTENT);
+ if (text == null || text.isBlank() || !releasedText(metadata))
{
+ continue;
+ }
+ MediaType type =
MediaType.parse(metadata.get(HttpHeaders.CONTENT_TYPE));
+ if (!binding.accepts(InputKind.TEXT,
+ type == null ? MediaType.OCTET_STREAM : type)) {
+ continue;
+ }
+ if (binding.getMaxBytes() >= 0
+ && text.getBytes(StandardCharsets.UTF_8).length >
binding.getMaxBytes()) {
+ skipped++;
+ continue;
+ }
+ units.add(new InferenceUnit(type, metadata, null, text));
+ }
+ if (skipped > 0) {
+ root.add(TikaCoreProperties.TIKA_META_EXCEPTION_WARNING,
"inference binding "
+ + binding.getId() + " over maxBytes: skipped " +
skipped + " units");
+ }
+ if (units.isEmpty()) {
+ continue;
+ }
+ for (InferenceTask task : b.tasks()) {
+ try {
+ task.run(binding, units, b.engine(), context);
+ } catch (Exception e) {
+ LOG.warn("inference binding {} failed", binding.getId(),
e);
+ root.add(TikaCoreProperties.TIKA_META_EXCEPTION_WARNING,
+ "inference binding " + binding.getId() + ": " +
e.getMessage());
+ }
+ }
+ }
+ }
+
+ /** Absent means the parser released its text; a list without TEXT means
it did not. */
+ private static boolean releasedText(Metadata metadata) {
+ String[] released =
metadata.getValues(TikaCoreProperties.INFERENCE_RELEASED);
+ if (released.length == 0) {
+ return true;
+ }
+ for (String r : released) {
+ if (InputKind.TEXT.name().equals(r)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** The bindings of this kind that run for the request: enabled and
selected. */
+ private List<Bound> running(InputKind kind, ParseContext context) throws
TikaException {
+ List<Bound> result = new ArrayList<>();
+ for (Bound b : bound) {
+ if (b.binding().getInput() == kind && runs(b.binding(), context)) {
+ result.add(b);
+ }
+ }
+ return result;
+ }
+
private boolean runs(InferenceBinding binding, ParseContext context)
throws TikaException {
if (!binding.isEnabled()) {
return false;
diff --git
a/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceUnit.java
b/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceUnit.java
index 0ab0a76934..4724bbc1a1 100644
---
a/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceUnit.java
+++
b/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceUnit.java
@@ -17,6 +17,7 @@
package org.apache.tika.parser.inference;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -40,6 +41,7 @@ public final class InferenceUnit {
private final String targetIdPath;
private final String parentIdPath;
private final Path path;
+ private final String text;
private final long size;
private final int page;
@@ -53,12 +55,20 @@ public final class InferenceUnit {
Path path, int page) throws IOException {
this(kind, type, target, parent,
target.get(TikaCoreProperties.EMBEDDED_ID_PATH),
parent == null ? null :
parent.get(TikaCoreProperties.EMBEDDED_ID_PATH), path,
- Files.size(path), page);
+ null, Files.size(path), page);
+ }
+
+ /** A {@link InputKind#TEXT} unit: the target's extracted text, held in
memory. */
+ public InferenceUnit(MediaType type, Metadata target, Metadata parent,
String text) {
+ this(InputKind.TEXT, type, target, parent,
+ target.get(TikaCoreProperties.EMBEDDED_ID_PATH),
+ parent == null ? null :
parent.get(TikaCoreProperties.EMBEDDED_ID_PATH), null,
+ text, text.getBytes(StandardCharsets.UTF_8).length, -1);
}
private InferenceUnit(InputKind kind, MediaType type, Metadata target,
Metadata parent,
- String targetIdPath, String parentIdPath, Path path,
long size,
- int page) {
+ String targetIdPath, String parentIdPath, Path path,
String text,
+ long size, int page) {
this.kind = kind;
this.type = type;
this.target = target;
@@ -66,6 +76,7 @@ public final class InferenceUnit {
this.targetIdPath = targetIdPath;
this.parentIdPath = parentIdPath;
this.path = path;
+ this.text = text;
this.size = size;
this.page = page;
}
@@ -73,7 +84,12 @@ public final class InferenceUnit {
/** The same unit aimed at the metadata objects that are still read. */
InferenceUnit retargeted(Metadata target, Metadata parent) {
return new InferenceUnit(kind, type, target, parent, targetIdPath,
parentIdPath, path,
- size, page);
+ text, size, page);
+ }
+
+ /** The text of a {@link InputKind#TEXT} unit; null for the rest. */
+ public String getText() {
+ return text;
}
/** The 1-based page this unit renders, for {@link InputKind#PAGES}; -1
otherwise. */
@@ -117,6 +133,6 @@ public final class InferenceUnit {
/** Readable during a task run only; the file is deleted when the flush
ends. */
public byte[] getBytes() throws IOException {
- return Files.readAllBytes(path);
+ return text != null ? text.getBytes(StandardCharsets.UTF_8) :
Files.readAllBytes(path);
}
}
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
b/tika-core/src/main/java/org/apache/tika/parser/inference/TextChunker.java
similarity index 59%
copy from
tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
copy to
tika-core/src/main/java/org/apache/tika/parser/inference/TextChunker.java
index 169b233967..5fcadb98b8 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
+++ b/tika-core/src/main/java/org/apache/tika/parser/inference/TextChunker.java
@@ -14,21 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.inference;
+package org.apache.tika.parser.inference;
-import java.io.IOException;
import java.util.List;
-import org.apache.tika.exception.TikaException;
-import org.apache.tika.parser.ParseContext;
-import org.apache.tika.parser.inference.Engine;
-
-/** An engine that turns images into vectors, many per request. */
-public interface EmbeddingEngine extends Engine {
-
- /** One request; the vectors come back in the order of the images. */
- List<float[]> embedImages(List<byte[]> images, List<String> mimeTypes,
ParseContext context)
- throws IOException, TikaException;
+/**
+ * Cuts a document's text into the pieces a {@link InputKind#TEXT} binding
sends to its engine.
+ * Configured per binding under {@code "chunker"}.
+ *
+ * @since Apache Tika 4.1
+ */
+public interface TextChunker {
- int getMaxBatchSize();
+ /** Half-open {@code [start, end)} character ranges of the chunks, in
document order. */
+ List<int[]> spans(String text);
}
diff --git
a/tika-core/src/test/java/org/apache/tika/parser/inference/InferenceTextStageTest.java
b/tika-core/src/test/java/org/apache/tika/parser/inference/InferenceTextStageTest.java
new file mode 100644
index 0000000000..87e89a2344
--- /dev/null
+++
b/tika-core/src/test/java/org/apache/tika/parser/inference/InferenceTextStageTest.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.parser.inference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.metadata.HttpHeaders;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+import org.apache.tika.mime.MediaType;
+import org.apache.tika.parser.ParseContext;
+
+public class InferenceTextStageTest {
+
+ static final class RecordingTask implements InferenceTask {
+ final List<List<InferenceUnit>> runs = new ArrayList<>();
+
+ @Override
+ public void run(InferenceBinding binding, List<InferenceUnit> units,
Engine engine,
+ ParseContext context) {
+ runs.add(List.copyOf(units));
+ }
+ }
+
+ private static InferenceBinding text(String id, Set<MediaType> include,
long maxBytes) {
+ return new InferenceBinding(id, "engine", InputKind.TEXT,
List.of("t"), include, null,
+ -1, maxBytes, true);
+ }
+
+ private static Metadata doc(String type, String content) {
+ Metadata m = new Metadata();
+ m.set(HttpHeaders.CONTENT_TYPE, type);
+ if (content != null) {
+ m.set(TikaCoreProperties.TIKA_CONTENT, content);
+ }
+ return m;
+ }
+
+ @Test
+ public void testEveryDocumentWithTextIsOneUnitOfOneRun() throws Exception {
+ RecordingTask task = new RecordingTask();
+ InferenceDispatcher dispatcher = new InferenceDispatcher(List.of(
+ new InferenceDispatcher.Bound(text("t", null, -1), new
Engine() { },
+ List.of(task))));
+ Metadata root = doc("application/zip", "root text");
+ Metadata child = doc("text/plain", "child text");
+ Metadata blank = doc("image/png", " ");
+ Metadata pages = doc("application/pdf", "pdf text");
+ pages.add(TikaCoreProperties.INFERENCE_RELEASED, "PAGES");
+ List<Metadata> list = List.of(root, child, blank, pages);
+
+ dispatcher.text(list, new ParseContext());
+
+ assertEquals(1, task.runs.size(), "the whole list in one run");
+ List<InferenceUnit> units = task.runs.get(0);
+ assertEquals(2, units.size(), "blank content and a document released
as pages are skipped");
+ assertSame(root, units.get(0).getTarget());
+ assertEquals("root text", units.get(0).getText());
+ assertEquals(InputKind.TEXT, units.get(0).getKind());
+ assertSame(child, units.get(1).getTarget());
+ assertEquals(MediaType.TEXT_PLAIN, units.get(1).getType());
+ }
+
+ @Test
+ public void testFiltersBudgetAndSelection() throws Exception {
+ RecordingTask plain = new RecordingTask();
+ RecordingTask small = new RecordingTask();
+ InferenceDispatcher dispatcher = new InferenceDispatcher(List.of(
+ new InferenceDispatcher.Bound(text("plain",
Set.of(MediaType.TEXT_PLAIN), -1),
+ new Engine() { }, List.of(plain)),
+ new InferenceDispatcher.Bound(text("small", null, 5), new
Engine() { },
+ List.of(small))));
+ Metadata root = doc("application/zip", "root text");
+ Metadata child = doc("text/plain", "abc");
+ List<Metadata> list = List.of(root, child);
+
+ dispatcher.text(list, new ParseContext());
+ assertEquals(1, plain.runs.get(0).size(), "_mime-include narrows by
document type");
+ assertSame(child, plain.runs.get(0).get(0).getTarget());
+ assertEquals(1, small.runs.get(0).size(), "maxBytes drops the 9-byte
root");
+ assertTrue(root.get(TikaCoreProperties.TIKA_META_EXCEPTION_WARNING)
+ .contains("small over maxBytes: skipped 1 units"));
+
+ ParseContext off = new ParseContext();
+ InferenceSelection selection = new InferenceSelection();
+ selection.setEnabled(false);
+ off.set(InferenceSelection.class, selection);
+ dispatcher.text(List.of(doc("text/plain", "x")), off);
+ assertEquals(1, plain.runs.size(), "switched off for the request: no
run");
+ }
+
+ @Test
+ public void testTaskFailureIsAWarningOnTheRoot() throws Exception {
+ InferenceTask broken = (binding, units, engine, context) -> {
+ throw new IllegalStateException("boom");
+ };
+ InferenceDispatcher dispatcher = new InferenceDispatcher(List.of(
+ new InferenceDispatcher.Bound(text("t", null, -1), new
Engine() { },
+ List.of(broken))));
+ Metadata root = doc("text/plain", "hello");
+ dispatcher.text(List.of(root), new ParseContext());
+ assertTrue(root.get(TikaCoreProperties.TIKA_META_EXCEPTION_WARNING)
+ .contains("inference binding t: boom"));
+ }
+}
diff --git
a/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-key-fields.json
b/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-key-fields.json
index 99646bb828..6cdb062614 100644
---
a/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-key-fields.json
+++
b/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-key-fields.json
@@ -529,6 +529,7 @@
{"class":"org.apache.tika.metadata.TikaCoreProperties","field":"FORMAT","key":"dc:format"},
{"class":"org.apache.tika.metadata.TikaCoreProperties","field":"HAS_SIGNATURE","key":"tk:has-signature"},
{"class":"org.apache.tika.metadata.TikaCoreProperties","field":"IDENTIFIER","key":"dc:identifier"},
+
{"class":"org.apache.tika.metadata.TikaCoreProperties","field":"INFERENCE_RELEASED","key":"tk:inference-released"},
{"class":"org.apache.tika.metadata.TikaCoreProperties","field":"INTERNAL_PATH","key":"tk:internal-path"},
{"class":"org.apache.tika.metadata.TikaCoreProperties","field":"IS_ENCRYPTED","key":"tk:encrypted"},
{"class":"org.apache.tika.metadata.TikaCoreProperties","field":"LANGUAGE","key":"dc:language"},
diff --git
a/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-keys.json
b/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-keys.json
index f75945ed23..accd99efd8 100644
---
a/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-keys.json
+++
b/tika-metadata-schema/src/main/resources/org/apache/tika/metadata/metadata-keys.json
@@ -603,6 +603,7 @@
{"key":"tk:exception:write-limit-reached","namespace":"tk","valueType":"BOOLEAN","cardinality":"SIMPLE","module":"tika-core"},
{"key":"tk:final-embedded-resource-path","namespace":"tk","valueType":"TEXT","cardinality":"SIMPLE","module":"tika-core"},
{"key":"tk:has-signature","namespace":"tk","valueType":"BOOLEAN","cardinality":"SIMPLE","module":"tika-core"},
+
{"key":"tk:inference-released","namespace":"tk","valueType":"TEXT","cardinality":"BAG","module":"tika-core"},
{"key":"tk:internal-path","namespace":"tk","valueType":"TEXT","cardinality":"SIMPLE","module":"tika-core"},
{"key":"tk:num-images","namespace":"tk","valueType":"INTEGER","cardinality":"SIMPLE","module":"tika-core"},
{"key":"tk:orig-resource-name","namespace":"tk","valueType":"TEXT","cardinality":"BAG","module":"tika-core"},
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/Chunk.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/Chunk.java
index a51e95daf9..21a246ee20 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/Chunk.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/Chunk.java
@@ -34,6 +34,7 @@ public class Chunk {
private final String text;
private final Locators locators;
private float[] vector;
+ private String producer;
public Chunk(String text, Locators locators) {
this.text = text;
@@ -78,6 +79,15 @@ public class Chunk {
return -1;
}
+ /** The inference binding that wrote this chunk; null for chunks from
other sources. */
+ public String getProducer() {
+ return producer;
+ }
+
+ public void setProducer(String producer) {
+ this.producer = producer;
+ }
+
public float[] getVector() {
return vector;
}
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/ChunkSerializer.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/ChunkSerializer.java
index bcef910dd2..67fdc53aac 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/ChunkSerializer.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/ChunkSerializer.java
@@ -60,6 +60,9 @@ public final class ChunkSerializer {
if (chunk.getVector() != null) {
node.put("vector", VectorSerializer.encode(chunk.getVector()));
}
+ if (chunk.getProducer() != null) {
+ node.put("producer", chunk.getProducer());
+ }
serializeLocators(node, chunk.getLocators());
}
return MAPPER.writeValueAsString(array);
@@ -107,6 +110,9 @@ public final class ChunkSerializer {
Locators locators = deserializeLocators(node.get("locators"));
Chunk chunk = new Chunk(text, locators);
+ if (node.hasNonNull("producer")) {
+ chunk.setProducer(node.get("producer").asText());
+ }
JsonNode vectorNode = node.get("vector");
if (vectorNode != null && !vectorNode.isNull()) {
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbedTask.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbedTask.java
index 9fab518a83..58f96a5c86 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbedTask.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbedTask.java
@@ -18,13 +18,16 @@ package org.apache.tika.inference;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.IdentityHashMap;
import java.util.List;
+import java.util.Map;
import org.apache.tika.annotation.TikaComponent;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.inference.locator.Locators;
import org.apache.tika.inference.locator.PaginatedLocator;
+import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.metadata.TikaPagedText;
import org.apache.tika.parser.ParseContext;
@@ -35,9 +38,11 @@ import org.apache.tika.parser.inference.InferenceUnit;
import org.apache.tika.parser.inference.InputKind;
/**
- * The {@code embed} task: one vector chunk per unit, written where the unit's
chunks belong
- * (its parent for an inline picture or a page render, itself otherwise), with
the page
- * locator when the unit is a page or has one. Units go to the engine in
batches of its size.
+ * The {@code embed} task. Images and pages: one vector chunk per unit,
written where the
+ * unit's chunks belong (its parent for an inline picture or a page render,
itself otherwise),
+ * with the page locator when the unit is a page or has one. Text: the unit's
text is chunked
+ * and every chunk gets a vector, on the unit's own document. Requests are
filled to the
+ * engine's batch size; every chunk names the binding as its producer.
*/
@TikaComponent(name = "embed", spi = false)
public class EmbedTask implements InferenceTask {
@@ -48,8 +53,8 @@ public class EmbedTask implements InferenceTask {
throw new TikaConfigException("task \"embed\" needs an embedding
engine; \""
+ binding.getEngine() + "\" is not one");
}
- if (binding.getInput() != InputKind.IMAGES && binding.getInput() !=
InputKind.PAGES) {
- throw new TikaConfigException("task \"embed\" takes IMAGES or
PAGES; binding \""
+ if (binding.getInput() == InputKind.MEDIA) {
+ throw new TikaConfigException("task \"embed\" takes TEXT, IMAGES
or PAGES; binding \""
+ binding.getId() + "\" is on " + binding.getInput());
}
}
@@ -58,12 +63,16 @@ public class EmbedTask implements InferenceTask {
public void run(InferenceBinding binding, List<InferenceUnit> units,
Engine engine,
ParseContext context) throws IOException, TikaException {
EmbeddingEngine embedder = (EmbeddingEngine) engine;
+ if (binding.getInput() == InputKind.TEXT) {
+ embedText(binding, units, embedder, context);
+ return;
+ }
int batchSize = Math.max(1, embedder.getMaxBatchSize());
TikaException first = null;
for (int start = 0; start < units.size(); start += batchSize) {
List<InferenceUnit> batch = units.subList(start, Math.min(start +
batchSize, units.size()));
try {
- embed(batch, embedder, context);
+ embed(binding, batch, embedder, context);
} catch (TikaException e) {
if (batch.size() == 1) {
first = first == null ? e : first;
@@ -72,7 +81,7 @@ public class EmbedTask implements InferenceTask {
// one bad image fails a whole batch; retry singly so the rest
keep their vectors
for (InferenceUnit unit : batch) {
try {
- embed(List.of(unit), embedder, context);
+ embed(binding, List.of(unit), embedder, context);
} catch (TikaException single) {
first = first == null ? single : first;
}
@@ -84,8 +93,100 @@ public class EmbedTask implements InferenceTask {
}
}
- private static void embed(List<InferenceUnit> batch, EmbeddingEngine
embedder,
- ParseContext context) throws IOException,
TikaException {
+ /** A chunk of text and the document it belongs to. */
+ private record Owned(Chunk chunk, Metadata owner) {
+ }
+
+ /**
+ * Chunks every unit, then fills each request with the next {@code
maxBatchSize} chunks in
+ * order, so one request may hold a container's last chunks and an
attachment's first.
+ * {@code maxChunks} caps the chunks of the whole list. A rejected request
is retried one
+ * document at a time.
+ */
+ private static void embedText(InferenceBinding binding,
List<InferenceUnit> units,
+ EmbeddingEngine embedder, ParseContext
context)
+ throws IOException, TikaException {
+ List<Owned> owned = new ArrayList<>();
+ for (InferenceUnit unit : units) {
+ String text = unit.getText();
+ List<int[]> spans = binding.getChunker() == null
+ ? List.of(new int[]{0, text.length()}) :
binding.getChunker().spans(text);
+ for (int[] span : spans) {
+ Chunk chunk = new Chunk(text.substring(span[0], span[1]),
span[0], span[1]);
+ chunk.setProducer(binding.getId());
+ owned.add(new Owned(chunk, unit.getTarget()));
+ }
+ }
+ if (binding.getMaxChunks() >= 0 && owned.size() >
binding.getMaxChunks()) {
+ owned = owned.subList(0, binding.getMaxChunks());
+ }
+ int batchSize = Math.max(1, embedder.getMaxBatchSize());
+ TikaException first = null;
+ List<Owned> embedded = new ArrayList<>();
+ for (int start = 0; start < owned.size(); start += batchSize) {
+ List<Owned> batch = owned.subList(start, Math.min(start +
batchSize, owned.size()));
+ try {
+ vectors(batch, embedder, context);
+ embedded.addAll(batch);
+ } catch (TikaException e) {
+ if (batch.size() == 1) {
+ first = first == null ? e : first;
+ continue;
+ }
+ for (List<Owned> byOwner : byOwner(batch)) {
+ try {
+ vectors(byOwner, embedder, context);
+ embedded.addAll(byOwner);
+ } catch (TikaException single) {
+ first = first == null ? single : first;
+ }
+ }
+ }
+ }
+ for (List<Owned> byOwner : byOwner(embedded)) {
+ List<Chunk> chunks = new ArrayList<>(byOwner.size());
+ for (Owned o : byOwner) {
+ chunks.add(o.chunk());
+ }
+ ChunkTarget.self(byOwner.get(0).owner())
+ .write(chunks, TikaCoreProperties.TIKA_CHUNKS.getName());
+ }
+ if (first != null) {
+ throw first;
+ }
+ }
+
+ private static void vectors(List<Owned> batch, EmbeddingEngine embedder,
+ ParseContext context) throws IOException,
TikaException {
+ List<String> texts = new ArrayList<>(batch.size());
+ for (Owned o : batch) {
+ texts.add(o.chunk().getText());
+ }
+ List<float[]> vectors = embedder.embedTexts(texts, context);
+ for (int i = 0; i < batch.size(); i++) {
+ batch.get(i).chunk().setVector(vectors.get(i));
+ }
+ }
+
+ /** Runs of chunks sharing a document, in first-seen order. */
+ private static List<List<Owned>> byOwner(List<Owned> owned) {
+ List<List<Owned>> groups = new ArrayList<>();
+ Map<Metadata, List<Owned>> byOwner = new IdentityHashMap<>();
+ for (Owned o : owned) {
+ List<Owned> group = byOwner.get(o.owner());
+ if (group == null) {
+ group = new ArrayList<>();
+ byOwner.put(o.owner(), group);
+ groups.add(group);
+ }
+ group.add(o);
+ }
+ return groups;
+ }
+
+ private static void embed(InferenceBinding binding, List<InferenceUnit>
batch,
+ EmbeddingEngine embedder, ParseContext context)
+ throws IOException, TikaException {
List<byte[]> images = new ArrayList<>(batch.size());
List<String> mimeTypes = new ArrayList<>(batch.size());
for (InferenceUnit unit : batch) {
@@ -104,6 +205,7 @@ public class EmbedTask implements InferenceTask {
}
Chunk chunk = new Chunk(null, locators);
chunk.setVector(vectors.get(i));
+ chunk.setProducer(binding.getId());
ChunkTarget.resolve(unit.getTarget(), unit.getParent())
.write(List.of(chunk),
TikaCoreProperties.TIKA_CHUNKS.getName());
}
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
index 169b233967..86a38b1580 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
@@ -30,5 +30,9 @@ public interface EmbeddingEngine extends Engine {
List<float[]> embedImages(List<byte[]> images, List<String> mimeTypes,
ParseContext context)
throws IOException, TikaException;
+ /** One request; the vectors come back in the order of the texts. */
+ List<float[]> embedTexts(List<String> texts, ParseContext context)
+ throws IOException, TikaException;
+
int getMaxBatchSize();
}
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MarkdownChunker.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MarkdownChunker.java
index 2bbbfe184a..5e8262cf6c 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MarkdownChunker.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MarkdownChunker.java
@@ -21,6 +21,11 @@ import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import org.apache.tika.annotation.TikaComponent;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.parser.inference.TextChunker;
+
/**
* Splits markdown text into chunks that respect structural boundaries.
* <p>
@@ -32,7 +37,8 @@ import java.util.regex.Pattern;
* Consecutive chunks can overlap by a configurable number of characters
* to avoid losing context at boundaries.
*/
-public class MarkdownChunker {
+@TikaComponent(name = "markdown-chunker", spi = false)
+public class MarkdownChunker implements TextChunker, Initializable {
/**
* Matches a markdown heading at the start of a line (e.g. {@code ## Foo}).
@@ -40,10 +46,29 @@ public class MarkdownChunker {
*/
private static final Pattern HEADING_PATTERN =
Pattern.compile("(?m)^#{1,6} ");
- private final int maxChunkChars;
- private final int overlapChars;
+ private int maxChunkChars = 1500;
+ private int overlapChars = 200;
+
+ /** The binding form: {@code {"markdown-chunker": {"maxChunkChars": ..,
"overlapChars": ..}}}. */
+ public MarkdownChunker() {
+ }
public MarkdownChunker(int maxChunkChars, int overlapChars) {
+ this.maxChunkChars = maxChunkChars;
+ this.overlapChars = overlapChars;
+ validate();
+ }
+
+ @Override
+ public void initialize() throws TikaConfigException {
+ try {
+ validate();
+ } catch (IllegalArgumentException e) {
+ throw new TikaConfigException("markdown-chunker: " +
e.getMessage(), e);
+ }
+ }
+
+ private void validate() {
if (maxChunkChars <= 0) {
throw new IllegalArgumentException("maxChunkChars must be > 0");
}
@@ -54,10 +79,33 @@ public class MarkdownChunker {
throw new IllegalArgumentException(
"overlapChars must be < maxChunkChars");
}
+ }
+
+ public int getMaxChunkChars() {
+ return maxChunkChars;
+ }
+
+ public void setMaxChunkChars(int maxChunkChars) {
this.maxChunkChars = maxChunkChars;
+ }
+
+ public int getOverlapChars() {
+ return overlapChars;
+ }
+
+ public void setOverlapChars(int overlapChars) {
this.overlapChars = overlapChars;
}
+ @Override
+ public List<int[]> spans(String text) {
+ List<int[]> spans = new ArrayList<>();
+ for (Chunk chunk : chunk(text)) {
+ spans.add(new int[]{chunk.getStartOffset(), chunk.getEndOffset()});
+ }
+ return spans;
+ }
+
/**
* Chunk the given markdown text.
*
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
index 90e85f5d2d..517844eac4 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/OpenAIEmbeddingEngine.java
@@ -73,17 +73,38 @@ public class OpenAIEmbeddingEngine implements
EmbeddingEngine, Initializable, Cl
@Override
public List<float[]> embedImages(List<byte[]> images, List<String>
mimeTypes,
ParseContext context) throws IOException,
TikaException {
- if (httpClient == null) {
- httpClient = TikaHttpClient.build(30);
+ ObjectNode root = request();
+ ArrayNode input = root.putArray("input");
+ for (int i = 0; i < images.size(); i++) {
+ input.addObject().put("image", "data:" + mimeTypes.get(i) +
";base64,"
+ + Base64.getEncoder().encodeToString(images.get(i)));
+ }
+ return post(root, images.size(), context);
+ }
+
+ @Override
+ public List<float[]> embedTexts(List<String> texts, ParseContext context)
+ throws IOException, TikaException {
+ ObjectNode root = request();
+ ArrayNode input = root.putArray("input");
+ for (String text : texts) {
+ input.add(text);
}
+ return post(root, texts.size(), context);
+ }
+
+ private ObjectNode request() {
ObjectNode root = MAPPER.createObjectNode();
if (!StringUtils.isBlank(model)) {
root.put("model", model);
}
- ArrayNode input = root.putArray("input");
- for (int i = 0; i < images.size(); i++) {
- input.addObject().put("image", "data:" + mimeTypes.get(i) +
";base64,"
- + Base64.getEncoder().encodeToString(images.get(i)));
+ return root;
+ }
+
+ private List<float[]> post(ObjectNode root, int expected, ParseContext
context)
+ throws IOException, TikaException {
+ if (httpClient == null) {
+ httpClient = TikaHttpClient.build(30);
}
Map<String, String> headers = new HashMap<>();
if (!StringUtils.isBlank(apiKey)) {
@@ -92,7 +113,7 @@ public class OpenAIEmbeddingEngine implements
EmbeddingEngine, Initializable, Cl
String url = baseUrl.replaceAll("/+$", "") + embeddingsPath;
String body = httpClient.postJson(url, root.toString(), headers,
timeoutMillis, context);
ParseTimeout.checkpoint(context);
- return parseResponse(body, images.size());
+ return parseResponse(body, expected);
}
static List<float[]> parseResponse(String body, int expected) throws
TikaException {
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/ChunkSerializerTest.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/ChunkSerializerTest.java
index a0357b00d3..07381f6949 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/ChunkSerializerTest.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/ChunkSerializerTest.java
@@ -152,6 +152,16 @@ public class ChunkSerializerTest {
assertNull(restored.get(0).getLocators().getPaginated().get(0).getBbox());
}
+ @Test
+ void testProducerRoundTrips() throws Exception {
+ Chunk chunk = new Chunk("t", 0, 1);
+ chunk.setProducer("text-vectors");
+ List<Chunk> back =
ChunkSerializer.fromJson(ChunkSerializer.toJson(List.of(chunk)));
+ assertEquals("text-vectors", back.get(0).getProducer());
+ assertNull(ChunkSerializer.fromJson(ChunkSerializer.toJson(
+ List.of(new Chunk("t", 0, 1)))).get(0).getProducer());
+ }
+
@Test
void testWithoutVector() throws Exception {
Chunk c = new Chunk("No vector", 0, 9);
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskTest.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskTest.java
index 0980bd4edd..7759a4d385 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskTest.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskTest.java
@@ -161,6 +161,62 @@ public class EmbedTaskTest {
assertNull(chunks.get(0).getLocators().getEmbedded(), "the pdf keeps
its own pages");
}
+ /** Text units are chunked, batched across documents, and land on their
own document. */
+ @Test
+ public void testTextChunksBatchAcrossDocuments() throws Exception {
+ engine.setMaxBatchSize(3);
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(0, 1,
2)));
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(0)));
+ Metadata parent = new Metadata();
+ Metadata attachment = new Metadata();
+ attachment.set(TikaCoreProperties.EMBEDDED_RESOURCE_TYPE,
"ATTACHMENT");
+ attachment.set(TikaCoreProperties.EMBEDDED_ID_PATH, "/1");
+ InferenceBinding text = new InferenceBinding("text-vectors", "clip",
InputKind.TEXT,
+ List.of("embed"), null, null, -1, -1, true, new
MarkdownChunker(10, 0));
+ EmbedTask task = new EmbedTask();
+ task.validate(text, engine);
+ // "one two" and "three" from the parent, "four five" and "six" from
the attachment
+ task.run(text, List.of(
+ new InferenceUnit(MediaType.TEXT_PLAIN, parent, null, "one
two\n\nthree"),
+ new InferenceUnit(MediaType.TEXT_PLAIN, attachment, parent,
"four five\n\nsix")),
+ engine, new ParseContext());
+
+ assertEquals(2, server.getRequestCount(), "four chunks in batches of
three");
+ JsonNode first =
MAPPER.readTree(server.takeRequest().body()).get("input");
+ assertEquals(List.of("one two", "three", "four five"),
+ List.of(first.get(0).asText(), first.get(1).asText(),
first.get(2).asText()),
+ "a request holds the parent's last chunk and the attachment's
first");
+ List<Chunk> parentChunks = ChunkSerializer.fromJson(
+ parent.get(TikaCoreProperties.TIKA_CHUNKS));
+ List<Chunk> attachmentChunks = ChunkSerializer.fromJson(
+ attachment.get(TikaCoreProperties.TIKA_CHUNKS));
+ assertEquals(2, parentChunks.size());
+ assertEquals(2, attachmentChunks.size(), "an attachment keeps its own
text vectors");
+ assertEquals("three", parentChunks.get(1).getText());
+ assertEquals(9,
parentChunks.get(1).getLocators().getText().get(0).getStartOffset());
+ assertEquals("text-vectors", parentChunks.get(0).getProducer());
+ assertEquals(2.0f, attachmentChunks.get(0).getVector()[0], "placed by
index");
+ assertEquals(0.0f, attachmentChunks.get(1).getVector()[0], "second
request, index 0");
+ }
+
+ /** Without a chunker the whole text is one chunk; maxChunks caps the
list. */
+ @Test
+ public void testTextWithoutChunkerAndBudget() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(0)));
+ Metadata a = new Metadata();
+ Metadata b = new Metadata();
+ InferenceBinding text = new InferenceBinding("t", "clip",
InputKind.TEXT,
+ List.of("embed"), null, null, 1, -1, true);
+ new EmbedTask().run(text, List.of(
+ new InferenceUnit(MediaType.TEXT_PLAIN, a, null, "whole
document"),
+ new InferenceUnit(MediaType.TEXT_PLAIN, b, null, "dropped")),
+ engine, new ParseContext());
+ assertEquals(1, server.getRequestCount());
+ assertEquals("whole document", ChunkSerializer.fromJson(
+ a.get(TikaCoreProperties.TIKA_CHUNKS)).get(0).getText());
+ assertNull(b.get(TikaCoreProperties.TIKA_CHUNKS), "over maxChunks");
+ }
+
@Test
public void testValidateRejectsANonEmbeddingEngine() {
assertThrows(TikaConfigException.class,
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/MarkdownChunkerTest.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/MarkdownChunkerTest.java
index db3a874834..4a1d7b4e9f 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/MarkdownChunkerTest.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/MarkdownChunkerTest.java
@@ -24,6 +24,8 @@ import java.util.List;
import org.junit.jupiter.api.Test;
+import org.apache.tika.exception.TikaConfigException;
+
public class MarkdownChunkerTest {
@Test
@@ -37,6 +39,25 @@ public class MarkdownChunkerTest {
assertTrue(chunks.get(1).getText().startsWith("# Heading 2"));
}
+ @Test
+ void testSpansMatchChunkOffsets() throws Exception {
+ MarkdownChunker chunker = new MarkdownChunker();
+ chunker.setMaxChunkChars(10);
+ chunker.setOverlapChars(0);
+ chunker.initialize();
+ String text = "one two\n\nthree four\n\nfive";
+ List<Chunk> chunks = chunker.chunk(text);
+ List<int[]> spans = chunker.spans(text);
+ assertEquals(chunks.size(), spans.size());
+ for (int i = 0; i < chunks.size(); i++) {
+ assertEquals(chunks.get(i).getStartOffset(), spans.get(i)[0]);
+ assertEquals(chunks.get(i).getEndOffset(), spans.get(i)[1]);
+ assertEquals(chunks.get(i).getText(),
text.substring(spans.get(i)[0], spans.get(i)[1]));
+ }
+ chunker.setOverlapChars(10);
+ assertThrows(TikaConfigException.class, chunker::initialize);
+ }
+
@Test
void testOffsets() {
String md = "# A\n\nText A.\n\n# B\n\nText B.";
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/OpenAIEmbeddingEngineTest.java
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/OpenAIEmbeddingEngineTest.java
index 04ed8d387c..fdb658c5f0 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/OpenAIEmbeddingEngineTest.java
+++
b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/OpenAIEmbeddingEngineTest.java
@@ -79,6 +79,19 @@ public class OpenAIEmbeddingEngineTest {
assertEquals("data:image/jpeg;base64,Yg==",
body.get("input").get(1).get("image").asText());
}
+ @Test
+ public void testTextsGoAsPlainStrings() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200,
+
"{\"data\":[{\"index\":1,\"embedding\":[1.0]},{\"index\":0,\"embedding\":[0.0]}]}"));
+ List<float[]> vectors = engine.embedTexts(List.of("alpha", "beta"),
new ParseContext());
+ JsonNode request = new
ObjectMapper().readTree(server.takeRequest().body());
+ assertEquals("clip", request.get("model").asText());
+ assertEquals("alpha", request.get("input").get(0).asText());
+ assertEquals("beta", request.get("input").get(1).asText());
+ assertEquals(0.0f, vectors.get(0)[0]);
+ assertEquals(1.0f, vectors.get(1)[0]);
+ }
+
@Test
public void testMalformedResponsesAreRefused() throws Exception {
String[] bad = {
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/parser/pdf/AbstractPDF2XHTML.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/parser/pdf/AbstractPDF2XHTML.java
index 725f63f508..155df57869 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/parser/pdf/AbstractPDF2XHTML.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/main/java/org/apache/tika/parser/pdf/AbstractPDF2XHTML.java
@@ -281,6 +281,12 @@ class AbstractPDF2XHTML extends PDFTextStripper {
}
}
this.pagesForInference = wantsPages;
+ if (wantsPages) {
+ // the TEXT stage skips a document released as pages unless TEXT
is listed too
+ for (InferenceConfig.Input input :
config.getInference().getInput()) {
+ metadata.add(TikaCoreProperties.INFERENCE_RELEASED,
input.name());
+ }
+ }
}
private static void addNonNullAttribute(String name, String value,
AttributesImpl attributes) {
diff --git
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/test/java/org/apache/tika/parser/pdf/PDFParserTest.java
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/test/java/org/apache/tika/parser/pdf/PDFParserTest.java
index b48638488e..ca61564f9a 100644
---
a/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/test/java/org/apache/tika/parser/pdf/PDFParserTest.java
+++
b/tika-parsers/tika-parsers-standard/tika-parsers-standard-modules/tika-parser-pdf-module/src/test/java/org/apache/tika/parser/pdf/PDFParserTest.java
@@ -1978,6 +1978,9 @@ public class PDFParserTest extends TikaTest {
assertTrue(hook.sizes.get(0) > 0, label + ": the render has
bytes");
assertEquals(2, tracker.getNextId() - 1, label + ": one render per
page");
assertEquals(strategy == OcrConfig.Strategy.NO_OCR ? 0 : 2,
recognizer.calls, label);
+ assertArrayEquals(new String[]{"PAGES"},
+ metadata.getValues(TikaCoreProperties.INFERENCE_RELEASED),
+ label + ": the PDF says it released its pages, not its
text");
}
}
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
index 8b16047e33..b09c3f8f9f 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java
@@ -315,7 +315,8 @@ public class ConnectionHandler implements Runnable,
Closeable {
Long thresholdBytes =
pipesConfig.getEmitStrategy().getThresholdBytes();
long threshold = (thresholdBytes != null) ? thresholdBytes :
EmitStrategyConfig.DEFAULT_DIRECT_EMIT_THRESHOLD_BYTES;
EmitHandler emitHandler = new
EmitHandler(resources.getDefaultMetadataFilter(),
- resources.getEmitStrategy(), resources.getEmitterManager(),
threshold);
+ resources.getInferenceDispatcher(),
resources.getEmitStrategy(),
+ resources.getEmitterManager(), threshold);
return new PipesWorker(fetchEmitTuple, mergedContext,
resources.getAutoDetectParser(),
resources.getEmitterManager(), fetchHandler, parseHandler,
emitHandler,
resources.getDefaultMetadataWriteLimiterFactory(),
pipesConfig.getParseMode());
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java
index 7a73feb4a7..114b5a980f 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java
@@ -37,6 +37,7 @@ import org.apache.tika.metadata.filter.MetadataFilter;
import org.apache.tika.metadata.filter.NoOpFilter;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.ParseRecord;
+import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.pipes.api.FetchEmitTuple;
import org.apache.tika.pipes.api.ParseMode;
import org.apache.tika.pipes.api.PipesResult;
@@ -57,13 +58,16 @@ class EmitHandler {
private static final Logger LOG =
LoggerFactory.getLogger(EmitHandler.class);
private final MetadataFilter defaultMetadataFilter;
+ private final InferenceDispatcher inferenceDispatcher;
private final EmitStrategy emitStrategy;
private final EmitterManager emitterManager;
private final long directEmitThresholdBytes;
- public EmitHandler(MetadataFilter defaultMetadataFilter, EmitStrategy
emitStrategy, EmitterManager emitterManager, long directEmitThresholdBytes) {
+ public EmitHandler(MetadataFilter defaultMetadataFilter,
InferenceDispatcher inferenceDispatcher,
+ EmitStrategy emitStrategy, EmitterManager
emitterManager, long directEmitThresholdBytes) {
this.defaultMetadataFilter = defaultMetadataFilter;
+ this.inferenceDispatcher = inferenceDispatcher;
this.emitStrategy = emitStrategy;
this.emitterManager = emitterManager;
this.directEmitThresholdBytes = directEmitThresholdBytes;
@@ -111,11 +115,15 @@ class EmitHandler {
private PipesResult emitParseDataInternal(FetchEmitTuple t,
MetadataListAndEmbeddedBytes parseData, ParseContext parseContext) {
long start = System.currentTimeMillis();
String stack = getContainerStacktrace(t, parseData.getMetadataList());
+ FetchEmitTuple.ON_PARSE_EXCEPTION onParseException =
t.getOnParseException();
+ boolean emit = StringUtils.isBlank(stack) ||
+ onParseException == FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT;
+ if (emit) {
+ runTextInference(parseData, parseContext);
+ }
//we need to apply the metadata filter after we pull out the stacktrace
filterMetadata(parseData, parseContext);
- FetchEmitTuple.ON_PARSE_EXCEPTION onParseException =
t.getOnParseException();
- if (StringUtils.isBlank(stack) ||
- onParseException == FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT) {
+ if (emit) {
injectUserMetadata(t.getMetadata(), parseData.getMetadataList());
EmitKey emitKey = t.getEmitKey();
if (StringUtils.isBlank(emitKey.getEmitKey())) {
@@ -319,6 +327,19 @@ class EmitHandler {
}
}
+ /** The TEXT inference stage: before the filters so they shape its output;
not for CONTENT_ONLY, whose filter would drop it. */
+ private void runTextInference(MetadataListAndEmbeddedBytes parseData,
ParseContext parseContext) {
+ if (inferenceDispatcher == null
+ || parseContext.get(ParseMode.class) ==
ParseMode.CONTENT_ONLY) {
+ return;
+ }
+ try {
+ inferenceDispatcher.text(parseData.getMetadataList(),
parseContext);
+ } catch (TikaException e) {
+ LOG.warn("text inference failed", e);
+ }
+ }
+
private void filterMetadata(MetadataListAndEmbeddedBytes parseData,
ParseContext parseContext) {
MetadataFilter filter = parseContext.get(MetadataFilter.class);
if (filter == null) {
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
index df8fbb5816..a900caad76 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java
@@ -62,6 +62,7 @@ import
org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.RecursiveParserWrapper;
+import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.pipes.api.FetchEmitTuple;
import org.apache.tika.pipes.api.PipesResult;
import org.apache.tika.pipes.core.EmitStrategy;
@@ -185,6 +186,7 @@ public class PipesServer implements AutoCloseable {
private final PipesConfig pipesConfig;
private final Socket socket;
private final MetadataFilter defaultMetadataFilter;
+ private final InferenceDispatcher inferenceDispatcher;
private final ContentHandlerFactory defaultContentHandlerFactory;
private final MetadataWriteLimiterFactory
defaultMetadataWriteLimiterFactory;
private AutoDetectParser autoDetectParser;
@@ -257,6 +259,7 @@ public class PipesServer implements AutoCloseable {
this.socket = socket;
socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMillis());
this.defaultMetadataFilter = tikaLoader.loadMetadataFilters();
+ this.inferenceDispatcher = tikaLoader.get(InferenceDispatcher.class);
this.defaultContentHandlerFactory =
tikaLoader.loadContentHandlerFactory();
this.defaultMetadataWriteLimiterFactory =
configContext.get(MetadataWriteLimiterFactory.class);
this.input = in;
@@ -538,7 +541,8 @@ public class PipesServer implements AutoCloseable {
rMetaParser, defaultContentHandlerFactory,
pipesConfig.getParseMode());
Long thresholdBytes =
pipesConfig.getEmitStrategy().getThresholdBytes();
long threshold = (thresholdBytes != null) ? thresholdBytes :
EmitStrategyConfig.DEFAULT_DIRECT_EMIT_THRESHOLD_BYTES;
- EmitHandler emitHandler = new EmitHandler(defaultMetadataFilter,
emitStrategy, emitterManager, threshold);
+ EmitHandler emitHandler = new EmitHandler(defaultMetadataFilter,
inferenceDispatcher,
+ emitStrategy, emitterManager, threshold);
return new PipesWorker(fetchEmitTuple, mergedContext,
autoDetectParser, emitterManager,
fetchHandler, parseHandler, emitHandler,
defaultMetadataWriteLimiterFactory,
pipesConfig.getParseMode());
diff --git
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
index 5c97818673..80aa8e4c18 100644
---
a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
+++
b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java
@@ -32,6 +32,7 @@ import
org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.RecursiveParserWrapper;
+import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.pipes.core.EmitStrategy;
import org.apache.tika.pipes.core.PipesConfig;
import org.apache.tika.pipes.core.config.ConfigStore;
@@ -62,6 +63,7 @@ public class SharedServerResources {
private final FetcherManager fetcherManager;
private final EmitterManager emitterManager;
private final MetadataFilter defaultMetadataFilter;
+ private final InferenceDispatcher inferenceDispatcher;
private final ContentHandlerFactory defaultContentHandlerFactory;
private final MetadataWriteLimiterFactory
defaultMetadataWriteLimiterFactory;
private final EmitStrategy emitStrategy;
@@ -73,6 +75,7 @@ public class SharedServerResources {
AutoDetectParser autoDetectParser, Detector
detector,
RecursiveParserWrapper rMetaParser,
FetcherManager fetcherManager,
EmitterManager emitterManager,
MetadataFilter defaultMetadataFilter,
+ InferenceDispatcher inferenceDispatcher,
ContentHandlerFactory
defaultContentHandlerFactory,
MetadataWriteLimiterFactory
defaultMetadataWriteLimiterFactory,
EmitStrategy emitStrategy, ConfigStore
configStore,
@@ -86,6 +89,7 @@ public class SharedServerResources {
this.fetcherManager = fetcherManager;
this.emitterManager = emitterManager;
this.defaultMetadataFilter = defaultMetadataFilter;
+ this.inferenceDispatcher = inferenceDispatcher;
this.defaultContentHandlerFactory = defaultContentHandlerFactory;
this.defaultMetadataWriteLimiterFactory =
defaultMetadataWriteLimiterFactory;
this.emitStrategy = emitStrategy;
@@ -124,6 +128,7 @@ public class SharedServerResources {
// Load filters and factories
MetadataFilter metadataFilter = tikaLoader.loadMetadataFilters();
+ InferenceDispatcher inferenceDispatcher =
tikaLoader.get(InferenceDispatcher.class);
ContentHandlerFactory contentHandlerFactory =
tikaLoader.loadContentHandlerFactory();
ParseContext configContext = tikaLoader.loadParseContext();
MetadataWriteLimiterFactory metadataWriteLimiterFactory =
@@ -136,7 +141,8 @@ public class SharedServerResources {
PresetRegistry.load(tikaJsonConfig,
tikaLoader.getClassLoader());
return new SharedServerResources(tikaLoader, pipesConfig,
autoDetectParser, detector,
- rMetaParser, fetcherManager, emitterManager, metadataFilter,
contentHandlerFactory,
+ rMetaParser, fetcherManager, emitterManager, metadataFilter,
inferenceDispatcher,
+ contentHandlerFactory,
metadataWriteLimiterFactory, emitStrategy, configStore,
ExceptionReporting.get(configContext), presetRegistry);
}
@@ -211,6 +217,11 @@ public class SharedServerResources {
return defaultMetadataFilter;
}
+ /** Null when no {@code "inference"} bindings are configured. */
+ public InferenceDispatcher getInferenceDispatcher() {
+ return inferenceDispatcher;
+ }
+
public ContentHandlerFactory getDefaultContentHandlerFactory() {
return defaultContentHandlerFactory;
}
diff --git
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java
index 6a953d62f2..2021d09b64 100644
---
a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java
+++
b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/EmitHandlerUserMetadataTest.java
@@ -36,7 +36,7 @@ public class EmitHandlerUserMetadataTest {
private static EmitHandler emitHandler() {
// injectUserMetadata touches no instance state; the collaborators are
irrelevant
- return new EmitHandler(null, null, null, 0);
+ return new EmitHandler(null, null, null, null, 0);
}
@Test
diff --git
a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java
index b805aba5fd..575905e088 100644
---
a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java
+++
b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java
@@ -107,6 +107,26 @@ public class PipesClientTest {
Metadata metadata =
pipesResult.emitData().getMetadataList().get(0);
assertEquals("mock-images", metadata.get(MockTask.MARKER_KEY));
assertEquals("1", metadata.get(MockTask.UNITS_KEY));
+
+ // the TEXT stage runs at the emit edge of the fork
+ Files.writeString(inputDir.resolve("note.txt"), "a note with
text");
+ pipesResult = pipesClient.process(
+ new FetchEmitTuple("note.txt", new FetchKey(fetcherName,
"note.txt"),
+ new EmitKey(), new Metadata(), new ParseContext(),
+ FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+ metadata = pipesResult.emitData().getMetadataList().get(0);
+ assertEquals("mock-text", metadata.get(MockTask.MARKER_KEY));
+ assertEquals("1", metadata.get(MockTask.UNITS_KEY));
+
+ // a per-request filter list neither drops the TEXT stage nor runs
ahead of it
+ ParseContext filtered = new ParseContext();
+ filtered.setJsonConfig("metadata-filters",
"[\"mock-upper-case-filter\"]");
+ pipesResult = pipesClient.process(
+ new FetchEmitTuple("note.txt", new FetchKey(fetcherName,
"note.txt"),
+ new EmitKey(), new Metadata(), filtered,
+ FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+ metadata = pipesResult.emitData().getMetadataList().get(0);
+ assertEquals("MOCK-TEXT", metadata.get(MockTask.MARKER_KEY));
}
}
diff --git
a/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-inference-bindings.json
b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-inference-bindings.json
index 90b215fb4f..95fabb581c 100644
---
a/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-inference-bindings.json
+++
b/tika-pipes/tika-pipes-integration-tests/src/test/resources/configs/tika-config-inference-bindings.json
@@ -1,6 +1,9 @@
{
"engines": { "mock": { "mock-engine": {} } },
- "inference": [ { "id": "mock-images", "engine": "mock", "input": "IMAGES",
"tasks": ["mock-task"] } ],
+ "inference": [
+ { "id": "mock-images", "engine": "mock", "input": "IMAGES", "tasks":
["mock-task"] },
+ { "id": "mock-text", "engine": "mock", "input": "TEXT", "tasks":
["mock-task"] }
+ ],
"content-handler-factory": {
"basic-content-handler-factory": {
"type": "TEXT",
diff --git
a/tika-serialization/src/main/java/org/apache/tika/config/loader/InferenceLoader.java
b/tika-serialization/src/main/java/org/apache/tika/config/loader/InferenceLoader.java
index 3a55a76c02..a345313430 100644
---
a/tika-serialization/src/main/java/org/apache/tika/config/loader/InferenceLoader.java
+++
b/tika-serialization/src/main/java/org/apache/tika/config/loader/InferenceLoader.java
@@ -21,6 +21,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
+import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
@@ -35,6 +36,7 @@ import org.apache.tika.parser.inference.InferenceBinding;
import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.parser.inference.InferenceTask;
import org.apache.tika.parser.inference.InputKind;
+import org.apache.tika.parser.inference.TextChunker;
/**
* Loads {@code "inference"}: bindings of a named engine to an input kind and
tasks. Every
@@ -45,7 +47,7 @@ class InferenceLoader implements
ComponentLoader<InferenceDispatcher> {
static final String KEY = "inference";
private static final Set<String> KNOWN = Set.of("id", "engine", "input",
"tasks",
- "maxChunks", "maxBytes", "enabled", "_mime-include",
"_mime-exclude");
+ "maxChunks", "maxBytes", "enabled", "_mime-include",
"_mime-exclude", "chunker");
private static final Pattern LEGAL_ID = Pattern.compile("[A-Za-z0-9._-]+");
/** Largest unit a binding takes unless it says otherwise. */
static final long DEFAULT_MAX_BYTES = 20L * 1024 * 1024;
@@ -128,9 +130,10 @@ class InferenceLoader implements
ComponentLoader<InferenceDispatcher> {
exclude.add(MediaType.parse(t));
}
}
+ TextChunker chunker = chunker(entry, id, input, context);
InferenceBinding binding = new InferenceBinding(id, engineName,
input, taskNames,
include, exclude, maxChunks, maxBytes,
- entry.path("enabled").asBoolean(true));
+ entry.path("enabled").asBoolean(true), chunker);
List<InferenceTask> tasks = new ArrayList<>();
for (String taskName : taskNames) {
InferenceTask task;
@@ -151,6 +154,32 @@ class InferenceLoader implements
ComponentLoader<InferenceDispatcher> {
return new InferenceDispatcher(bound);
}
+ /** {@code "chunker": {"<name>": {...}}} on a TEXT binding; absent means
one chunk per document. */
+ private static TextChunker chunker(JsonNode entry, String id, InputKind
input,
+ LoaderContext context) throws
TikaConfigException {
+ if (!entry.has("chunker")) {
+ return null;
+ }
+ if (input != InputKind.TEXT) {
+ throw new TikaConfigException("binding \"" + id + "\": \"chunker\"
applies to TEXT "
+ + "bindings; " + input + " units are not chunked");
+ }
+ JsonNode node = entry.get("chunker");
+ if (!node.isObject() || node.size() != 1) {
+ throw new TikaConfigException("binding \"" + id + "\": \"chunker\"
must be one "
+ + "{\"<chunker>\": {...}} object, e.g.
{\"markdown-chunker\": "
+ + "{\"maxChunkChars\": 1500}}");
+ }
+ Map.Entry<String, JsonNode> type = node.fields().next();
+ try {
+ return ComponentInstantiator.instantiateComponent(type.getKey(),
type.getValue(),
+ context.getObjectMapper(), context.getClassLoader(),
TextChunker.class);
+ } catch (TikaConfigException e) {
+ throw new TikaConfigException("binding \"" + id + "\" chunker \""
+ type.getKey()
+ + "\": " + e.getMessage(), e);
+ }
+ }
+
private static String required(JsonNode entry, String field) throws
TikaConfigException {
if (!entry.hasNonNull(field) || !entry.get(field).isTextual()) {
throw new TikaConfigException("\"" + KEY + "\" entry needs \"" +
field + "\"");
diff --git
a/tika-serialization/src/test/java/org/apache/tika/config/loader/InferenceLoaderTest.java
b/tika-serialization/src/test/java/org/apache/tika/config/loader/InferenceLoaderTest.java
index 04c0c0c80b..a5a4b5fc5b 100644
---
a/tika-serialization/src/test/java/org/apache/tika/config/loader/InferenceLoaderTest.java
+++
b/tika-serialization/src/test/java/org/apache/tika/config/loader/InferenceLoaderTest.java
@@ -28,11 +28,16 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.filter.CompositeMetadataFilter;
+import org.apache.tika.metadata.filter.MetadataFilter;
+import org.apache.tika.metadata.filter.NoOpFilter;
import org.apache.tika.mime.MediaType;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
@@ -115,6 +120,54 @@ public class InferenceLoaderTest {
new ParseContext()));
}
+ /** A TEXT binding carries its chunker; the metadata-filter chain is not
its vehicle. */
+ @Test
+ public void testTextBindingLoadsWithChunker() throws Exception {
+ TikaLoader loader = load("{" + ENGINES + ", \"inference\": ["
+ + " { \"id\": \"text-vectors\", \"engine\": \"one\",
\"input\": \"TEXT\","
+ + " \"tasks\": [\"test-task\"], \"chunker\": {
\"test-chunker\": { \"size\": 3 } } } ],"
+ + " \"metadata-filters\": [ { \"no-op-filter\": {} } ] }");
+ InferenceBinding text =
loader.get(InferenceDispatcher.class).getBound().get(0).binding();
+ assertEquals(InputKind.TEXT, text.getInput());
+ assertTrue(text.getChunker() instanceof TestChunker);
+ assertEquals(3, ((TestChunker) text.getChunker()).getSize());
+ assertEquals(2, text.getChunker().spans("abcde").size());
+
+ MetadataFilter filters = loader.get(MetadataFilter.class);
+ assertTrue(filters instanceof CompositeMetadataFilter);
+ assertEquals(1, ((CompositeMetadataFilter)
filters).getFilters().size(),
+ "only the configured filter: the TEXT stage is not a filter");
+
+ Path dump = Files.createTempFile(tmp, "dump", ".json");
+ loader.save(dump.toFile());
+ JsonNode dumped = new ObjectMapper().readTree(Files.readString(dump));
+ assertEquals(1, dumped.get("metadata-filters").size());
+ assertEquals("TEXT",
dumped.get("inference").get(0).get("input").asText());
+
+ TikaLoader noFilters = load("{" + ENGINES + ", \"inference\": ["
+ + " { \"engine\": \"one\", \"input\": \"TEXT\", \"tasks\":
[\"test-task\"] } ] }");
+ assertSame(NoOpFilter.NOOP_FILTER, noFilters.get(MetadataFilter.class),
+ "a TEXT binding adds nothing to the chain");
+ }
+
+ @Test
+ public void testChunkerMisconfigurationsFailLoad() {
+ String[] bad = {
+ "{" + ENGINES + ", \"inference\": [ { \"engine\": \"one\",
\"input\": \"IMAGES\","
+ + " \"tasks\": [\"test-task\"], \"chunker\": {
\"test-chunker\": {} } } ] }",
+ "{" + ENGINES + ", \"inference\": [ { \"engine\": \"one\",
\"input\": \"TEXT\","
+ + " \"tasks\": [\"test-task\"], \"chunker\": {
\"no-such-chunker\": {} } } ] }",
+ "{" + ENGINES + ", \"inference\": [ { \"engine\": \"one\",
\"input\": \"TEXT\","
+ + " \"tasks\": [\"test-task\"], \"chunker\":
\"test-chunker\" } ] }",
+ "{" + ENGINES + ", \"inference\": [ { \"engine\": \"one\",
\"input\": \"TEXT\","
+ + " \"tasks\": [\"test-task\"], \"chunker\": {
\"test-engine\": {} } } ] }",
+ };
+ for (String json : bad) {
+ assertThrows(TikaConfigException.class,
+ () -> load(json).get(InferenceDispatcher.class), json);
+ }
+ }
+
@Test
public void testPerRequestSelectionResolves() throws Exception {
TikaLoader loader = load("{" + ENGINES + ", \"inference\": ["
diff --git
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
b/tika-serialization/src/test/java/org/apache/tika/config/loader/TestChunker.java
similarity index 51%
copy from
tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
copy to
tika-serialization/src/test/java/org/apache/tika/config/loader/TestChunker.java
index 169b233967..fbb6abbea9 100644
---
a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/EmbeddingEngine.java
+++
b/tika-serialization/src/test/java/org/apache/tika/config/loader/TestChunker.java
@@ -14,21 +14,34 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.inference;
+package org.apache.tika.config.loader;
-import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
-import org.apache.tika.exception.TikaException;
-import org.apache.tika.parser.ParseContext;
-import org.apache.tika.parser.inference.Engine;
+import org.apache.tika.annotation.TikaComponent;
+import org.apache.tika.parser.inference.TextChunker;
-/** An engine that turns images into vectors, many per request. */
-public interface EmbeddingEngine extends Engine {
+/** Cuts text into fixed-size pieces; a stand-in for a real chunker in loader
tests. */
+@TikaComponent(name = "test-chunker", spi = false)
+public class TestChunker implements TextChunker {
- /** One request; the vectors come back in the order of the images. */
- List<float[]> embedImages(List<byte[]> images, List<String> mimeTypes,
ParseContext context)
- throws IOException, TikaException;
+ private int size = 4;
- int getMaxBatchSize();
+ public int getSize() {
+ return size;
+ }
+
+ public void setSize(int size) {
+ this.size = size;
+ }
+
+ @Override
+ public List<int[]> spans(String text) {
+ List<int[]> spans = new ArrayList<>();
+ for (int start = 0; start < text.length(); start += size) {
+ spans.add(new int[]{start, Math.min(start + size, text.length())});
+ }
+ return spans;
+ }
}
diff --git
a/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/InferenceBindingsWireTest.java
b/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/InferenceBindingsWireTest.java
index ca674fe06f..e6a7bfb224 100644
---
a/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/InferenceBindingsWireTest.java
+++
b/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/InferenceBindingsWireTest.java
@@ -29,6 +29,8 @@ import java.io.Reader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
import javax.imageio.ImageIO;
import jakarta.ws.rs.core.Response;
@@ -47,6 +49,7 @@ import org.apache.pdfbox.pdmodel.font.Standard14Fonts;
import org.junit.jupiter.api.Test;
import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.parser.mock.MockTask;
import org.apache.tika.serialization.JsonMetadataList;
import org.apache.tika.server.core.CXFTestBase;
@@ -130,6 +133,33 @@ public class InferenceBindingsWireTest extends CXFTestBase
{
assertEquals("2", metadata.get(MockTask.UNITS_KEY));
}
+ /** The TEXT stage runs in the worker over the whole list: every document
with text is a unit. */
+ @Test
+ public void testTextBindingSeesTheWholeTree() throws Exception {
+ ByteArrayOutputStream zip = new ByteArrayOutputStream();
+ try (ZipOutputStream out = new ZipOutputStream(zip)) {
+ out.putNextEntry(new ZipEntry("a.txt"));
+ out.write("alpha note".getBytes(UTF_8));
+ out.closeEntry();
+ out.putNextEntry(new ZipEntry("b.txt"));
+ out.write("beta note".getBytes(UTF_8));
+ out.closeEntry();
+ }
+ Response response = WebClient
+ .create(endPoint + META_PATH)
+ .accept("application/json")
+ .put(zip.toByteArray());
+
+ Reader reader = new InputStreamReader((InputStream)
response.getEntity(), UTF_8);
+ List<Metadata> metadataList = JsonMetadataList.fromJson(reader);
+ assertEquals(3, metadataList.size());
+ for (Metadata metadata : metadataList) {
+ assertEquals("mock-text", metadata.get(MockTask.MARKER_KEY),
+ metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY));
+ assertEquals("3", metadata.get(MockTask.UNITS_KEY), "one run over
the three documents");
+ }
+ }
+
private static byte[] twoPagePdf() throws IOException {
try (PDDocument document = new PDDocument()) {
for (int i = 1; i <= 2; i++) {
diff --git
a/tika-server/tika-server-standard/src/test/resources/configs/tika-config-inference-bindings.json
b/tika-server/tika-server-standard/src/test/resources/configs/tika-config-inference-bindings.json
index 601646c106..f20098f2cc 100644
---
a/tika-server/tika-server-standard/src/test/resources/configs/tika-config-inference-bindings.json
+++
b/tika-server/tika-server-standard/src/test/resources/configs/tika-config-inference-bindings.json
@@ -2,7 +2,8 @@
"engines": { "mock": { "mock-engine": {} } },
"inference": [
{ "id": "mock-images", "engine": "mock", "input": "IMAGES", "tasks":
["mock-task"] },
- { "id": "mock-pages", "engine": "mock", "input": "PAGES", "tasks":
["mock-task"] }
+ { "id": "mock-pages", "engine": "mock", "input": "PAGES", "tasks":
["mock-task"] },
+ { "id": "mock-text", "engine": "mock", "input": "TEXT", "tasks":
["mock-task"] }
],
"auto-detect-parser": { "throwOnZeroBytes": false },
"parse-context": { "timeout-limits": { "progressTimeoutMillis": 30000 } }
