You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2026/09/17 23:24:49 UTC
(tika) branch main updated: TIKA-4900: media segments review fixes (#3189)
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 deaccdb8e5 TIKA-4900: media segments review fixes (#3189)
deaccdb8e5 is described below
commit deaccdb8e59c0b9f75d974186244104216009a39
Author: Tim Allison <ta...@apache.org>
AuthorDate: Thu Sep 17 19:24:43 2026 -0400
TIKA-4900: media segments review fixes (#3189)
---
CHANGES.txt | 11 +-
.../ROOT/pages/configuration/inference.adoc | 154 +++++++++++-
.../pages/migration-to-4x/chunk-strategies.adoc | 111 +++++++--
.../ROOT/pages/pipes/plugins/elasticsearch.adoc | 4 +
.../ROOT/pages/pipes/plugins/opensearch.adoc | 4 +
.../apache/tika/cli/TikaCLIMediaBindingsTest.java | 140 +++++++++++
.../tika/parser/inference/InferenceBinding.java | 106 ++++++--
.../tika/parser/inference/InferenceDispatcher.java | 3 +-
.../apache/tika/parser/inference/InputKind.java | 11 +-
.../apache/tika/parser/inference/MediaConfig.java | 90 +++++++
.../inference/{InputKind.java => Modality.java} | 27 +--
.../parser/inference/InferenceDispatcherTest.java | 54 ++++-
.../parser/inference/InferenceTextStageTest.java | 4 +-
.../elasticsearch/tests/ElasticsearchTest.java | 3 +-
.../pipes/opensearch/tests/OpenSearchTest.java | 3 +-
.../main/java/org/apache/tika/inference/Chunk.java | 25 ++
.../org/apache/tika/inference/ChunkSerializer.java | 24 ++
.../java/org/apache/tika/inference/EmbedTask.java | 205 +++++++++++++++-
.../org/apache/tika/inference/EmbeddingEngine.java | 11 +
.../org/apache/tika/inference/FfmpegSegmenter.java | 156 ++++++++++++
.../java/org/apache/tika/inference/MediaInput.java | 25 +-
.../org/apache/tika/inference/MediaSegmenter.java | 94 +++++++
.../tika/inference/OpenAIEmbeddingEngine.java | 40 +++
.../apache/tika/inference/ChunkSerializerTest.java | 25 ++
.../tika/inference/EmbedTaskMediaContractTest.java | 270 +++++++++++++++++++++
.../apache/tika/inference/EmbedTaskMediaTest.java | 240 ++++++++++++++++++
.../org/apache/tika/inference/EmbedTaskTest.java | 23 +-
.../apache/tika/inference/FfmpegSegmenterTest.java | 116 +++++++++
.../org/apache/tika/pipes/emitter/es/ESClient.java | 161 +++++++++++-
.../apache/tika/pipes/emitter/es/ESEmitter.java | 1 +
.../tika/pipes/emitter/es/ESEmitterConfig.java | 36 ++-
.../apache/tika/pipes/emitter/es/ESClientTest.java | 175 +++++++++++++
.../apache/tika/pipes/es/ConfigExamplesTest.java | 2 +
.../test/resources/config-examples/es-emitter.json | 1 +
.../pipes/emitter/opensearch/OpenSearchClient.java | 160 +++++++++++-
.../emitter/opensearch/OpenSearchEmitter.java | 1 +
.../opensearch/OpenSearchEmitterConfig.java | 34 ++-
.../emitter/opensearch/OpenSearchClientTest.java | 188 ++++++++++++++
.../tika/pipes/opensearch/ConfigExamplesTest.java | 2 +
.../config-examples/opensearch-emitter.json | 1 +
.../apache/tika/config/loader/InferenceLoader.java | 55 ++++-
.../tika/config/loader/InferenceLoaderTest.java | 57 +++++
42 files changed, 2724 insertions(+), 129 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 7a90a8a12f..e72d60114c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,16 +1,11 @@
Release 4.1.0 - unreleased
+ * Enable audio/video chunking with ffmpeg and inference (TIKA-4900).
+
* PDF incremental-update scanning reads the file in blocks instead of one
byte at a time; same output, about 4x less CPU (TIKA-4898).
- * Metadata never stores an unpaired UTF-16 surrogate: every write replaces
- one with U+FFFD (StringUtils.wellFormed), so no reader meets one; the
- Smile encoder on the pipes channel rejected them and the parent counted
- a worker crash. tk:content is exempt, since SafeContentHandler already
- guarantees it. A result the worker cannot encode is reported for its
- document with its own status (a fetch failure stays a fetch failure)
- instead of counting as a crash, and that report cannot itself overflow
- the payload limit (TIKA-4897).
+ * Metadata never stores an unpaired UTF-16 surrogate (TIKA-4897).
* openai-embedding-engine gains "requestParameters" to enable custom
parameters per vendor (TIKA-4896).
diff --git a/docs/modules/ROOT/pages/configuration/inference.adoc b/docs/modules/ROOT/pages/configuration/inference.adoc
index 8a958813a6..0ff602e259 100644
--- a/docs/modules/ROOT/pages/configuration/inference.adoc
+++ b/docs/modules/ROOT/pages/configuration/inference.adoc
@@ -27,7 +27,7 @@ NOTE: Inference is experimental in 4.1.0. The `inference` section, the keys of a
or VLM engine's entry, the `tika-inference` and `tika-vlm` classes, and the chunk and vector
output (`tk:chunks`, `tk:inference-released`) may change in a minor release without a
deprecation cycle, and no compatibility shims are kept for earlier shapes; 4.2 adds batched
-recognition, media inputs and document-level tasks. The `engines` map itself, and the
+recognition and document-level tasks. The `engines` map itself, and the
`text-recognizers` list, are stable. Pin the Tika version you validated against and read the
release notes before moving.
@@ -50,7 +50,7 @@ different models. Secrets and paths live here and only here.
`openai-embedding-engine` posts to `/v1/embeddings` in the OpenAI shape and sends up to
`maxBatchSize` images per request. Options: `baseUrl`, `model`, `apiKey`, `timeoutMillis`
(120000), `maxBatchSize` (32), `embeddingsPath`, `apiKeyHeaderName`, `apiKeyPrefix`,
-`requestParameters`, `imageInput`.
+`requestParameters`, `imageInput`, `mediaInput` (see <<media>>).
=== Vendor keys: `requestParameters`
@@ -169,17 +169,22 @@ A list of bindings. Each names an engine, says what it is fed, and what to ask:
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` is the extracted text of every document in the tree (see <<text>>).
- `MEDIA` (audio and video) is accepted by a binding but no task takes it in 4.1: 4.2 adds
- media chunkers (time windows for transcription, sampled frames or clips for an image or
- video model) that run at the flush, and a transcribing task.
+ `MEDIA` is every document in the `audio/` and `video/` families, cut into segments at the
+ flush (see <<media>>); playlists (m3u, pls, xspf, asx) are text that names other files and
+ are never media. Container types Tika detects as `application/*` (some ogg, matroska, mxf)
+ are not offered.
+`modality`:: What the engine is shown: `text`, `visual` or `audio`. Implied by `TEXT`,
+ `IMAGES` and `PAGES`; a `MEDIA` binding must say `visual` or `audio`.
`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; for `TEXT`, how many chunks across the tree; -1 for no limit.
+ document; for `TEXT`, how many chunks across the tree; for `MEDIA`, how many segments of one
+ file (480 unless set: the ceiling a request's `media` block cannot raise); -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.
+ ones are skipped and counted. For `MEDIA` the unit is one cut segment, not the file: a
+ film of any size is offered, and a segment over the limit is the one that is skipped.
`enabled`:: `false` keeps the binding configured but idle until a request selects it.
`_mime-include`, `_mime-exclude`:: Narrow by media type within the input kind. With neither,
an `IMAGES` binding skips vector and CAD image types (SVG, EMF, DWG, ...) that no embedding
@@ -190,7 +195,7 @@ on a chat engine), or an unknown key fails config load.
== How it runs
-Parsers do not call engines. Every image document Tika parses is offered to one dispatcher
+Parsers do not call engines. Every image, audio or video document Tika parses is offered to one dispatcher
after its own parser has run, whatever container it came from and however deep. The
dispatcher keeps the units of the whole document tree until the top-level parse ends, then
sends each binding's units to its engine in requests of up to `maxBatchSize` units: a docx
@@ -304,6 +309,139 @@ The 4.0 `openai-embedding-filter` and `jina-embedding-filter` keep working and e
document per request; the binding batches the tree and shares the engine with the other
bindings.
+[[media]]
+== Media
+
+A `MEDIA` binding embeds audio and video. The engine sees one channel, so the binding says
+which: `modality` is `visual` (the picture, as short silent clips) or `audio` (the sound,
+as mono Opus). Two bindings on one engine index both channels of a video:
+
+[source,json]
+----
+{
+ "engines": {
+ "jina": { "openai-embedding-engine": { "baseUrl": "https://api.jina.ai", "apiKey": "...",
+ "model": "jina-embeddings-v5-omni-small", "mediaInput": "object",
+ "requestParameters": { "task": "retrieval.passage", "dimensions": 1024 } } }
+ },
+ "inference": [
+ { "id": "jina-video", "engine": "jina", "input": "MEDIA", "modality": "visual" },
+ { "id": "jina-audio", "engine": "jina", "input": "MEDIA", "modality": "audio" }
+ ]
+}
+----
+
+Tika cuts the file into segments with ffmpeg (on the `PATH`) on one grid per document, set
+in the parse context:
+
+[source,json]
+----
+{ "parse-context": { "media": { "segment": { "seconds": 30, "overlap": 5 }, "maxSegments": 480 } } }
+----
+
+Those are the defaults: 30 s windows, each starting 25 s after the last, so any event of
+5 s or less is whole in at least one segment; `maxSegments` caps a document, the rest of the
+file is not embedded. A block with `overlap` outside `[0, seconds)`, or `maxSegments` of 0
+or below -1, fails at config load and per request. The segmenter defines the unit and every binding on the document sees
+the same segments, so each segment's chunks share the segment's id: a chunk carries a
+`temporal` locator (`start_ms`, `end_ms` on the container's presentation timeline), its
+`producer` (the binding), its `modality` (`visual` or `audio`) and a `correlator`, the
+segment's id. Group by `correlator` to get a segment with both channels:
+
+[source,json]
+----
+[
+ { "producer": "jina-video", "modality": "visual", "correlator": "t:25000-55000",
+ "vector": "<base64 float32>", "locators": { "temporal": [{"start_ms": 25000, "end_ms": 55000}] } },
+ { "producer": "jina-audio", "modality": "audio", "correlator": "t:25000-55000",
+ "vector": "<base64 float32>", "locators": { "temporal": [{"start_ms": 25000, "end_ms": 55000}] } }
+]
+----
+
+An audio file has no picture and a silent video has no sound: the binding for the missing
+channel skips the document; cover art in an audio file is not a picture. The binding's
+`maxChunks` (480 unless set) also caps segments per document, and since a request may send
+its own `media` block, it is the ceiling an operator relies on: the smaller of the two wins.
+`maxBytes` bounds one cut segment (the cuts are small: 360p H.264 and 32 kbps Opus); the
+file's own size is not gated. The engine must accept media, which the OpenAI shape has no
+form for, so the vendor's form is declared: `mediaInput: "object"` sends
+`{"audio": data-uri}` and `{"video": data-uri}` items as Jina names them; without it a MEDIA
+binding on the engine fails at load. Each segment is one request input; a request holds up
+to `maxBatchSize` of them (the engine default is 32; the cuts are 100-500 KB each as base64,
+so set it lower for a small request body). A request the engine rejects is retried one
+segment at a time from the same cut files; nothing is cut twice. A file ffprobe cannot read
+is reported once as the binding's warning and the next file runs.
+
+ffmpeg and ffprobe are looked up on the `PATH` once, at config load: a MEDIA binding without
+them logs one warning and stands by, and each audio or video document is then skipped with a
+warning in its metadata rather than failed, the way OCR behaves without tesseract.
+
+ffmpeg runs under the parse's time budget: each probe and cut asks for its own cap (60 s and
+5 min) and gets the smaller of that and what remains of the request's total timeout, with
+the wait counted as progress. When the budget runs out mid-file the task writes the segments
+it has embedded, stops, and reports the timeout once as the binding's warning on the root
+document; every later MEDIA binding's first probe fails fast the same way.
+
+=== Jina omni
+
+`jina-embeddings-v5-omni-small` (1024 dimensions) and `-nano` (768) embed text, images,
+pages, audio and video in one space, one modality per input item, mixed within a request.
+There is no Jina engine class; it is the OpenAI shape with Jina's keys:
+
+[cols="1,3"]
+|===
+|Setting |Value for Jina omni
+
+|`baseUrl`
+|`https://api.jina.ai` (the engine appends `/v1/embeddings`).
+
+|`apiKey`
+|The Jina key; sent as `Authorization: Bearer`.
+
+|`model`
+|`jina-embeddings-v5-omni-small` or `jina-embeddings-v5-omni-nano`.
+
+|`imageInput`
+|`object` (the default): images and pages go as `{"image": data-uri}`.
+
+|`mediaInput`
+|`object`: audio as `{"audio": data-uri}`, video as `{"video": data-uri}`. Required for
+MEDIA bindings.
+
+|`maxBatchSize`
+|Items per request across modalities; the segment cuts are 100-500 KB each as base64, so
+keep the request well under Jina's body limit (start at 8).
+
+|`requestParameters`
+|`task`: `retrieval.passage` for indexing (default nothing is sent; set it), `retrieval.query`
+for queries, `text-matching`, `classification`, `separation`. `dimensions`: a Matryoshka
+size up to the model's. `normalized`: `true` for unit vectors, which makes cosine scores
+comparable across the picture, sound and text fields. `truncate` and `late_chunking` apply
+to text items. `embedding_type` is refused: the engine reads float vectors by index.
+|===
+
+Jina's documented limits: images 5 MB, PDFs 8 MB and one per request; audio as WAV, MP3,
+FLAC, OGG, M4A or Opus with no documented duration cap; video as 32 uniformly sampled frames
+whatever its length, and nothing says the video item's audio track is used, which is why
+Tika sends the sound as its own item. The 32 frames are why the grid matters: a whole film
+as one input is 32 frames for the film, a 30 s segment is about one frame a second. Byte
+caps for audio and video items are not documented; measure before raising `maxBytes`.
+
+A query is the same call with the query task, and one vector searches every field:
+
+[source,bash]
+----
+curl https://api.jina.ai/v1/embeddings -H "Authorization: Bearer $JINA_API_KEY" \
+ -H "Content-Type: application/json" \
+ -d '{"model": "jina-embeddings-v5-omni-small", "task": "retrieval.query",
+ "dimensions": 1024, "normalized": true, "input": ["a horse jumping a fence"]}'
+----
+
+For search, index one document per segment so a hit is a segment and its two vectors score
+together: the Elasticsearch and OpenSearch emitters do that with `chunkStrategy: DOCUMENTS`,
+which merges the chunks of one correlator into one document (see
+xref:migration-to-4x/chunk-strategies.adoc[Chunk strategies]).
+
== Per request
[source,json]
diff --git a/docs/modules/ROOT/pages/migration-to-4x/chunk-strategies.adoc b/docs/modules/ROOT/pages/migration-to-4x/chunk-strategies.adoc
index ad8027e683..82ab0f651a 100644
--- a/docs/modules/ROOT/pages/migration-to-4x/chunk-strategies.adoc
+++ b/docs/modules/ROOT/pages/migration-to-4x/chunk-strategies.adoc
@@ -37,28 +37,38 @@ by which locators it carries:
|Chunk kind |Produced by, and what it carries
|Text
-|`AbstractEmbeddingFilter` subclasses (e.g. `OpenAIEmbeddingFilter`) split `tk:content` with
-the `MarkdownChunker` and send each piece to a text embeddings endpoint. Carries `text`, a
-`vector`, and a `TextLocator` (character offsets).
+|A `TEXT` binding (or the 4.0 `AbstractEmbeddingFilter` subclasses) splits `tk:content` with
+the `MarkdownChunker` and sends each piece to a text embeddings endpoint. Carries `text`, a
+vector, and a `TextLocator` (character offsets).
|Image
-|`OpenAIImageEmbeddingParser` sends rendered PDF page images (PDFBox or Poppler) to a
-CLIP-like image embeddings endpoint, e.g. Jina CLIP v2. Carries a `vector` and a
-`PaginatedLocator` (page number), and no `text`.
+|An `IMAGES` or `PAGES` binding sends pictures or rendered pages to an image embeddings
+endpoint. Carries a vector and a `PaginatedLocator` (page number) or an `EmbeddedLocator`,
+and no `text`.
+
+|Media segment
+|Two `MEDIA` bindings on the two channels of a video. Each writes a chunk with a
+`TemporalLocator` (millisecond range) and a vector; the two chunks of one segment share a
+`correlator`.
|===
-`SpatialLocator` (bounding box) and `TemporalLocator` (millisecond range) are already defined
-and round-tripped by `ChunkSerializer`; no component produces them yet. `EmbeddedLocator`
-(`id_path`, `name`) marks a chunk that was lifted from an embedded document into its parent;
-see <<lifting>>.
+`SpatialLocator` (bounding box) is defined and round-tripped by `ChunkSerializer`; no
+component produces it yet. `EmbeddedLocator` (`id_path`, `name`) marks a chunk that was
+lifted from an embedded document into its parent; see <<lifting>>.
-A single chunk looks like:
+A chunk carries one vector and says where it came from three ways: `producer` is the binding
+that wrote it, `modality` is what the engine saw (`text`, `visual`, `audio`), and
+`correlator` is the id of the unit it was cut from, assigned by the chunker that defined the
+unit, so chunks of one unit written by different bindings can be grouped (a segment's sound
+and picture; later, a page region's OCR text and its crop):
[source,json]
----
{
"text": "Revenue grew 15% year-over-year...",
"vector": "base64-encoded-float32-be",
+ "producer": "text-vectors",
+ "modality": "text",
"locators": {
"text": [{"start_offset": 0, "end_offset": 120}],
"paginated": [{"page": 1}]
@@ -66,9 +76,10 @@ A single chunk looks like:
}
----
-When several pipeline components produce chunks for the same document — the image embedder
-during parsing, the text embedder as a metadata filter — they merge into the same array via
-`ChunkSerializer.mergeInto()`.
+When several components produce chunks for the same document, they append to the same
+array via `ChunkSerializer.mergeInto()`. The 4.0 filters set neither `producer`, `modality`
+nor `correlator`. A `modality` value this Tika does not know (a newer writer's) reads as
+unset, so a foreign chunk never breaks the array for the chunks after it.
== What Tika Emits Today
@@ -165,7 +176,67 @@ xref:configuration/parsers/pdf-parser.adoc[PDF Parser]).
== Elasticsearch/OpenSearch Mapping
-With `nested` kNN support, a mapping like this works:
+Two ways to index chunks. The emitters' `chunkStrategy: DOCUMENTS` writes one document per
+unit next to the file's document: chunks sharing a `correlator` merge into one document, a
+chunk without one is its own, the locators are flattened to fields and every vector is
+decoded to a float array under `v.<producer>`:
+
+[source,json]
+----
+{
+ "file_id": "s3://bucket/talk.mp4", "container_id": "s3://bucket/talk.mp4",
+ "correlator": "t:25000-55000", "start_ms": 25000, "end_ms": 55000,
+ "v": { "jina-video": [0.01, ...], "jina-audio": [0.02, ...] }
+}
+----
+
+A hit is then a chunk, and Elasticsearch adds up scores per document, so two `knn` clauses
+(or an `rrf` retriever) over `v.jina-video` and `v.jina-audio` rank a segment that matches
+in both channels above one that matches in either: search one channel with one clause, both
+with two. That cannot be done with segments nested inside the file's document, where each
+clause scores the file by its best child, possibly a different child per clause. Mapping:
+
+[source,json]
+----
+{
+ "mappings": {
+ "properties": {
+ "file_id": {"type": "keyword"},
+ "container_id": {"type": "keyword"},
+ "correlator": {"type": "keyword"},
+ "start_ms": {"type": "long"},
+ "end_ms": {"type": "long"},
+ "page": {"type": "integer"},
+ "text": {"type": "text"},
+ "v": {
+ "properties": {
+ "jina-video": {"type": "dense_vector", "dims": 1024, "index": true, "similarity": "cosine"},
+ "jina-audio": {"type": "dense_vector", "dims": 1024, "index": true, "similarity": "cosine"}
+ }
+ }
+ }
+ }
+}
+----
+
+Group hits back to files with `collapse` on `container_id`, the emit key, which is stable
+across runs; `file_id` is the emitter's id for the document the chunk came from, and for an
+attachment that id is minted per run. Text chunks carry `start_offset`, `end_offset` and
+`text`; page chunks `page` (and `bbox` when the locator has one); lifted chunks
+`embedded_id_path` and `embedded_name`. `modality` and `producer` are not fields on a merged
+document: the vector's key under `v` is the producer.
+
+Chunk documents carry no join field, so `DOCUMENTS` needs `attachmentStrategy:
+SEPARATE_DOCUMENTS`; the emitter refuses the pair at load. Re-indexing a file replaces its
+chunk documents by id (`<id>-chunk-0` ...) but never deletes the tail of an earlier run that
+had more of them, and under `UPSERT` a removed binding's `v.<producer>` survives on the
+documents that are rewritten: delete by `container_id` before re-indexing a changed grid or
+binding set. A chunk whose `vector` is not base64 float32 keeps its other fields and loses
+the vector, with a warning; a `tk:chunks` value that is not a JSON array stays on the file's
+document as `INLINE` would write it.
+
+The default, `chunkStrategy: INLINE`, keeps `tk:chunks` inside the file's document as a
+JSON array. With `nested` kNN support, a mapping like this works:
[source,json]
----
@@ -180,6 +251,9 @@ With `nested` kNN support, a mapping like this works:
"type": "nested",
"properties": {
"text": {"type": "text"},
+ "producer": {"type": "keyword"},
+ "modality": {"type": "keyword"},
+ "correlator": {"type": "keyword"},
"vector": {
"type": "dense_vector",
"dims": 1024,
@@ -210,10 +284,9 @@ With `nested` kNN support, a mapping like this works:
}
----
-NOTE: `vector` holds a base64-encoded float32 array throughout Tika processing, and the
-emitter writes it as-is. To use a `dense_vector` mapping you need an ingest pipeline or
-custom serialization to decode it at index time; alternatively, map `vector` as `keyword`
-and decode at query time.
+NOTE: Inline, each `vector` is a base64-encoded big-endian float32 array and the emitter
+writes it as-is. Elasticsearch 9.3+ accepts that encoding for `dense_vector`; earlier
+versions need an ingest pipeline to decode it, or `chunkStrategy: DOCUMENTS`, which decodes.
== Alternatives Considered
diff --git a/docs/modules/ROOT/pages/pipes/plugins/elasticsearch.adoc b/docs/modules/ROOT/pages/pipes/plugins/elasticsearch.adoc
index ffd0757e71..1b63064fb8 100644
--- a/docs/modules/ROOT/pages/pipes/plugins/elasticsearch.adoc
+++ b/docs/modules/ROOT/pages/pipes/plugins/elasticsearch.adoc
@@ -133,6 +133,10 @@ include::example$pipes-elasticsearch-emitter.json[]
|_unused_
|Accepted by the config parser but never read. Control refresh visibility on the Elasticsearch side instead.
+|`chunkStrategy`
+|`INLINE`
+|Where `tk:chunks` goes. `INLINE` — inside the document, as a JSON array; `DOCUMENTS` — one document per unit (`_id` is the document's id plus `-chunk-N`): chunks sharing a `correlator` merge into one, with `file_id`, `container_id`, `correlator`, the locators as flat fields (`start_ms`/`end_ms`, `page`, `start_offset`/`end_offset`, `embedded_id_path`) and every vector decoded to a float array under `v.<producer>`; the document itself is written without `tk:chunks`. Needs `attachmentStra [...]
+
|`embeddedFileFieldName`
|_none_
|Name of the field used to hold embedded-file content (used by `PARENT_CHILD`).
diff --git a/docs/modules/ROOT/pages/pipes/plugins/opensearch.adoc b/docs/modules/ROOT/pages/pipes/plugins/opensearch.adoc
index 6eb07fc6fb..a57f45b48a 100644
--- a/docs/modules/ROOT/pages/pipes/plugins/opensearch.adoc
+++ b/docs/modules/ROOT/pages/pipes/plugins/opensearch.adoc
@@ -131,6 +131,10 @@ include::example$pipes-opensearch-emitter.json[]
|_unused_
|Accepted by the config parser but never read. Control refresh visibility on the OpenSearch side instead.
+|`chunkStrategy`
+|`INLINE`
+|Where `tk:chunks` goes. `INLINE` — inside the document, as a JSON array; `DOCUMENTS` — one document per unit (`_id` is the document's id plus `-chunk-N`): chunks sharing a `correlator` merge into one, with `file_id`, `container_id`, `correlator`, the locators as flat fields (`start_ms`/`end_ms`, `page`, `start_offset`/`end_offset`, `embedded_id_path`) and every vector decoded to a float array under `v.<producer>`; the document itself is written without `tk:chunks`. Needs `attachmentStra [...]
+
|`embeddedFileFieldName`
|_none_
|Name of the field used to hold embedded-file content (used by `PARENT_CHILD`).
diff --git a/tika-app/src/test/java/org/apache/tika/cli/TikaCLIMediaBindingsTest.java b/tika-app/src/test/java/org/apache/tika/cli/TikaCLIMediaBindingsTest.java
new file mode 100644
index 0000000000..977b58bb2b
--- /dev/null
+++ b/tika-app/src/test/java/org/apache/tika/cli/TikaCLIMediaBindingsTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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 static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+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.http.TikaTestHttpServer;
+import org.apache.tika.inference.Chunk;
+import org.apache.tika.inference.ChunkSerializer;
+import org.apache.tika.parser.inference.Modality;
+import org.apache.tika.utils.ProcessUtils;
+
+/**
+ * The consumer's path for media: a video on the command line, two bindings on one Jina engine
+ * (picture and audio), and -J shows two chunks per 30 s segment sharing the segment's id.
+ * Needs ffmpeg on the PATH; the clip is synthesized.
+ */
+public class TikaCLIMediaBindingsTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ @TempDir
+ Path tmp;
+
+ private static String response(int n) {
+ StringBuilder sb = new StringBuilder("{\"data\":[");
+ for (int i = 0; i < n; i++) {
+ sb.append(i > 0 ? "," : "").append("{\"index\":").append(i)
+ .append(",\"embedding\":[").append(i).append(".0,0.5]}");
+ }
+ return sb.append("]}").toString();
+ }
+
+ @Test
+ public void testSegmentsCarryBothVectors() throws Exception {
+ assumeTrue(answers("ffmpeg") && answers("ffprobe"), "ffmpeg not on the PATH");
+ Path mp4 = tmp.resolve("clip.mp4");
+ assertEquals(0, ProcessUtils.execute(new ProcessBuilder("ffmpeg", "-loglevel", "error",
+ "-y", "-f", "lavfi", "-i", "testsrc=duration=70:size=160x120:rate=5",
+ "-f", "lavfi", "-i", "sine=frequency=440:duration=70",
+ "-c:v", "libx264", "-preset", "ultrafast", "-c:a", "aac", "-shortest",
+ mp4.toString()), 60_000, 10_000, 10_000).getExitValue());
+ try (TikaTestHttpServer server = new TikaTestHttpServer()) {
+ // 70 s at 30/5 is three cells; one request per binding at the default batch size
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(3)));
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(3)));
+ Path config = tmp.resolve("tika-config.json");
+ Files.writeString(config, """
+ {
+ "engines": {
+ "jina": { "openai-embedding-engine": { "baseUrl": "BASE_URL", "apiKey": "x",
+ "model": "jina-embeddings-v5-omni-small", "mediaInput": "object",
+ "requestParameters": { "task": "retrieval.passage" } } }
+ },
+ "inference": [
+ { "id": "jina-video", "engine": "jina", "input": "MEDIA", "modality": "visual" },
+ { "id": "jina-audio", "engine": "jina", "input": "MEDIA", "modality": "audio" }
+ ]
+ }
+ """.replace("BASE_URL", server.url()));
+ JsonNode output = MAPPER.readTree(run("--config=" + config, "-J", mp4.toUri().toString()));
+ assertEquals(2, server.getRequestCount(), "one request per channel");
+ JsonNode video = MAPPER.readTree(server.takeRequest().body());
+ assertEquals(3, video.get("input").size());
+ assertTrue(video.get("input").get(0).get("video").asText().startsWith("data:video/mp4;base64,"));
+ assertEquals("retrieval.passage", video.get("task").asText());
+ JsonNode audio = MAPPER.readTree(server.takeRequest().body());
+ assertTrue(audio.get("input").get(2).get("audio").asText().startsWith("data:audio/ogg;base64,"));
+
+ JsonNode root = output.get(0);
+ assertNotNull(root.get("tk:chunks"), root.toString());
+ List<Chunk> chunks = ChunkSerializer.fromJson(root.get("tk:chunks").asText());
+ assertEquals(6, chunks.size(), "three cells, two channels");
+ Chunk video2 = chunks.get(1);
+ Chunk audio2 = chunks.get(4);
+ assertEquals(25000, video2.getLocators().getTemporal().get(0).getStartMs());
+ assertEquals(55000, video2.getLocators().getTemporal().get(0).getEndMs());
+ assertEquals("t:25000-55000", video2.getCorrelator());
+ assertEquals(video2.getCorrelator(), audio2.getCorrelator());
+ assertEquals(Modality.VISUAL, video2.getModality());
+ assertEquals(Modality.AUDIO, audio2.getModality());
+ assertEquals("jina-audio", audio2.getProducer());
+ assertEquals(1.0f, video2.getVector()[0]);
+ assertEquals(1.0f, audio2.getVector()[0]);
+ }
+ }
+
+ private static boolean answers(String tool) {
+ try {
+ return ProcessUtils.execute(new ProcessBuilder(tool, "-version"), 10_000, 1000, 1000)
+ .getExitValue() == 0;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
+ 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/parser/inference/InferenceBinding.java b/tika-core/src/main/java/org/apache/tika/parser/inference/InferenceBinding.java
index a9ca185223..913e765168 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
@@ -18,6 +18,7 @@ package org.apache.tika.parser.inference;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
import java.util.Set;
import org.apache.tika.mime.MediaType;
@@ -35,26 +36,92 @@ public final class InferenceBinding {
private final long maxBytes;
private final boolean enabled;
private final TextChunker chunker;
+ private final Modality modality;
+
+ private InferenceBinding(Builder b) {
+ this.id = b.id;
+ this.engine = b.engine;
+ this.input = b.input;
+ this.tasks = List.copyOf(b.tasks);
+ this.include = b.include == null ? Collections.emptySet() : Set.copyOf(b.include);
+ this.exclude = b.exclude == null ? Collections.emptySet() : Set.copyOf(b.exclude);
+ this.maxChunks = b.maxChunks;
+ this.maxBytes = b.maxBytes;
+ this.enabled = b.enabled;
+ this.chunker = b.chunker;
+ this.modality = b.modality == null ? Modality.implied(b.input) : b.modality;
+ }
- 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);
+ /** The three things every binding needs; everything else has a default. */
+ public static Builder builder(String id, String engine, InputKind input) {
+ return new Builder(id, engine, input);
}
- 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;
- this.tasks = List.copyOf(tasks);
- this.include = include == null ? Collections.emptySet() : Set.copyOf(include);
- this.exclude = exclude == null ? Collections.emptySet() : Set.copyOf(exclude);
- this.maxChunks = maxChunks;
- this.maxBytes = maxBytes;
- this.enabled = enabled;
+ public static final class Builder {
+ private final String id;
+ private final String engine;
+ private final InputKind input;
+ private List<String> tasks = List.of("embed");
+ private Set<MediaType> include;
+ private Set<MediaType> exclude;
+ private int maxChunks = -1;
+ private long maxBytes = -1;
+ private boolean enabled = true;
+ private TextChunker chunker;
+ private Modality modality;
+
+ private Builder(String id, String engine, InputKind input) {
+ this.id = Objects.requireNonNull(id, "id");
+ this.engine = Objects.requireNonNull(engine, "engine");
+ this.input = Objects.requireNonNull(input, "input");
+ }
+
+ public Builder tasks(List<String> tasks) {
+ this.tasks = tasks;
+ return this;
+ }
+
+ public Builder include(Set<MediaType> include) {
+ this.include = include;
+ return this;
+ }
+
+ public Builder exclude(Set<MediaType> exclude) {
+ this.exclude = exclude;
+ return this;
+ }
+
+ /** Units per document; -1 (the default) for no limit. */
+ public Builder maxChunks(int maxChunks) {
+ this.maxChunks = maxChunks;
+ return this;
+ }
+
+ /** Largest unit accepted, in bytes; -1 (the default) for no limit. */
+ public Builder maxBytes(long maxBytes) {
+ this.maxBytes = maxBytes;
+ return this;
+ }
+
+ public Builder enabled(boolean enabled) {
+ this.enabled = enabled;
+ return this;
+ }
+
+ public Builder chunker(TextChunker chunker) {
+ this.chunker = chunker;
+ return this;
+ }
+
+ /** Required for MEDIA; implied by the input kind otherwise. */
+ public Builder modality(Modality modality) {
+ this.modality = modality;
+ return this;
+ }
+
+ public InferenceBinding build() {
+ return new InferenceBinding(this);
+ }
}
public String getId() {
@@ -87,6 +154,11 @@ public final class InferenceBinding {
return enabled;
}
+ /** What the engine is shown; implied by the input kind except for MEDIA, where it is required. */
+ public Modality getModality() {
+ return modality;
+ }
+
/** How a TEXT binding cuts a document's text; null means the whole text is one chunk. */
public TextChunker getChunker() {
return chunker;
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 fd3ace34e2..49e2cbda81 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
@@ -170,7 +170,8 @@ public final class InferenceDispatcher implements ParseHook, TransientParseState
if (!runs(binding, context) || !binding.accepts(kind, type)) {
continue;
}
- if (binding.getMaxBytes() >= 0 && size > binding.getMaxBytes()) {
+ // a media unit is the whole file; its maxBytes bounds one segment, in the task
+ if (kind != InputKind.MEDIA && binding.getMaxBytes() >= 0 && size > binding.getMaxBytes()) {
state.dropped.merge(binding.getId() + " over maxBytes", 1, Integer::sum);
continue;
}
diff --git a/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java b/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java
index af1d6a1fd3..efa0ee2e3c 100644
--- a/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java
+++ b/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java
@@ -16,15 +16,22 @@
*/
package org.apache.tika.parser.inference;
+import java.util.Set;
+
import org.apache.tika.mime.MediaType;
/** What a binding is fed: a document's text, its rendered pages, an image, or media bytes. */
public enum InputKind {
TEXT, PAGES, IMAGES, MEDIA;
- /** The kind a document's bytes are, by media type family; null for the rest. */
+ // text that names other files; ffmpeg would open them
+ private static final Set<String> PLAYLISTS = Set.of("audio/x-mpegurl",
+ "application/vnd.apple.mpegurl", "video/vnd.mpegurl", "audio/x-scpls",
+ "application/xspf+xml", "audio/x-ms-wax", "video/x-ms-wvx", "application/x-ms-asx");
+
+ /** The kind a document's bytes are, by media type family; null for the rest and for playlists. */
public static InputKind of(MediaType type) {
- if (type == null) {
+ if (type == null || PLAYLISTS.contains(type.getBaseType().toString())) {
return null;
}
switch (type.getType()) {
diff --git a/tika-core/src/main/java/org/apache/tika/parser/inference/MediaConfig.java b/tika-core/src/main/java/org/apache/tika/parser/inference/MediaConfig.java
new file mode 100644
index 0000000000..a359ed495a
--- /dev/null
+++ b/tika-core/src/main/java/org/apache/tika/parser/inference/MediaConfig.java
@@ -0,0 +1,90 @@
+/*
+ * 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 java.io.Serializable;
+
+import org.apache.tika.annotation.TikaComponent;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.exception.TikaConfigException;
+
+/**
+ * The {@code "media"} parse-context block: how audio and video are cut into segments before
+ * a MEDIA binding sees them. One grid per document, shared by every binding, so the vectors of
+ * one segment land on one chunk. Defaults: 30 s windows, 5 s overlap, 480 segments.
+ */
+@TikaComponent(name = "media", spi = false)
+public class MediaConfig implements Serializable, Initializable {
+
+ private static final long serialVersionUID = 1L;
+
+ public static class Segment implements Serializable {
+ private static final long serialVersionUID = 1L;
+ private int seconds = 30;
+ private int overlap = 5;
+
+ public int getSeconds() {
+ return seconds;
+ }
+
+ public void setSeconds(int seconds) {
+ this.seconds = seconds;
+ }
+
+ /** Seconds each segment extends past the next one's start; the longest event kept whole. */
+ public int getOverlap() {
+ return overlap;
+ }
+
+ public void setOverlap(int overlap) {
+ this.overlap = overlap;
+ }
+ }
+
+ private Segment segment = new Segment();
+ private int maxSegments = 480;
+
+ public Segment getSegment() {
+ return segment;
+ }
+
+ public void setSegment(Segment segment) {
+ this.segment = segment == null ? new Segment() : segment;
+ }
+
+ /** Segments per document; the rest of the file is not embedded. -1 for no limit. */
+ public int getMaxSegments() {
+ return maxSegments;
+ }
+
+ public void setMaxSegments(int maxSegments) {
+ this.maxSegments = maxSegments;
+ }
+
+ /** Runs at config load and per request; the task checks again for a programmatic block. */
+ @Override
+ public void initialize() throws TikaConfigException {
+ if (segment.seconds <= 0 || segment.overlap < 0 || segment.overlap >= segment.seconds) {
+ throw new TikaConfigException("media.segment needs seconds > overlap >= 0, not "
+ + segment.seconds + "/" + segment.overlap);
+ }
+ if (maxSegments == 0 || maxSegments < -1) {
+ throw new TikaConfigException("media.maxSegments must be -1 (no limit) or positive, "
+ + "not " + maxSegments);
+ }
+ }
+}
diff --git a/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java b/tika-core/src/main/java/org/apache/tika/parser/inference/Modality.java
similarity index 61%
copy from tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java
copy to tika-core/src/main/java/org/apache/tika/parser/inference/Modality.java
index af1d6a1fd3..b0f2534c47 100644
--- a/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java
+++ b/tika-core/src/main/java/org/apache/tika/parser/inference/Modality.java
@@ -16,23 +16,18 @@
*/
package org.apache.tika.parser.inference;
-import org.apache.tika.mime.MediaType;
+/** What an engine is shown for a unit, and so what a vector stands for. */
+public enum Modality {
+ TEXT, VISUAL, AUDIO;
-/** What a binding is fed: a document's text, its rendered pages, an image, or media bytes. */
-public enum InputKind {
- TEXT, PAGES, IMAGES, MEDIA;
-
- /** The kind a document's bytes are, by media type family; null for the rest. */
- public static InputKind of(MediaType type) {
- if (type == null) {
- return null;
- }
- switch (type.getType()) {
- case "image":
- return IMAGES;
- case "audio":
- case "video":
- return MEDIA;
+ /** The one modality an input kind implies; null for MEDIA, whose bindings must say. */
+ public static Modality implied(InputKind kind) {
+ switch (kind) {
+ case TEXT:
+ return TEXT;
+ case IMAGES:
+ case PAGES:
+ return VISUAL;
default:
return null;
}
diff --git a/tika-core/src/test/java/org/apache/tika/parser/inference/InferenceDispatcherTest.java b/tika-core/src/test/java/org/apache/tika/parser/inference/InferenceDispatcherTest.java
index 265a887f10..1cb3bda51d 100644
--- a/tika-core/src/test/java/org/apache/tika/parser/inference/InferenceDispatcherTest.java
+++ b/tika-core/src/test/java/org/apache/tika/parser/inference/InferenceDispatcherTest.java
@@ -43,6 +43,7 @@ import org.xml.sax.SAXException;
import org.xml.sax.helpers.DefaultHandler;
import org.apache.tika.detect.Detector;
+import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.extractor.EmbeddedDocumentUtil;
import org.apache.tika.extractor.EmbeddedMetadataLookup;
@@ -151,8 +152,8 @@ public class InferenceDispatcherTest {
private static InferenceBinding binding(String id, InputKind kind, Set<MediaType> include,
int maxChunks) {
- return new InferenceBinding(id, "engine", kind, List.of("embed"), include, null,
- maxChunks, -1, true);
+ return InferenceBinding.builder(id, "engine", kind).include(include)
+ .maxChunks(maxChunks).build();
}
@TempDir
@@ -246,8 +247,8 @@ public class InferenceDispatcherTest {
InferenceDispatcher dispatcher = new InferenceDispatcher(List.of(
new InferenceDispatcher.Bound(binding("png", InputKind.IMAGES, null, -1),
new RecordingEngine(), List.of(pngTask)),
- new InferenceDispatcher.Bound(new InferenceBinding("small", "engine",
- InputKind.IMAGES, List.of("embed"), null, null, -1, 2, true),
+ new InferenceDispatcher.Bound(InferenceBinding.builder("small", "engine",
+ InputKind.IMAGES).maxBytes(2).build(),
new RecordingEngine(), List.of(smallTask))));
ParseContext off = new ParseContext();
@@ -515,4 +516,49 @@ public class InferenceDispatcherTest {
assertFalse(Files.exists(held));
assertNull(context.get(InferenceDispatcher.State.class));
}
+
+ /** A MEDIA unit is the whole file; the segment cap applies later, not the file size. */
+ @Test
+ public void testMediaIsNotSizeGatedAtTheOffer() throws Exception {
+ RecordingTask task = new RecordingTask();
+ InferenceDispatcher dispatcher = new InferenceDispatcher(List.of(
+ new InferenceDispatcher.Bound(InferenceBinding.builder("v", "engine",
+ InputKind.MEDIA).modality(Modality.VISUAL).maxBytes(2).build(),
+ new RecordingEngine(), List.of(task))));
+ ParseContext context = new ParseContext();
+ MediaType mp4 = MediaType.video("mp4");
+ assertTrue(dispatcher.wants(mp4, new Metadata(), context));
+ dispatcher.offer(mp4, new Metadata(), new Metadata(), file("more than two"), context);
+ dispatcher.flush(new Metadata(), context);
+ assertEquals(1, task.runs.size());
+ assertEquals(1, task.runs.get(0).size(), "maxBytes does not drop a media file");
+
+ // a playlist is text that names other files; it is never a media unit
+ assertFalse(dispatcher.wants(MediaType.audio("x-mpegurl"), new Metadata(), context));
+ assertNull(InputKind.of(MediaType.parse("application/vnd.apple.mpegurl")));
+ assertNull(InputKind.of(MediaType.audio("x-scpls")));
+ assertNull(InputKind.of(MediaType.parse("video/vnd.mpegurl")));
+ assertNull(InputKind.of(MediaType.parse("application/xspf+xml")));
+ assertEquals(InputKind.MEDIA, InputKind.of(MediaType.audio("mpeg")));
+ }
+
+ @Test
+ public void testMediaConfigValidation() throws Exception {
+ MediaConfig config = new MediaConfig();
+ config.initialize();
+ config.getSegment().setOverlap(-1);
+ assertThrows(TikaConfigException.class, config::initialize, "negative overlap");
+ config.getSegment().setOverlap(30);
+ assertThrows(TikaConfigException.class, config::initialize, "overlap >= seconds");
+ config.getSegment().setOverlap(0);
+ config.getSegment().setSeconds(0);
+ assertThrows(TikaConfigException.class, config::initialize, "no window");
+ config.getSegment().setSeconds(10);
+ config.setMaxSegments(0);
+ assertThrows(TikaConfigException.class, config::initialize, "0 embeds nothing");
+ config.setMaxSegments(-2);
+ assertThrows(TikaConfigException.class, config::initialize);
+ config.setMaxSegments(-1);
+ config.initialize();
+ }
}
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
index 87e89a2344..5916402853 100644
--- 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
@@ -45,8 +45,8 @@ public class InferenceTextStageTest {
}
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);
+ return InferenceBinding.builder(id, "engine", InputKind.TEXT).tasks(List.of("t"))
+ .include(include).maxBytes(maxBytes).build();
}
private static Metadata doc(String type, String content) {
diff --git a/tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java b/tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java
index 571cbaf9a0..fd5b07e377 100644
--- a/tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java
+++ b/tika-integration-tests/tika-pipes-es-integration-tests/src/test/java/org/apache/tika/pipes/elasticsearch/tests/ElasticsearchTest.java
@@ -611,7 +611,8 @@ public class ElasticsearchTest {
ESEmitterConfig.UpdateStrategy.OVERWRITE,
10, DEFAULT_EMBEDDED_FILE_FIELD_NAME, null,
new HttpClientConfig(null, null, null,
- -1, -1, null, 0, false));
+ -1, -1, null, 0, false),
+ ESEmitterConfig.ChunkStrategy.INLINE);
return new ElasticsearchTestClient(config,
httpClientFactory.build());
diff --git a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java
index 040472f7c1..be2909474c 100644
--- a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java
+++ b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/OpenSearchTest.java
@@ -426,7 +426,8 @@ public class OpenSearchTest {
httpClientFactory.setVerifySsl(false);
OpenSearchEmitterConfig config = new OpenSearchEmitterConfig(CONTAINER.getHttpHostAddress(), "_id", OpenSearchEmitterConfig.AttachmentStrategy.SEPARATE_DOCUMENTS,
OpenSearchEmitterConfig.UpdateStrategy.OVERWRITE, 10, DEFAULT_EMBEDDED_FILE_FIELD_NAME,
- new HttpClientConfig(null, null, null, -1, -1, null, -1, false));
+ new HttpClientConfig(null, null, null, -1, -1, null, -1, false),
+ OpenSearchEmitterConfig.ChunkStrategy.INLINE);
return new OpensearchTestClient(config, httpClientFactory.build());
}
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 21a246ee20..1df8f123e6 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
@@ -18,6 +18,7 @@ package org.apache.tika.inference;
import org.apache.tika.inference.locator.Locators;
import org.apache.tika.inference.locator.TextLocator;
+import org.apache.tika.parser.inference.Modality;
/**
* A content chunk with multimodal locators and an optional embedding vector.
@@ -35,6 +36,8 @@ public class Chunk {
private final Locators locators;
private float[] vector;
private String producer;
+ private Modality modality;
+ private String correlator;
public Chunk(String text, Locators locators) {
this.text = text;
@@ -88,6 +91,28 @@ public class Chunk {
this.producer = producer;
}
+ /** What the engine was shown for this vector; null when the writer did not say. */
+ public Modality getModality() {
+ return modality;
+ }
+
+ public void setModality(Modality modality) {
+ this.modality = modality;
+ }
+
+ /**
+ * The id of the unit this chunk was cut from, assigned by the chunker that defined it, so
+ * chunks of one unit written by different bindings (a segment's sound and picture, a
+ * region's text and crop) can be grouped; null for a chunk that is its own unit.
+ */
+ public String getCorrelator() {
+ return correlator;
+ }
+
+ public void setCorrelator(String correlator) {
+ this.correlator = correlator;
+ }
+
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 67fdc53aac..fbc0bab309 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
@@ -19,6 +19,7 @@ package org.apache.tika.inference;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Locale;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -32,6 +33,7 @@ import org.apache.tika.inference.locator.SpatialLocator;
import org.apache.tika.inference.locator.TemporalLocator;
import org.apache.tika.inference.locator.TextLocator;
import org.apache.tika.metadata.TikaCoreProperties;
+import org.apache.tika.parser.inference.Modality;
/**
* Serializes and deserializes a list of {@link Chunk} objects to/from JSON.
@@ -63,6 +65,12 @@ public final class ChunkSerializer {
if (chunk.getProducer() != null) {
node.put("producer", chunk.getProducer());
}
+ if (chunk.getModality() != null) {
+ node.put("modality", chunk.getModality().name().toLowerCase(Locale.ROOT));
+ }
+ if (chunk.getCorrelator() != null) {
+ node.put("correlator", chunk.getCorrelator());
+ }
serializeLocators(node, chunk.getLocators());
}
return MAPPER.writeValueAsString(array);
@@ -113,6 +121,12 @@ public final class ChunkSerializer {
if (node.hasNonNull("producer")) {
chunk.setProducer(node.get("producer").asText());
}
+ if (node.hasNonNull("modality")) {
+ chunk.setModality(modality(node.get("modality").asText()));
+ }
+ if (node.hasNonNull("correlator")) {
+ chunk.setCorrelator(node.get("correlator").asText());
+ }
JsonNode vectorNode = node.get("vector");
if (vectorNode != null && !vectorNode.isNull()) {
@@ -125,6 +139,16 @@ public final class ChunkSerializer {
// ---- locator serialization --------------------------------------------
+ /** A value this reader does not know (a newer writer's) reads as unsaid, not as a failure. */
+ private static Modality modality(String text) {
+ for (Modality m : Modality.values()) {
+ if (m.name().equalsIgnoreCase(text)) {
+ return m;
+ }
+ }
+ return null;
+ }
+
private static void serializeLocators(ObjectNode parent, Locators loc) {
if (loc == null || loc.isEmpty()) {
return;
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 63fbe001b6..e23f9e2448 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
@@ -17,16 +17,24 @@
package org.apache.tika.inference;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Stream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.tika.annotation.TikaComponent;
+import org.apache.tika.config.ParseContextConfig;
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.inference.locator.TemporalLocator;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.metadata.TikaPagedText;
@@ -36,9 +44,11 @@ import org.apache.tika.parser.inference.InferenceBinding;
import org.apache.tika.parser.inference.InferenceTask;
import org.apache.tika.parser.inference.InferenceUnit;
import org.apache.tika.parser.inference.InputKind;
+import org.apache.tika.parser.inference.MediaConfig;
+import org.apache.tika.parser.inference.Modality;
/**
- * The {@code embed} task. Images and pages: one vector chunk per unit, written on the
+ * The {@code embed} task. Media: see {@link #embedMedia}. Images and pages: one vector chunk per unit, written on the
* unit's destination 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
@@ -47,6 +57,11 @@ import org.apache.tika.parser.inference.InputKind;
@TikaComponent(name = "embed", spi = false)
public class EmbedTask implements InferenceTask {
+ private static final Logger LOG = LoggerFactory.getLogger(EmbedTask.class);
+
+ /** The one segmenter; tests swap in a fake. */
+ static MediaSegmenter segmenter = new FfmpegSegmenter();
+
@Override
public void validate(InferenceBinding binding, Engine engine) throws TikaConfigException {
if (!(engine instanceof EmbeddingEngine)) {
@@ -54,8 +69,19 @@ public class EmbedTask implements InferenceTask {
+ binding.getEngine() + "\" is not one");
}
if (binding.getInput() == InputKind.MEDIA) {
- throw new TikaConfigException("task \"embed\" takes TEXT, IMAGES or PAGES; binding \""
- + binding.getId() + "\" is on " + binding.getInput());
+ if (!((EmbeddingEngine) engine).supportsMedia()) {
+ throw new TikaConfigException("binding \"" + binding.getId() + "\" is on MEDIA "
+ + "but engine \"" + binding.getEngine() + "\" does not embed audio or video");
+ }
+ if (binding.getModality() != Modality.AUDIO && binding.getModality() != Modality.VISUAL) {
+ throw new TikaConfigException("binding \"" + binding.getId() + "\" is on MEDIA "
+ + "and needs \"modality\": \"audio\" or \"visual\"");
+ }
+ if (!segmenter.available()) {
+ // an external tool, so stand by like OCR without tesseract: warn once, skip at run
+ LOG.warn("binding \"{}\" is on MEDIA but ffmpeg/ffprobe are not on the PATH; "
+ + "audio and video will be skipped until they are", binding.getId());
+ }
}
}
@@ -67,6 +93,10 @@ public class EmbedTask implements InferenceTask {
embedText(binding, units, embedder, context);
return;
}
+ if (binding.getInput() == InputKind.MEDIA) {
+ embedMedia(binding, units, embedder, context);
+ return;
+ }
int batchSize = Math.max(1, embedder.getMaxBatchSize());
TikaException first = null;
for (int start = 0; start < units.size(); start += batchSize) {
@@ -114,6 +144,7 @@ public class EmbedTask implements InferenceTask {
for (int[] span : spans) {
Chunk chunk = new Chunk(text.substring(span[0], span[1]), span[0], span[1]);
chunk.setProducer(binding.getId());
+ chunk.setModality(Modality.TEXT);
owned.add(new Owned(chunk, unit.getDestination()));
}
}
@@ -184,6 +215,173 @@ public class EmbedTask implements InferenceTask {
return groups;
}
+ /**
+ * Cuts every unit on the document's {@code media} grid and embeds one channel per cell;
+ * the cell's id is the chunk's correlator, so a second binding on the other channel writes
+ * a chunk that groups with this one.
+ * <p>
+ * Contract, per unit: a unit without the channel is skipped; a unit whose probe fails is
+ * reported once and the next unit runs; a request the engine rejects is retried one cell at
+ * a time, re-reading the cut files, never re-cutting; whatever was embedded before any
+ * failure is written on the unit before the failure is reported; the unit's temp dir is
+ * gone on every exit. A {@link MediaSegmenter.Timeout} (the parse budget ran out under the
+ * tool) stops the whole task after that write, so an exhausted budget does not iterate
+ * every remaining cell. An {@link IOException} is infrastructure and stops the task the
+ * same way, after the write.
+ */
+ private static void embedMedia(InferenceBinding binding, List<InferenceUnit> units,
+ EmbeddingEngine embedder, ParseContext context)
+ throws IOException, TikaException {
+ MediaConfig config;
+ try {
+ config = ParseContextConfig.getConfig(context, "media", MediaConfig.class,
+ new MediaConfig());
+ // a programmatic block skipped the loader's check
+ config.initialize();
+ } catch (TikaConfigException e) {
+ throw new TikaException("invalid \"media\" in parse-context", e);
+ }
+ if (!segmenter.available()) {
+ for (InferenceUnit unit : units) {
+ unit.getDestination().add(TikaCoreProperties.TIKA_META_EXCEPTION_WARNING,
+ "inference binding " + binding.getId()
+ + ": skipped, ffmpeg/ffprobe not on the PATH");
+ }
+ return;
+ }
+ Modality modality = binding.getModality();
+ int cap = config.getMaxSegments() >= 0 ? config.getMaxSegments() : Integer.MAX_VALUE;
+ if (binding.getMaxChunks() >= 0) {
+ cap = Math.min(cap, binding.getMaxChunks());
+ }
+ int batchSize = Math.max(1, embedder.getMaxBatchSize());
+ TikaException first = null;
+ for (InferenceUnit unit : units) {
+ MediaSegmenter.Probe probe;
+ try {
+ probe = segmenter.probe(unit.getPath(), context);
+ } catch (MediaSegmenter.Timeout t) {
+ throw t;
+ } catch (TikaException e) {
+ // one unreadable file does not cost the others their vectors
+ first = first == null ? e : first;
+ continue;
+ }
+ if (modality == Modality.AUDIO ? !probe.hasAudio() : !probe.hasVideo()) {
+ continue;
+ }
+ List<MediaSegmenter.Cell> cells = MediaSegmenter.cells(probe.durationMs(), config, cap);
+ Path dir = Files.createTempDirectory("tika-media-");
+ List<Chunk> chunks = new ArrayList<>();
+ try {
+ for (int start = 0; start < cells.size(); start += batchSize) {
+ List<MediaSegmenter.Cell> batch =
+ cells.subList(start, Math.min(start + batchSize, cells.size()));
+ try {
+ chunks.addAll(embedCells(binding, unit, batch, modality, dir, embedder,
+ context));
+ } catch (MediaSegmenter.Timeout t) {
+ throw t;
+ } catch (TikaException e) {
+ if (batch.size() == 1) {
+ first = first == null ? e : first;
+ continue;
+ }
+ for (MediaSegmenter.Cell cell : batch) {
+ try {
+ chunks.addAll(embedCells(binding, unit, List.of(cell), modality,
+ dir, embedder, context));
+ } catch (MediaSegmenter.Timeout t) {
+ throw t;
+ } catch (TikaException single) {
+ first = first == null ? single : first;
+ }
+ }
+ }
+ }
+ } catch (IOException | RuntimeException e) {
+ write(unit, chunks, e);
+ throw e;
+ } catch (MediaSegmenter.Timeout t) {
+ write(unit, chunks, t);
+ throw t;
+ } finally {
+ deleteTree(dir);
+ }
+ write(unit, chunks, null);
+ }
+ if (first != null) {
+ throw first;
+ }
+ }
+
+ /** Writes what was embedded; a write failure rides on the failure already in flight. */
+ private static void write(InferenceUnit unit, List<Chunk> chunks, Exception inFlight)
+ throws IOException {
+ if (chunks.isEmpty()) {
+ return;
+ }
+ try {
+ ChunkTarget.of(unit).write(chunks, TikaCoreProperties.TIKA_CHUNKS.getName());
+ } catch (IOException e) {
+ if (inFlight == null) {
+ throw e;
+ }
+ inFlight.addSuppressed(e);
+ }
+ }
+
+ private static void deleteTree(Path dir) {
+ try (Stream<Path> paths = Files.list(dir)) {
+ paths.forEach(p -> {
+ try {
+ Files.deleteIfExists(p);
+ } catch (IOException ignored) {
+ // best effort; the directory is under the JVM's temp dir
+ }
+ });
+ Files.deleteIfExists(dir);
+ } catch (IOException ignored) {
+ // as above
+ }
+ }
+
+ /** One request for a batch of cells; the cut files are deleted only once it succeeded. */
+ private static List<Chunk> embedCells(InferenceBinding binding, InferenceUnit unit,
+ List<MediaSegmenter.Cell> cells, Modality modality,
+ Path dir, EmbeddingEngine embedder,
+ ParseContext context)
+ throws IOException, TikaException {
+ List<MediaInput> inputs = new ArrayList<>(cells.size());
+ List<Path> cuts = new ArrayList<>(cells.size());
+ String mime = MediaSegmenter.mimeType(modality);
+ for (MediaSegmenter.Cell cell : cells) {
+ Path cut = segmenter.cut(unit.getPath(), cell, modality, dir, context);
+ if (binding.getMaxBytes() >= 0 && Files.size(cut) > binding.getMaxBytes()) {
+ Files.deleteIfExists(cut);
+ throw new TikaException("segment " + cell.index() + " is over maxBytes");
+ }
+ inputs.add(new MediaInput(Files.readAllBytes(cut), mime, modality));
+ cuts.add(cut);
+ }
+ List<float[]> vectors = embedder.embedMedia(inputs, context);
+ for (Path cut : cuts) {
+ Files.deleteIfExists(cut);
+ }
+ List<Chunk> chunks = new ArrayList<>(cells.size());
+ for (int i = 0; i < cells.size(); i++) {
+ MediaSegmenter.Cell cell = cells.get(i);
+ Chunk chunk = new Chunk(null, new Locators()
+ .addTemporal(new TemporalLocator(cell.startMs(), cell.endMs())));
+ chunk.setVector(vectors.get(i));
+ chunk.setProducer(binding.getId());
+ chunk.setModality(modality);
+ chunk.setCorrelator(MediaSegmenter.correlator(unit.getTargetIdPath(), cell));
+ chunks.add(chunk);
+ }
+ return chunks;
+ }
+
private static void embed(InferenceBinding binding, List<InferenceUnit> batch,
EmbeddingEngine embedder, ParseContext context)
throws IOException, TikaException {
@@ -206,6 +404,7 @@ public class EmbedTask implements InferenceTask {
Chunk chunk = new Chunk(null, locators);
chunk.setVector(vectors.get(i));
chunk.setProducer(binding.getId());
+ chunk.setModality(binding.getModality());
ChunkTarget.of(unit).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 86a38b1580..df6a18be6b 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
@@ -35,4 +35,15 @@ public interface EmbeddingEngine extends Engine {
throws IOException, TikaException;
int getMaxBatchSize();
+
+ /** Whether {@link #embedMedia} is implemented; a MEDIA binding on an engine without it fails at load. */
+ default boolean supportsMedia() {
+ return false;
+ }
+
+ /** One request; the vectors come back in the order of the inputs. */
+ default List<float[]> embedMedia(List<MediaInput> inputs, ParseContext context)
+ throws IOException, TikaException {
+ throw new TikaException("this engine does not embed audio or video");
+ }
}
diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/FfmpegSegmenter.java b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/FfmpegSegmenter.java
new file mode 100644
index 0000000000..f60bbc2994
--- /dev/null
+++ b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/FfmpegSegmenter.java
@@ -0,0 +1,156 @@
+/*
+ * 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.inference;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+
+import org.apache.tika.exception.TikaException;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.parser.inference.Modality;
+import org.apache.tika.utils.FileProcessResult;
+import org.apache.tika.utils.ProcessUtils;
+
+/**
+ * {@link MediaSegmenter} on ffmpeg and ffprobe from the PATH. Cut files are small on purpose:
+ * mono Opus for audio, silent downscaled H.264 for video, since they travel base64 in JSON.
+ */
+final class FfmpegSegmenter implements MediaSegmenter {
+
+ // per call; ProcessUtils clips both to what remains of the parse's total timeout
+ private static final long PROBE_TIMEOUT_MS = 60_000;
+ private static final long CUT_TIMEOUT_MS = 300_000;
+ // ffprobe prints every stream before the format's duration: room for thousands
+ private static final int PROBE_BUFFER = 1_000_000;
+ private static final int CUT_BUFFER = 20_000;
+
+ private static volatile Boolean available;
+
+ /** Whether ffmpeg and ffprobe answer on the PATH; probed once per JVM, outside any parse budget. */
+ @Override
+ public boolean available() {
+ Boolean known = available;
+ if (known == null) {
+ known = answers("ffmpeg") && answers("ffprobe");
+ available = known;
+ }
+ return known;
+ }
+
+ private static boolean answers(String tool) {
+ try {
+ FileProcessResult r = ProcessUtils.execute(
+ new ProcessBuilder(tool, "-version"), 10_000, 1000, 1000);
+ return r.getExitValue() == 0;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
+ @Override
+ public Probe probe(Path media, ParseContext context) throws IOException, TikaException {
+ ProcessBuilder pb = new ProcessBuilder("ffprobe", "-v", "error", "-show_entries",
+ "format=duration:stream=codec_type:stream_disposition=attached_pic",
+ "-of", "default=nw=1", media.toAbsolutePath().toString());
+ FileProcessResult r = ProcessUtils.execute(pb, context, PROBE_TIMEOUT_MS, PROBE_BUFFER,
+ CUT_BUFFER);
+ if (r.isTimeout()) {
+ throw new Timeout("ffprobe ran out of the parse's time budget after "
+ + r.getProcessTimeMillis() + " ms");
+ }
+ if (r.getExitValue() != 0) {
+ throw new TikaException("ffprobe failed: " + r.getStderr());
+ }
+ if (r.isStdoutTruncated()) {
+ throw new TikaException("ffprobe output over " + PROBE_BUFFER + " chars");
+ }
+ return parse(r.getStdout());
+ }
+
+ /** One stream at a time: a cover-art stream is not a picture channel, whichever stream it is. */
+ static Probe parse(String stdout) throws TikaException {
+ double seconds = -1;
+ boolean audio = false;
+ boolean video = false;
+ boolean streamIsVideo = false;
+ boolean streamIsCover = false;
+ for (String line : stdout.split("\\R")) {
+ if (line.startsWith("codec_type=")) {
+ video |= streamIsVideo && !streamIsCover;
+ streamIsVideo = line.equals("codec_type=video");
+ streamIsCover = false;
+ audio |= line.equals("codec_type=audio");
+ } else if (line.equals("DISPOSITION:attached_pic=1")) {
+ streamIsCover = true;
+ } else if (line.startsWith("duration=")) {
+ try {
+ seconds = Double.parseDouble(line.substring("duration=".length()));
+ } catch (NumberFormatException e) {
+ // "N/A" for streams without a container duration
+ }
+ }
+ }
+ video |= streamIsVideo && !streamIsCover;
+ if (seconds <= 0 || Double.isNaN(seconds) || Double.isInfinite(seconds)) {
+ throw new TikaException("ffprobe found no duration");
+ }
+ return new Probe(Math.round(seconds * 1000), audio, video);
+ }
+
+ @Override
+ public Path cut(Path media, Cell cell, Modality modality, Path dir, ParseContext context)
+ throws IOException, TikaException {
+ boolean audio = modality == Modality.AUDIO;
+ Path out = dir.resolve(cell.index() + (audio ? ".ogg" : ".mp4"));
+ if (Files.exists(out)) {
+ return out;
+ }
+ List<String> cmd = new ArrayList<>(List.of("ffmpeg", "-hide_banner", "-loglevel", "error",
+ "-y", "-ss", ms(cell.startMs()), "-t", ms(cell.endMs() - cell.startMs()),
+ "-i", media.toAbsolutePath().toString()));
+ if (audio) {
+ cmd.addAll(List.of("-vn", "-ac", "1", "-c:a", "libopus", "-b:a", "32k"));
+ } else {
+ cmd.addAll(List.of("-an", "-vf", "scale=-2:360", "-c:v", "libx264", "-preset",
+ "veryfast", "-crf", "28", "-movflags", "+faststart"));
+ }
+ cmd.add(out.toAbsolutePath().toString());
+ FileProcessResult r = ProcessUtils.execute(new ProcessBuilder(cmd), context,
+ CUT_TIMEOUT_MS, CUT_BUFFER, CUT_BUFFER);
+ if (r.isTimeout() || r.getExitValue() != 0 || !Files.exists(out)) {
+ // a partial file must not be taken for the cut on a retry
+ Files.deleteIfExists(out);
+ }
+ if (r.isTimeout()) {
+ throw new Timeout("ffmpeg ran out of the parse's time budget on segment "
+ + cell.index() + " after " + r.getProcessTimeMillis() + " ms");
+ }
+ if (r.getExitValue() != 0 || !Files.exists(out)) {
+ throw new TikaException("ffmpeg failed on segment " + cell.index() + ": "
+ + r.getStderr());
+ }
+ return out;
+ }
+
+ private static String ms(long millis) {
+ return String.format(Locale.ROOT, "%d.%03d", millis / 1000, millis % 1000);
+ }
+}
diff --git a/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MediaInput.java
similarity index 54%
copy from tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java
copy to tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MediaInput.java
index af1d6a1fd3..ca3de1da98 100644
--- a/tika-core/src/main/java/org/apache/tika/parser/inference/InputKind.java
+++ b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MediaInput.java
@@ -14,27 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.parser.inference;
+package org.apache.tika.inference;
-import org.apache.tika.mime.MediaType;
+import org.apache.tika.parser.inference.Modality;
-/** What a binding is fed: a document's text, its rendered pages, an image, or media bytes. */
-public enum InputKind {
- TEXT, PAGES, IMAGES, MEDIA;
-
- /** The kind a document's bytes are, by media type family; null for the rest. */
- public static InputKind of(MediaType type) {
- if (type == null) {
- return null;
- }
- switch (type.getType()) {
- case "image":
- return IMAGES;
- case "audio":
- case "video":
- return MEDIA;
- default:
- return null;
- }
- }
+/** One audio or video item for an engine: the bytes, their type, and which channel they are. */
+public record MediaInput(byte[] bytes, String mimeType, Modality modality) {
}
diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MediaSegmenter.java b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MediaSegmenter.java
new file mode 100644
index 0000000000..21cbe9b3ef
--- /dev/null
+++ b/tika-parsers/tika-parsers-ml/tika-inference/src/main/java/org/apache/tika/inference/MediaSegmenter.java
@@ -0,0 +1,94 @@
+/*
+ * 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.inference;
+
+import java.io.IOException;
+import java.nio.file.Path;
+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.MediaConfig;
+import org.apache.tika.parser.inference.Modality;
+
+/**
+ * Cuts audio and video into the cells of the document's {@link MediaConfig} grid. Every
+ * binding on a document sees the same cells, computed from the probed duration, so their
+ * vectors join on the cell.
+ */
+interface MediaSegmenter {
+
+ /** One cell of the grid, times in ms on the container's presentation timeline. */
+ record Cell(int index, long startMs, long endMs) {
+ }
+
+ /** What the probe found: length and which channels exist. */
+ record Probe(long durationMs, boolean hasAudio, boolean hasVideo) {
+ }
+
+ /** The parse's budget ran out under the tool; the caller stops cutting, it does not retry. */
+ final class Timeout extends TikaException {
+ Timeout(String message) {
+ super(message);
+ }
+ }
+
+ /** Whether the tool answers; a MEDIA binding stands by without it. */
+ boolean available();
+
+ Probe probe(Path media, ParseContext context) throws IOException, TikaException;
+
+ /**
+ * Writes one cell's audio or picture into {@code dir} and returns the file. Idempotent
+ * per cell: an existing cut is returned, not redone, so a retried request re-reads it.
+ */
+ Path cut(Path media, Cell cell, Modality modality, Path dir, ParseContext context)
+ throws IOException, TikaException;
+
+ /**
+ * The grid over a duration: fixed windows, each starting {@code seconds - overlap} after
+ * the last, at most {@code cap} of them. A tail inside the previous cell is not a cell.
+ */
+ static List<Cell> cells(long durationMs, MediaConfig config, int cap) throws TikaException {
+ long window = config.getSegment().getSeconds() * 1000L;
+ long stride = window - config.getSegment().getOverlap() * 1000L;
+ if (window <= 0 || stride <= 0 || stride > window) {
+ throw new TikaException("media.segment needs seconds > overlap >= 0");
+ }
+ List<Cell> cells = new ArrayList<>();
+ long start = 0;
+ long lastEnd = -1;
+ while (start < durationMs && lastEnd < durationMs && cells.size() < cap) {
+ long end = Math.min(start + window, durationMs);
+ cells.add(new Cell(cells.size(), start, end));
+ lastEnd = end;
+ start += stride;
+ }
+ return cells;
+ }
+
+ /** The cell's id on its document: the chunks every binding writes for it carry this. */
+ static String correlator(String idPath, Cell cell) {
+ return (idPath == null ? "" : idPath + ":") + "t:" + cell.startMs() + "-" + cell.endMs();
+ }
+
+ /** The media type of what {@link #cut} writes for a modality. */
+ static String mimeType(Modality modality) {
+ return modality == Modality.AUDIO ? "audio/ogg" : "video/mp4";
+ }
+}
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 d64485e4e2..31451b9118 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
@@ -37,6 +37,7 @@ import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.http.TikaHttpClient;
import org.apache.tika.parser.ParseContext;
+import org.apache.tika.parser.inference.Modality;
import org.apache.tika.utils.StringUtils;
/**
@@ -65,6 +66,9 @@ public class OpenAIEmbeddingEngine implements EmbeddingEngine, Initializable {
private String apiKeyPrefix = "Bearer ";
private Map<String, Object> requestParameters = new LinkedHashMap<>();
private String imageInput = IMAGE_INPUT_OBJECT;
+ private String mediaInput = MEDIA_INPUT_NONE;
+ public static final String MEDIA_INPUT_NONE = "none";
+ public static final String MEDIA_INPUT_OBJECT = "object";
public static final String IMAGE_INPUT_OBJECT = "object";
public static final String IMAGE_INPUT_DATA_URI = "data-uri";
@@ -88,6 +92,10 @@ public class OpenAIEmbeddingEngine implements EmbeddingEngine, Initializable {
throw new TikaConfigException("imageInput must be \"" + IMAGE_INPUT_OBJECT + "\" or \""
+ IMAGE_INPUT_DATA_URI + "\", not \"" + imageInput + "\"");
}
+ if (!MEDIA_INPUT_NONE.equals(mediaInput) && !MEDIA_INPUT_OBJECT.equals(mediaInput)) {
+ throw new TikaConfigException("mediaInput must be \"" + MEDIA_INPUT_NONE + "\" or \""
+ + MEDIA_INPUT_OBJECT + "\", not \"" + mediaInput + "\"");
+ }
for (String key : requestParameters.keySet()) {
if (ENGINE_OWNED.contains(key)) {
throw new TikaConfigException("requestParameters may not set \"" + key
@@ -119,6 +127,29 @@ public class OpenAIEmbeddingEngine implements EmbeddingEngine, Initializable {
return post(root, images.size(), context);
}
+ @Override
+ public boolean supportsMedia() {
+ return MEDIA_INPUT_OBJECT.equals(mediaInput);
+ }
+
+ /** Object form only: {@code {"audio": data-uri}} or {@code {"video": data-uri}}, as Jina names them. */
+ @Override
+ public List<float[]> embedMedia(List<MediaInput> inputs, ParseContext context)
+ throws IOException, TikaException {
+ if (!supportsMedia()) {
+ return EmbeddingEngine.super.embedMedia(inputs, context);
+ }
+ ObjectNode root = request();
+ ArrayNode input = root.putArray("input");
+ for (MediaInput item : inputs) {
+ String key = item.modality() == Modality.AUDIO ? "audio"
+ : item.mimeType().startsWith("image/") ? "image" : "video";
+ input.addObject().put(key, "data:" + item.mimeType() + ";base64,"
+ + Base64.getEncoder().encodeToString(item.bytes()));
+ }
+ return post(root, inputs.size(), context);
+ }
+
@Override
public List<float[]> embedTexts(List<String> texts, ParseContext context)
throws IOException, TikaException {
@@ -270,6 +301,15 @@ public class OpenAIEmbeddingEngine implements EmbeddingEngine, Initializable {
this.imageInput = imageInput;
}
+ /** Whether audio and video may be sent, and how: {@code "none"} or {@code "object"} ({@code {"audio": ...}}, {@code {"video": ...}}). */
+ public String getMediaInput() {
+ return mediaInput;
+ }
+
+ public void setMediaInput(String mediaInput) {
+ this.mediaInput = mediaInput;
+ }
+
/** Extra keys for every request body, as the vendor names them; see the class note. */
public Map<String, Object> getRequestParameters() {
return requestParameters;
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 07381f6949..763fe9bd5e 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
@@ -20,6 +20,7 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
@@ -31,6 +32,7 @@ import org.apache.tika.inference.locator.PaginatedLocator;
import org.apache.tika.inference.locator.SpatialLocator;
import org.apache.tika.inference.locator.TemporalLocator;
import org.apache.tika.inference.locator.TextLocator;
+import org.apache.tika.parser.inference.Modality;
public class ChunkSerializerTest {
@@ -209,4 +211,27 @@ public class ChunkSerializerTest {
assertEquals("/4", restored.get(1).getLocators().getEmbedded().get(0).getIdPath());
assertNull(restored.get(1).getLocators().getEmbedded().get(0).getName());
}
+
+ @Test
+ void testModalityAndCorrelatorRoundTrip() throws Exception {
+ Chunk c = new Chunk(null, new Locators().addTemporal(new TemporalLocator(0, 30000)));
+ c.setVector(new float[]{1f});
+ c.setProducer("jina-audio");
+ c.setModality(Modality.AUDIO);
+ c.setCorrelator("t:0-30000");
+ String json = ChunkSerializer.toJson(List.of(c));
+ assertTrue(json.contains("\"modality\":\"audio\""), json);
+ Chunk back = ChunkSerializer.fromJson(json).get(0);
+ assertEquals(Modality.AUDIO, back.getModality());
+ assertEquals("t:0-30000", back.getCorrelator());
+ Chunk plain = ChunkSerializer.fromJson(ChunkSerializer.toJson(
+ List.of(new Chunk("t", 0, 1)))).get(0);
+ assertNull(plain.getModality());
+ assertNull(plain.getCorrelator());
+ // a newer writer's value reads as unsaid, and the rest of the array still reads
+ List<Chunk> foreign = ChunkSerializer.fromJson(
+ "[{\"modality\":\"smell\",\"text\":\"a\"},{\"modality\":\"Audio\",\"text\":\"b\"}]");
+ assertNull(foreign.get(0).getModality());
+ assertEquals(Modality.AUDIO, foreign.get(1).getModality());
+ }
}
diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskMediaContractTest.java b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskMediaContractTest.java
new file mode 100644
index 0000000000..2b9ecf8723
--- /dev/null
+++ b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskMediaContractTest.java
@@ -0,0 +1,270 @@
+/*
+ * 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.inference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.exception.TikaException;
+import org.apache.tika.http.TikaTestHttpServer;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+import org.apache.tika.mime.MediaType;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.parser.inference.InferenceBinding;
+import org.apache.tika.parser.inference.InferenceUnit;
+import org.apache.tika.parser.inference.InputKind;
+import org.apache.tika.parser.inference.Modality;
+
+/**
+ * The media path's contract with a fake segmenter, so no ffmpeg: what is written, kept and
+ * released on every exit of {@code EmbedTask.embedMedia}.
+ */
+public class EmbedTaskMediaContractTest {
+
+ /** Probes every file as 70 s of audio and video; cuts are one-byte files. */
+ static final class FakeSegmenter implements MediaSegmenter {
+ final boolean available;
+ final List<Path> cutDirs = new ArrayList<>();
+ int cuts;
+ Path failProbe;
+ int timeoutOnCut = -1;
+ int ioOnCut = -1;
+ long durationMs = 70_000;
+
+ FakeSegmenter(boolean available) {
+ this.available = available;
+ }
+
+ @Override
+ public boolean available() {
+ return available;
+ }
+
+ @Override
+ public Probe probe(Path media, ParseContext context) throws TikaException {
+ if (media.equals(failProbe)) {
+ throw new TikaException("ffprobe failed: not a media file");
+ }
+ return new Probe(durationMs, true, true);
+ }
+
+ @Override
+ public Path cut(Path media, Cell cell, Modality modality, Path dir, ParseContext context)
+ throws IOException, TikaException {
+ if (!cutDirs.contains(dir)) {
+ cutDirs.add(dir);
+ }
+ Path out = dir.resolve(cell.index() + ".bin");
+ if (Files.exists(out)) {
+ return out;
+ }
+ if (cell.index() == timeoutOnCut) {
+ throw new Timeout("budget gone on " + cell.index());
+ }
+ if (cell.index() == ioOnCut) {
+ throw new IOException("disk gone on " + cell.index());
+ }
+ cuts++;
+ return Files.write(out, new byte[]{(byte) cell.index()});
+ }
+ }
+
+ private TikaTestHttpServer server;
+ private OpenAIEmbeddingEngine engine;
+ private FakeSegmenter fake;
+ private MediaSegmenter real;
+
+ @TempDir
+ Path tmp;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ server = new TikaTestHttpServer();
+ engine = new OpenAIEmbeddingEngine();
+ engine.setBaseUrl(server.url());
+ engine.setModel("m");
+ engine.setMediaInput(OpenAIEmbeddingEngine.MEDIA_INPUT_OBJECT);
+ engine.setMaxBatchSize(2);
+ engine.initialize();
+ fake = new FakeSegmenter(true);
+ real = EmbedTask.segmenter;
+ EmbedTask.segmenter = fake;
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ EmbedTask.segmenter = real;
+ engine.close();
+ server.shutdown();
+ }
+
+ private static String response(int n) {
+ StringBuilder sb = new StringBuilder("{\"data\":[");
+ for (int i = 0; i < n; i++) {
+ sb.append(i > 0 ? "," : "").append("{\"index\":").append(i)
+ .append(",\"embedding\":[").append(i).append(".0]}");
+ }
+ return sb.append("]}").toString();
+ }
+
+ private static InferenceBinding binding() {
+ return InferenceBinding.builder("v", "jina", InputKind.MEDIA).modality(Modality.VISUAL).build();
+ }
+
+ private InferenceUnit unit(String name) throws Exception {
+ Path file = Files.write(tmp.resolve(name), new byte[]{1});
+ return new InferenceUnit(InputKind.MEDIA, MediaType.video("mp4"), new Metadata(), null, file);
+ }
+
+ private static List<Chunk> chunks(InferenceUnit unit) throws Exception {
+ String json = unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS);
+ return json == null ? List.of() : ChunkSerializer.fromJson(json);
+ }
+
+ private void assertReleased() {
+ assertFalse(fake.cutDirs.isEmpty());
+ for (Path dir : fake.cutDirs) {
+ assertFalse(Files.exists(dir), "temp dir released: " + dir);
+ }
+ }
+
+ /** 70 s at 30/5 is three cells: [0,30] [25,55] [50,70]; batch 2 means requests of 2 and 1. */
+ @Test
+ public void testHappyPathReleasesTheDir() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(2)));
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ InferenceUnit unit = unit("a.mp4");
+ new EmbedTask().run(binding(), List.of(unit), engine, new ParseContext());
+ assertEquals(3, chunks(unit).size());
+ assertEquals(3, fake.cuts);
+ assertReleased();
+ }
+
+ /** A rejected request is retried one cell at a time from the same cut files. */
+ @Test
+ public void testRejectedBatchRetriesSinglyWithoutRecutting() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(400, "{\"error\":\"too big\"}"));
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ server.enqueue(new TikaTestHttpServer.MockResponse(400, "{\"error\":\"bad cell\"}"));
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ InferenceUnit unit = unit("a.mp4");
+ TikaException e = assertThrows(TikaException.class,
+ () -> new EmbedTask().run(binding(), List.of(unit), engine, new ParseContext()));
+ assertTrue(e.getMessage().contains("bad cell"), e.getMessage());
+ List<Chunk> chunks = chunks(unit);
+ assertEquals(2, chunks.size(), "the good cells keep their vectors");
+ assertEquals(0, chunks.get(0).getLocators().getTemporal().get(0).getStartMs());
+ assertEquals(50_000, chunks.get(1).getLocators().getTemporal().get(0).getStartMs());
+ assertEquals(3, fake.cuts, "no cell is cut twice");
+ assertEquals(4, server.getRequestCount());
+ assertReleased();
+ }
+
+ /** The budget runs out under the tool on cell 2: cells 0-1 are written, nothing more is tried. */
+ @Test
+ public void testTimeoutKeepsWhatWasEmbedded() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(2)));
+ fake.timeoutOnCut = 2;
+ InferenceUnit first = unit("a.mp4");
+ InferenceUnit second = unit("b.mp4");
+ MediaSegmenter.Timeout t = assertThrows(MediaSegmenter.Timeout.class,
+ () -> new EmbedTask().run(binding(), List.of(first, second), engine, new ParseContext()));
+ assertTrue(t.getMessage().contains("budget"), t.getMessage());
+ assertEquals(2, chunks(first).size(), "paid for, so written");
+ assertNull(second.getTarget().get(TikaCoreProperties.TIKA_CHUNKS), "no later unit runs");
+ assertEquals(1, server.getRequestCount(), "no single retries after a timeout");
+ assertReleased();
+ }
+
+ /** An unreadable file costs only itself; the units after it still run. */
+ @Test
+ public void testProbeFailureSkipsOnlyThatUnit() throws Exception {
+ for (int i = 0; i < 2; i++) {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(2)));
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ }
+ InferenceUnit good = unit("a.mp4");
+ InferenceUnit bad = unit("b.mp4");
+ InferenceUnit alsoGood = unit("c.mp4");
+ fake.failProbe = bad.getPath();
+ TikaException e = assertThrows(TikaException.class, () -> new EmbedTask().run(binding(),
+ List.of(good, bad, alsoGood), engine, new ParseContext()));
+ assertTrue(e.getMessage().contains("ffprobe failed"), e.getMessage());
+ assertEquals(3, chunks(good).size());
+ assertEquals(3, chunks(alsoGood).size());
+ assertNull(bad.getTarget().get(TikaCoreProperties.TIKA_CHUNKS));
+ assertReleased();
+ }
+
+ /** Infrastructure failure mid-unit: the batch already embedded is written, then it propagates. */
+ @Test
+ public void testIOExceptionWritesWhatWasEmbedded() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(2)));
+ fake.ioOnCut = 2;
+ InferenceUnit unit = unit("a.mp4");
+ IOException e = assertThrows(IOException.class,
+ () -> new EmbedTask().run(binding(), List.of(unit), engine, new ParseContext()));
+ assertTrue(e.getMessage().contains("disk gone"));
+ assertEquals(2, chunks(unit).size());
+ assertReleased();
+ }
+
+ /** A cut over maxBytes is deleted at once and fails only its own cell. */
+ @Test
+ public void testOverMaxBytesCellIsDroppedAndDeleted() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ InferenceBinding tiny = InferenceBinding.builder("v", "jina", InputKind.MEDIA)
+ .modality(Modality.VISUAL).maxBytes(0).maxChunks(1).build();
+ InferenceUnit unit = unit("a.mp4");
+ TikaException e = assertThrows(TikaException.class,
+ () -> new EmbedTask().run(tiny, List.of(unit), engine, new ParseContext()));
+ assertTrue(e.getMessage().contains("over maxBytes"), e.getMessage());
+ assertEquals(0, server.getRequestCount());
+ assertNull(unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS));
+ assertReleased();
+ }
+
+ /** maxChunks on the binding and maxSegments in the block both cap cells; the smaller wins. */
+ @Test
+ public void testCapsBoundACLaimedDuration() throws Exception {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(2)));
+ fake.durationMs = Long.MAX_VALUE / 4;
+ InferenceBinding two = InferenceBinding.builder("v", "jina", InputKind.MEDIA)
+ .modality(Modality.VISUAL).maxChunks(2).build();
+ InferenceUnit unit = unit("a.mp4");
+ new EmbedTask().run(two, List.of(unit), engine, new ParseContext());
+ assertEquals(2, chunks(unit).size());
+ assertEquals(2, fake.cuts);
+ assertSame(fake, EmbedTask.segmenter);
+ }
+}
diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskMediaTest.java b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskMediaTest.java
new file mode 100644
index 0000000000..f51285419f
--- /dev/null
+++ b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/EmbedTaskMediaTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.inference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.config.ParseTimeout;
+import org.apache.tika.config.TimeoutLimits;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.http.TikaTestHttpServer;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+import org.apache.tika.mime.MediaType;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.parser.inference.InferenceBinding;
+import org.apache.tika.parser.inference.InferenceUnit;
+import org.apache.tika.parser.inference.InputKind;
+import org.apache.tika.parser.inference.MediaConfig;
+import org.apache.tika.parser.inference.Modality;
+import org.apache.tika.utils.ProcessUtils;
+
+/** Needs ffmpeg on the PATH; the clip is synthesized, so nothing is checked in. The contract without ffmpeg: {@link EmbedTaskMediaContractTest}. */
+public class EmbedTaskMediaTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private TikaTestHttpServer server;
+ private OpenAIEmbeddingEngine engine;
+
+ @TempDir
+ Path tmp;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ assumeTrue(new FfmpegSegmenter().available(), "ffmpeg not on the PATH");
+ server = new TikaTestHttpServer();
+ engine = new OpenAIEmbeddingEngine();
+ engine.setBaseUrl(server.url());
+ engine.setModel("jina-embeddings-v5-omni-small");
+ engine.setMediaInput(OpenAIEmbeddingEngine.MEDIA_INPUT_OBJECT);
+ engine.setRequestParameters(Map.of("task", "retrieval.passage"));
+ engine.setMaxBatchSize(2);
+ engine.initialize();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (engine != null) {
+ engine.close();
+ }
+ if (server != null) {
+ server.shutdown();
+ }
+ }
+
+ /** {@code seconds} of test pattern plus a tone; audio only when {@code video} is false. */
+ private Path clip(int seconds, boolean video) throws Exception {
+ Path out = tmp.resolve((video ? "clip" : "tone") + seconds + (video ? ".mp4" : ".m4a"));
+ ProcessBuilder pb = video
+ ? new ProcessBuilder("ffmpeg", "-loglevel", "error", "-y",
+ "-f", "lavfi", "-i", "testsrc=duration=" + seconds + ":size=160x120:rate=5",
+ "-f", "lavfi", "-i", "sine=frequency=440:duration=" + seconds,
+ "-c:v", "libx264", "-preset", "ultrafast", "-c:a", "aac", "-shortest",
+ out.toString())
+ : new ProcessBuilder("ffmpeg", "-loglevel", "error", "-y",
+ "-f", "lavfi", "-i", "sine=frequency=440:duration=" + seconds,
+ "-c:a", "aac", out.toString());
+ assertEquals(0, ProcessUtils.execute(pb, 60_000, 10_000, 10_000).getExitValue());
+ return out;
+ }
+
+ private static String response(int n) {
+ StringBuilder sb = new StringBuilder("{\"data\":[");
+ for (int i = 0; i < n; i++) {
+ int index = n - 1 - i;
+ sb.append(i > 0 ? "," : "").append("{\"index\":").append(index)
+ .append(",\"embedding\":[").append(index).append(".0,0.5]}");
+ }
+ return sb.append("]}").toString();
+ }
+
+ private static InferenceBinding binding(String id, Modality modality) {
+ return InferenceBinding.builder(id, "jina", InputKind.MEDIA).modality(modality).build();
+ }
+
+ private static InferenceUnit unit(Path file, String type) throws Exception {
+ return new InferenceUnit(InputKind.MEDIA, MediaType.parse(type), new Metadata(), null, file);
+ }
+
+ /** Two bindings on the two channels write two chunks per cell that share the cell's id. */
+ @Test
+ public void testAudioAndVideoJoinOnTheCell() throws Exception {
+ InferenceUnit unit = unit(clip(70, true), "video/mp4");
+ // 70 s at 30/5: [0,30] [25,55] [50,70]; batch size 2 -> requests of 2 and 1 per channel
+ for (int i = 0; i < 2; i++) {
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(2)));
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ }
+ EmbedTask task = new EmbedTask();
+ InferenceBinding video = binding("jina-video", Modality.VISUAL);
+ InferenceBinding audio = binding("jina-audio", Modality.AUDIO);
+ task.validate(video, engine);
+ task.run(video, List.of(unit), engine, new ParseContext());
+ task.run(audio, List.of(unit), engine, new ParseContext());
+
+ assertEquals(4, server.getRequestCount());
+ JsonNode first = MAPPER.readTree(server.takeRequest().body());
+ assertEquals("retrieval.passage", first.get("task").asText());
+ assertEquals("jina-embeddings-v5-omni-small", first.get("model").asText());
+ assertEquals(2, first.get("input").size());
+ assertTrue(first.get("input").get(0).get("video").asText().startsWith("data:video/mp4;base64,"));
+ server.takeRequest();
+ JsonNode third = MAPPER.readTree(server.takeRequest().body());
+ assertTrue(third.get("input").get(0).get("audio").asText().startsWith("data:audio/ogg;base64,"));
+
+ List<Chunk> chunks = ChunkSerializer.fromJson(
+ unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS));
+ assertEquals(6, chunks.size(), "one chunk per cell per channel");
+ Chunk video2 = chunks.get(1);
+ Chunk audio2 = chunks.get(4);
+ assertEquals(25000, video2.getLocators().getTemporal().get(0).getStartMs());
+ assertEquals(55000, video2.getLocators().getTemporal().get(0).getEndMs());
+ assertEquals(70000, chunks.get(2).getLocators().getTemporal().get(0).getEndMs());
+ assertEquals("t:25000-55000", video2.getCorrelator());
+ assertEquals(video2.getCorrelator(), audio2.getCorrelator(), "the cell's id groups them");
+ assertEquals("jina-video", video2.getProducer());
+ assertEquals(Modality.VISUAL, video2.getModality());
+ assertEquals(Modality.AUDIO, audio2.getModality());
+ assertEquals(1.0f, video2.getVector()[0], "placed by index");
+ assertEquals(0.0f, chunks.get(5).getVector()[0], "second request, index 0");
+ }
+
+ /** The request's media block sets the grid; maxSegments cuts the tail. */
+ @Test
+ public void testMediaBlockSetsTheGrid() throws Exception {
+ InferenceUnit unit = unit(clip(70, true), "video/mp4");
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ MediaConfig config = new MediaConfig();
+ config.getSegment().setSeconds(60);
+ config.getSegment().setOverlap(0);
+ config.setMaxSegments(1);
+ ParseContext context = new ParseContext();
+ context.set(MediaConfig.class, config);
+ new EmbedTask().run(binding("v", Modality.VISUAL), List.of(unit), engine, context);
+ List<Chunk> chunks = ChunkSerializer.fromJson(
+ unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS));
+ assertEquals(1, chunks.size());
+ assertEquals(60000, chunks.get(0).getLocators().getTemporal().get(0).getEndMs());
+ assertEquals(1, server.getRequestCount());
+ }
+
+ /** A file without the channel is skipped, not failed. */
+ @Test
+ public void testAudioOnlyFileSkipsTheVisualBinding() throws Exception {
+ InferenceUnit unit = unit(clip(10, false), "audio/mp4");
+ server.enqueue(new TikaTestHttpServer.MockResponse(200, response(1)));
+ new EmbedTask().run(binding("v", Modality.VISUAL), List.of(unit), engine, new ParseContext());
+ assertEquals(0, server.getRequestCount());
+ assertNull(unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS));
+ new EmbedTask().run(binding("a", Modality.AUDIO), List.of(unit), engine, new ParseContext());
+ assertEquals(1, server.getRequestCount());
+ assertEquals(1, ChunkSerializer.fromJson(
+ unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS)).size());
+ }
+
+ /** Without mediaInput the engine refuses MEDIA at load. */
+ @Test
+ public void testValidateNeedsMediaInput() throws Exception {
+ OpenAIEmbeddingEngine plain = new OpenAIEmbeddingEngine();
+ plain.initialize();
+ assertThrows(TikaConfigException.class,
+ () -> new EmbedTask().validate(binding("a", Modality.AUDIO), plain));
+ new EmbedTask().validate(binding("a", Modality.AUDIO), engine);
+ plain.setMediaInput("blob");
+ assertThrows(TikaConfigException.class, plain::initialize);
+ }
+
+
+ /** An exhausted parse budget stops the task at the first cut instead of failing every cell. */
+ @Test
+ public void testExhaustedBudgetStopsCutting() throws Exception {
+ InferenceUnit unit = unit(clip(70, true), "video/mp4");
+ ParseContext context = new ParseContext();
+ context.set(ParseTimeout.class, ParseTimeout.start(new TimeoutLimits(1, 1000)));
+ Thread.sleep(5);
+ MediaSegmenter.Timeout t = assertThrows(MediaSegmenter.Timeout.class,
+ () -> new EmbedTask().run(binding("v", Modality.VISUAL), List.of(unit), engine, context));
+ assertTrue(t.getMessage().contains("time budget"), t.getMessage());
+ assertEquals(0, server.getRequestCount());
+ assertNull(unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS));
+ }
+
+ /** Without ffmpeg a document is skipped with a warning, not failed. */
+ @Test
+ public void testMissingFfmpegSkipsWithAWarning() throws Exception {
+ InferenceUnit unit = unit(clip(10, true), "video/mp4");
+ MediaSegmenter real = EmbedTask.segmenter;
+ EmbedTask.segmenter = new EmbedTaskMediaContractTest.FakeSegmenter(false);
+ try {
+ new EmbedTask().validate(binding("v", Modality.VISUAL), engine);
+ new EmbedTask().run(binding("v", Modality.VISUAL), List.of(unit), engine, new ParseContext());
+ } finally {
+ EmbedTask.segmenter = real;
+ }
+ assertEquals(0, server.getRequestCount());
+ assertNull(unit.getTarget().get(TikaCoreProperties.TIKA_CHUNKS));
+ assertTrue(unit.getTarget().get(TikaCoreProperties.TIKA_META_EXCEPTION_WARNING)
+ .contains("ffmpeg"));
+ }
+
+}
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 7759a4d385..72143330d2 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
@@ -19,6 +19,7 @@ package org.apache.tika.inference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
@@ -94,8 +95,7 @@ public class EmbedTaskTest {
}
private static InferenceBinding binding() {
- return new InferenceBinding("pics", "clip", InputKind.IMAGES, List.of("embed"), null,
- null, -1, -1, true);
+ return InferenceBinding.builder("pics", "clip", InputKind.IMAGES).build();
}
@Test
@@ -146,8 +146,7 @@ public class EmbedTaskTest {
Path p2 = Files.createTempFile(tmp, "p2", ".png");
Files.write(p1, "one".getBytes(StandardCharsets.UTF_8));
Files.write(p2, "two".getBytes(StandardCharsets.UTF_8));
- InferenceBinding pages = new InferenceBinding("pages", "clip", InputKind.PAGES,
- List.of("embed"), null, null, -1, -1, true);
+ InferenceBinding pages = InferenceBinding.builder("pages", "clip", InputKind.PAGES).build();
EmbedTask task = new EmbedTask();
task.validate(pages, engine);
task.run(pages, List.of(
@@ -171,8 +170,8 @@ public class EmbedTaskTest {
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));
+ InferenceBinding text = InferenceBinding.builder("text-vectors", "clip", InputKind.TEXT)
+ .chunker(new MarkdownChunker(10, 0)).build();
EmbedTask task = new EmbedTask();
task.validate(text, engine);
// "one two" and "three" from the parent, "four five" and "six" from the attachment
@@ -205,8 +204,8 @@ public class EmbedTaskTest {
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);
+ InferenceBinding text = InferenceBinding.builder("t", "clip", InputKind.TEXT)
+ .maxChunks(1).build();
new EmbedTask().run(text, List.of(
new InferenceUnit(MediaType.TEXT_PLAIN, a, null, "whole document"),
new InferenceUnit(MediaType.TEXT_PLAIN, b, null, "dropped")),
@@ -221,10 +220,10 @@ public class EmbedTaskTest {
public void testValidateRejectsANonEmbeddingEngine() {
assertThrows(TikaConfigException.class,
() -> new EmbedTask().validate(binding(), new Engine() { }));
- InferenceBinding media = new InferenceBinding("clips", "clip", InputKind.MEDIA,
- List.of("embed"), null, null, -1, -1, true);
- assertThrows(TikaConfigException.class, () -> new EmbedTask().validate(media, engine),
- "embed takes IMAGES only");
+ InferenceBinding media = InferenceBinding.builder("clips", "clip", InputKind.MEDIA).build();
+ TikaConfigException e = assertThrows(TikaConfigException.class,
+ () -> new EmbedTask().validate(media, engine));
+ assertTrue(e.getMessage().contains("does not embed audio or video"), e.getMessage());
}
@Test
diff --git a/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/FfmpegSegmenterTest.java b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/FfmpegSegmenterTest.java
new file mode 100644
index 0000000000..a3f5a552a3
--- /dev/null
+++ b/tika-parsers/tika-parsers-ml/tika-inference/src/test/java/org/apache/tika/inference/FfmpegSegmenterTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.inference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.nio.file.Path;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.exception.TikaException;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.parser.inference.MediaConfig;
+import org.apache.tika.parser.inference.Modality;
+import org.apache.tika.utils.ProcessUtils;
+
+/** The grid and the probe parser need no ffmpeg; the last test synthesizes a tagged mp3. */
+public class FfmpegSegmenterTest {
+
+ @TempDir
+ Path tmp;
+
+ @Test
+ public void testCells() throws Exception {
+ MediaConfig config = new MediaConfig();
+ List<MediaSegmenter.Cell> cells = MediaSegmenter.cells(52_000, config, Integer.MAX_VALUE);
+ assertEquals(2, cells.size(), "a tail inside the previous cell's overlap is not a cell");
+ assertEquals(25_000, cells.get(1).startMs());
+ assertEquals(52_000, cells.get(1).endMs());
+ assertEquals(1, MediaSegmenter.cells(30_000, config, Integer.MAX_VALUE).size());
+ assertEquals(1, MediaSegmenter.cells(1_000, config, Integer.MAX_VALUE).size());
+ assertEquals(0, MediaSegmenter.cells(0, config, Integer.MAX_VALUE).size());
+ // a container may claim any duration: the cap bounds the work, not the claim
+ assertEquals(3, MediaSegmenter.cells(Long.MAX_VALUE / 2, config, 3).size());
+ config.getSegment().setOverlap(-5);
+ assertThrows(TikaException.class, () -> MediaSegmenter.cells(52_000, config, 10),
+ "a negative overlap would leave gaps");
+ config.getSegment().setOverlap(30);
+ assertThrows(TikaException.class, () -> MediaSegmenter.cells(52_000, config, 10));
+ }
+
+ @Test
+ public void testCorrelatorAndMime() {
+ MediaSegmenter.Cell cell = new MediaSegmenter.Cell(0, 25_000, 55_000);
+ assertEquals("t:25000-55000", MediaSegmenter.correlator(null, cell));
+ assertEquals("/1:t:25000-55000", MediaSegmenter.correlator("/1", cell),
+ "an embedded document's cells are its own");
+ assertEquals("audio/ogg", MediaSegmenter.mimeType(Modality.AUDIO));
+ assertEquals("video/mp4", MediaSegmenter.mimeType(Modality.VISUAL));
+ }
+
+ /** Cover art is a one-frame video stream; it is not a picture channel, wherever it sits. */
+ @Test
+ public void testProbeParse() throws Exception {
+ MediaSegmenter.Probe p = FfmpegSegmenter.parse(
+ "codec_type=audio\nDISPOSITION:attached_pic=0\n"
+ + "codec_type=video\nDISPOSITION:attached_pic=1\nduration=5.5\n");
+ assertTrue(p.hasAudio());
+ assertFalse(p.hasVideo());
+ assertEquals(5500, p.durationMs());
+ p = FfmpegSegmenter.parse(
+ "codec_type=video\nDISPOSITION:attached_pic=0\n"
+ + "codec_type=audio\nDISPOSITION:attached_pic=0\n"
+ + "codec_type=video\nDISPOSITION:attached_pic=1\nduration=70\n");
+ assertTrue(p.hasVideo(), "the film is still a film with a poster attached");
+ assertTrue(p.hasAudio());
+ p = FfmpegSegmenter.parse("codec_type=video\nDISPOSITION:attached_pic=0\nduration=1\n");
+ assertTrue(p.hasVideo());
+ assertFalse(p.hasAudio());
+ assertThrows(TikaException.class, () -> FfmpegSegmenter.parse("codec_type=audio\nduration=N/A\n"));
+ assertThrows(TikaException.class, () -> FfmpegSegmenter.parse("codec_type=audio\n"));
+ }
+
+ @Test
+ public void testCoverArtIsNotVideo() throws Exception {
+ FfmpegSegmenter segmenter = new FfmpegSegmenter();
+ assumeTrue(segmenter.available(), "ffmpeg not on the PATH");
+ Path plain = tmp.resolve("plain.mp3");
+ Path cover = tmp.resolve("cover.png");
+ Path tagged = tmp.resolve("cover.mp3");
+ assertEquals(0, ProcessUtils.execute(new ProcessBuilder("ffmpeg", "-loglevel", "error",
+ "-y", "-f", "lavfi", "-i", "sine=frequency=440:duration=5", "-c:a", "libmp3lame",
+ plain.toString()), 60_000, 10_000, 10_000).getExitValue());
+ assertEquals(0, ProcessUtils.execute(new ProcessBuilder("ffmpeg", "-loglevel", "error",
+ "-y", "-f", "lavfi", "-i", "testsrc=duration=1:size=64x64:rate=1", "-frames:v", "1",
+ cover.toString()), 60_000, 10_000, 10_000).getExitValue());
+ assertEquals(0, ProcessUtils.execute(new ProcessBuilder("ffmpeg", "-loglevel", "error",
+ "-y", "-i", plain.toString(), "-i", cover.toString(), "-map", "0:a", "-map", "1:v",
+ "-c:a", "copy", "-c:v", "mjpeg", "-disposition:v", "attached_pic",
+ tagged.toString()), 60_000, 10_000, 10_000).getExitValue());
+ MediaSegmenter.Probe probe = segmenter.probe(tagged, new ParseContext());
+ assertTrue(probe.hasAudio());
+ assertFalse(probe.hasVideo());
+ assertTrue(probe.durationMs() >= 5000);
+ }
+}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESClient.java b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESClient.java
index 6369722f35..4652fa7a7e 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESClient.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESClient.java
@@ -21,8 +21,14 @@ import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringWriter;
+import java.nio.ByteBuffer;
+import java.nio.FloatBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -31,6 +37,8 @@ import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.CloseableHttpResponse;
@@ -111,6 +119,8 @@ public class ESClient {
ESEmitterConfig.AttachmentStrategy.PARENT_CHILD)
? emitKey : null;
+ boolean chunkDocuments =
+ config.chunkStrategy() == ESEmitterConfig.ChunkStrategy.DOCUMENTS;
for (Metadata metadata : metadataList) {
StringBuilder id = new StringBuilder(emitKey);
if (i > 0) {
@@ -118,19 +128,151 @@ public class ESClient {
}
String indexJson = metadataToJsonWriter.getBulkJson(id.toString(), routing);
json.append(indexJson).append("\n");
+ JsonNode chunks = chunkDocuments ? chunkArray(metadata.get(CHUNKS_FIELD)) : null;
+ Metadata body = chunks == null ? metadata : without(metadata, CHUNKS_FIELD);
if (i == 0) {
json.append(metadataToJsonWriter.writeContainer(
- metadata, config.attachmentStrategy()));
+ body, config.attachmentStrategy()));
} else {
json.append(metadataToJsonWriter.writeEmbedded(
- metadata, config.attachmentStrategy(), emitKey,
+ body, config.attachmentStrategy(), emitKey,
config.embeddedFileFieldName()));
}
json.append("\n");
+ if (chunks != null) {
+ appendChunks(id.toString(), emitKey, routing, chunks, json);
+ }
i++;
}
}
+ static final String CHUNKS_FIELD = "tk:chunks";
+ private static final ObjectMapper CHUNK_MAPPER = new ObjectMapper();
+
+ /**
+ * Chunks as their own documents, one per unit: chunks sharing a correlator (a segment's
+ * sound and picture, a region's text and crop) merge into one document whose vectors sit
+ * under {@code v.<producer>}, so a search scores the unit and two vectors of one unit add
+ * up on the same hit. Locators are flattened to fields and vectors decoded to float arrays.
+ */
+ /** The chunk array, or null when the value is not one and so stays on the document. */
+ private static JsonNode chunkArray(String chunks) throws IOException {
+ if (chunks == null || !isValidJson(chunks)) {
+ return null;
+ }
+ JsonNode array = CHUNK_MAPPER.readTree(chunks);
+ return array.isArray() ? array : null;
+ }
+
+ private void appendChunks(String docId, String emitKey, String routing, JsonNode array,
+ StringBuilder json) throws IOException {
+ int n = 0;
+ for (ObjectNode doc : chunkDocuments(array, docId, emitKey)) {
+ json.append(metadataToJsonWriter.getBulkJson(docId + "-chunk-" + n, routing))
+ .append("\n");
+ json.append(metadataToJsonWriter.writeChunk(doc)).append("\n");
+ n++;
+ }
+ }
+
+ /** One document per correlator, in first-seen order; a chunk without one is its own. */
+ static List<ObjectNode> chunkDocuments(JsonNode chunks, String docId, String emitKey) {
+ List<ObjectNode> docs = new ArrayList<>();
+ Map<String, ObjectNode> byCorrelator = new HashMap<>();
+ for (JsonNode chunk : chunks) {
+ String correlator = chunk.hasNonNull("correlator") ? chunk.get("correlator").asText() : null;
+ ObjectNode doc = correlator == null ? null : byCorrelator.get(correlator);
+ if (doc == null) {
+ doc = CHUNK_MAPPER.createObjectNode();
+ doc.put("file_id", docId);
+ doc.put("container_id", emitKey);
+ if (correlator != null) {
+ doc.put("correlator", correlator);
+ byCorrelator.put(correlator, doc);
+ }
+ docs.add(doc);
+ }
+ mergeChunk(doc, chunk);
+ }
+ for (ObjectNode doc : docs) {
+ if (doc.get("v").isEmpty()) {
+ doc.remove("v");
+ }
+ }
+ return docs;
+ }
+
+ /** Adds what the chunk knows: fields it is the first to set, and its vector under its producer. */
+ private static void mergeChunk(ObjectNode doc, JsonNode chunk) {
+ if (chunk.hasNonNull("text") && !doc.has("text")) {
+ doc.put("text", chunk.get("text").asText());
+ }
+ JsonNode locators = chunk.get("locators");
+ if (locators != null && locators.isObject()) {
+ JsonNode t = first(locators, "temporal");
+ if (t != null) {
+ setIfAbsent(doc, "start_ms", t.get("start_ms"));
+ setIfAbsent(doc, "end_ms", t.get("end_ms"));
+ }
+ JsonNode p = first(locators, "paginated");
+ if (p != null) {
+ setIfAbsent(doc, "page", p.get("page"));
+ setIfAbsent(doc, "bbox", p.get("bbox"));
+ }
+ JsonNode s = first(locators, "spatial");
+ if (s != null) {
+ setIfAbsent(doc, "bbox", s.get("bbox"));
+ }
+ JsonNode x = first(locators, "text");
+ if (x != null) {
+ setIfAbsent(doc, "start_offset", x.get("start_offset"));
+ setIfAbsent(doc, "end_offset", x.get("end_offset"));
+ }
+ JsonNode e = first(locators, "embedded");
+ if (e != null) {
+ setIfAbsent(doc, "embedded_id_path", e.get("id_path"));
+ setIfAbsent(doc, "embedded_name", e.get("name"));
+ }
+ }
+ ObjectNode v = doc.has("v") ? (ObjectNode) doc.get("v") : doc.putObject("v");
+ if (chunk.hasNonNull("vector") && chunk.get("vector").isTextual()) {
+ String key = chunk.hasNonNull("producer") ? chunk.get("producer").asText() : "default";
+ try {
+ floats(v.putArray(key), chunk.get("vector").asText());
+ } catch (IllegalArgumentException e) {
+ // a foreign writer's encoding: the document keeps its other fields
+ LOG.warn("chunk vector for {} is not base64 float32; dropped", key);
+ v.remove(key);
+ }
+ }
+ }
+
+ private static void setIfAbsent(ObjectNode doc, String field, JsonNode value) {
+ if (value != null && !value.isNull() && !doc.has(field)) {
+ doc.set(field, value);
+ }
+ }
+
+ private static JsonNode first(JsonNode locators, String kind) {
+ JsonNode list = locators.get(kind);
+ return list != null && list.isArray() && !list.isEmpty() ? list.get(0) : null;
+ }
+
+ /** Base64 big-endian float32, as tika-inference writes vectors. */
+ private static void floats(ArrayNode into, String base64) {
+ FloatBuffer fb = ByteBuffer.wrap(Base64.getDecoder().decode(base64)).asFloatBuffer();
+ while (fb.hasRemaining()) {
+ into.add(fb.get());
+ }
+ }
+
+ private static Metadata without(Metadata metadata, String field) {
+ Metadata copy = new Metadata();
+ copy.putAll(metadata);
+ copy.remove(field);
+ return copy;
+ }
+
// Package-private for testing
static String metadataToJsonContainerInsert(
Metadata metadata,
@@ -205,10 +347,17 @@ public class ESClient {
throws IOException;
String getBulkJson(String id, String routing) throws IOException;
+
+ String writeChunk(ObjectNode chunk) throws IOException;
}
private static class InsertMetadataToJsonWriter implements MetadataToJsonWriter {
+ @Override
+ public String writeChunk(ObjectNode chunk) {
+ return chunk.toString();
+ }
+
@Override
public String writeContainer(Metadata metadata,
ESEmitterConfig.AttachmentStrategy strategy)
@@ -266,6 +415,14 @@ public class ESClient {
private static class UpsertMetadataToJsonWriter implements MetadataToJsonWriter {
+ @Override
+ public String writeChunk(ObjectNode chunk) {
+ ObjectNode wrapper = CHUNK_MAPPER.createObjectNode();
+ wrapper.set("doc", chunk);
+ wrapper.put("doc_as_upsert", true);
+ return wrapper.toString();
+ }
+
@Override
public String writeContainer(Metadata metadata,
ESEmitterConfig.AttachmentStrategy strategy)
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitter.java b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitter.java
index a628a17b61..619598d098 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitter.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitter.java
@@ -107,6 +107,7 @@ public class ESEmitter extends AbstractEmitter {
private void configure() throws TikaConfigException {
ConfigValidator.mustNotBeEmpty("esUrl", config.esUrl());
ConfigValidator.mustNotBeEmpty("idField", config.idField());
+ config.validate();
HttpClientConfig http = config.httpClientConfig();
if (http != null) {
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitterConfig.java b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitterConfig.java
index a68eb7c4cf..2348ca7068 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitterConfig.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESEmitterConfig.java
@@ -33,12 +33,32 @@ import org.apache.tika.plugins.PluginJson;
* If null/empty, falls back to httpClientConfig's userName/password
* for basic auth.
* @param httpClientConfig HTTP connection settings (basic auth, timeouts, proxy)
+ * @param chunkStrategy Where {@code tk:chunks} goes; null means {@link ChunkStrategy#INLINE}
*/
public record ESEmitterConfig(String esUrl, String idField,
AttachmentStrategy attachmentStrategy,
UpdateStrategy updateStrategy, int commitWithin,
String embeddedFileFieldName, String apiKey,
- HttpClientConfig httpClientConfig) {
+ HttpClientConfig httpClientConfig,
+ ChunkStrategy chunkStrategy) {
+ /** Where {@code tk:chunks} goes: inside its document, or one document per chunk. */
+ public enum ChunkStrategy {
+ INLINE, DOCUMENTS
+ }
+
+ public ESEmitterConfig {
+ chunkStrategy = chunkStrategy == null ? ChunkStrategy.INLINE : chunkStrategy;
+ }
+
+ /** The 4.0.0 shape: chunks stay inline. */
+ public ESEmitterConfig(String esUrl, String idField, AttachmentStrategy attachmentStrategy,
+ UpdateStrategy updateStrategy, int commitWithin,
+ String embeddedFileFieldName, String apiKey,
+ HttpClientConfig httpClientConfig) {
+ this(esUrl, idField, attachmentStrategy, updateStrategy, commitWithin,
+ embeddedFileFieldName, apiKey, httpClientConfig, ChunkStrategy.INLINE);
+ }
+
public enum AttachmentStrategy {
SEPARATE_DOCUMENTS, PARENT_CHILD,
}
@@ -49,7 +69,18 @@ public record ESEmitterConfig(String esUrl, String idField,
public static ESEmitterConfig load(final String json)
throws TikaConfigException {
- return PluginJson.read(json, ESEmitterConfig.class);
+ ESEmitterConfig config = PluginJson.read(json, ESEmitterConfig.class);
+ config.validate();
+ return config;
+ }
+
+ /** Chunk documents carry no join field, so they cannot live in a PARENT_CHILD index. */
+ public void validate() throws TikaConfigException {
+ if (chunkStrategy == ChunkStrategy.DOCUMENTS
+ && attachmentStrategy == AttachmentStrategy.PARENT_CHILD) {
+ throw new TikaConfigException("chunkStrategy DOCUMENTS needs attachmentStrategy "
+ + "SEPARATE_DOCUMENTS: chunk documents have no join field");
+ }
}
/** Overrides the record default to prevent {@code apiKey} leaking into logs. */
@@ -59,6 +90,7 @@ public record ESEmitterConfig(String esUrl, String idField,
", idField='" + idField + '\'' +
", attachmentStrategy=" + attachmentStrategy +
", updateStrategy=" + updateStrategy +
+ ", chunkStrategy=" + chunkStrategy +
", apiKey=" + (apiKey != null ? "[REDACTED]" : "null") +
'}';
}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/emitter/es/ESClientTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/emitter/es/ESClientTest.java
index d79ba86f63..4bfcdcae45 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/emitter/es/ESClientTest.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/emitter/es/ESClientTest.java
@@ -18,13 +18,23 @@ package org.apache.tika.pipes.emitter.es;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Locale;
+
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import org.junit.jupiter.api.Test;
import org.apache.tika.TikaTest;
+import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
@@ -168,4 +178,169 @@ public class ESClientTest extends TikaTest {
assertContains("tag2", json);
assertContains("tag3", json);
}
+
+ /** Chunks sharing a correlator become one document with a vector per producer. */
+ @Test
+ public void testChunkDocumentsGroupByCorrelator() throws Exception {
+ String one = Base64.getEncoder().encodeToString(
+ ByteBuffer.allocate(8).putFloat(1.0f).putFloat(0.5f).array());
+ String two = Base64.getEncoder().encodeToString(
+ ByteBuffer.allocate(4).putFloat(2.0f).array());
+ JsonNode chunks = MAPPER.readTree("["
+ + "{\"correlator\":\"t:25000-55000\",\"producer\":\"jina-video\",\"modality\":\"visual\","
+ + " \"vector\":\"" + one + "\",\"locators\":{\"temporal\":[{\"start_ms\":25000,\"end_ms\":55000}],"
+ + " \"embedded\":[{\"id_path\":\"/1\",\"name\":\"clip.mp4\"}]}},"
+ + "{\"text\":\"solo\",\"vector\":\"" + two + "\",\"locators\":{\"text\":[{\"start_offset\":0,\"end_offset\":4}]}},"
+ + "{\"correlator\":\"t:25000-55000\",\"producer\":\"jina-audio\",\"modality\":\"audio\","
+ + " \"vector\":\"" + two + "\",\"locators\":{\"temporal\":[{\"start_ms\":25000,\"end_ms\":55000}]}}"
+ + "]");
+ List<ObjectNode> docs =
+ ESClient.chunkDocuments(chunks, "k-abc", "k");
+ assertEquals(2, docs.size());
+ ObjectNode segment = docs.get(0);
+ assertEquals("k-abc", segment.get("file_id").asText());
+ assertEquals("k", segment.get("container_id").asText());
+ assertEquals("t:25000-55000", segment.get("correlator").asText());
+ assertEquals(25000, segment.get("start_ms").asLong());
+ assertEquals("/1", segment.get("embedded_id_path").asText());
+ assertEquals(2, segment.get("v").size());
+ assertEquals(1.0, segment.get("v").get("jina-video").get(0).asDouble(), 1e-6);
+ assertEquals(0.5, segment.get("v").get("jina-video").get(1).asDouble(), 1e-6);
+ assertEquals(2.0, segment.get("v").get("jina-audio").get(0).asDouble(), 1e-6);
+ assertFalse(segment.has("modality"));
+ ObjectNode solo = docs.get(1);
+ assertEquals("solo", solo.get("text").asText());
+ assertEquals(0, solo.get("start_offset").asInt());
+ assertEquals(2.0, solo.get("v").get("default").get(0).asDouble(), 1e-6);
+ assertFalse(solo.has("correlator"));
+ }
+
+ /** Captures the bulk body instead of posting it. */
+ private static final class CapturingClient extends ESClient {
+ final List<String> bodies = new ArrayList<>();
+
+ CapturingClient(ESEmitterConfig config) {
+ super(config, null);
+ }
+
+ @Override
+ public JsonResponse postJson(String url, String json) {
+ bodies.add(json);
+ return new JsonResponse(200, MAPPER.createObjectNode().put("errors", false));
+ }
+ }
+
+ private static ESEmitterConfig config(ESEmitterConfig.UpdateStrategy update,
+ ESEmitterConfig.ChunkStrategy chunks) {
+ return new ESEmitterConfig("http://es/idx", "_id",
+ ESEmitterConfig.AttachmentStrategy.SEPARATE_DOCUMENTS, update, 10,
+ "embedded", null, null, chunks);
+ }
+
+ private static String vector(float... floats) {
+ ByteBuffer bb = ByteBuffer.allocate(4 * floats.length);
+ for (float f : floats) {
+ bb.putFloat(f);
+ }
+ return Base64.getEncoder().encodeToString(bb.array());
+ }
+
+ /** A real Tika document: reserved keys next to the chunks. */
+ private static Metadata document(String chunks) {
+ Metadata metadata = new Metadata();
+ metadata.set(TikaCoreProperties.TIKA_CONTENT, "the text");
+ metadata.add(TikaCoreProperties.TIKA_PARSED_BY, "SomeParser");
+ metadata.add("tags", "a");
+ metadata.add("tags", "b");
+ setChunks(metadata, chunks);
+ return metadata;
+ }
+
+ private static List<JsonNode> lines(String bulk) throws Exception {
+ List<JsonNode> out = new ArrayList<>();
+ for (String line : bulk.split("\n")) {
+ if (!line.isBlank()) {
+ out.add(MAPPER.readTree(line));
+ }
+ }
+ return out;
+ }
+
+ /** The production path: reserved keys survive, tk:chunks leaves the document, chunk docs follow. */
+ @Test
+ public void testDocumentsBulkBody() throws Exception {
+ String chunks = "[{\"correlator\":\"t:0-30000\",\"producer\":\"v\",\"vector\":\""
+ + vector(1f) + "\",\"locators\":{\"temporal\":[{\"start_ms\":0,\"end_ms\":30000}]}},"
+ + "{\"correlator\":\"t:0-30000\",\"producer\":\"a\",\"vector\":\"" + vector(2f)
+ + "\",\"locators\":{\"temporal\":[{\"start_ms\":0,\"end_ms\":30000}]}},"
+ + "{\"text\":\"solo\",\"vector\":\"" + vector(3f) + "\"}]";
+ for (ESEmitterConfig.UpdateStrategy update : ESEmitterConfig.UpdateStrategy.values()) {
+ CapturingClient client = new CapturingClient(
+ config(update, ESEmitterConfig.ChunkStrategy.DOCUMENTS));
+ client.emitDocument("k", List.of(document(chunks)));
+ List<JsonNode> lines = lines(client.bodies.get(0));
+ assertEquals(6, lines.size(), update + ": one action+body per document");
+ boolean upsert = update == ESEmitterConfig.UpdateStrategy.UPSERT;
+ String action = upsert ? "update" : "index";
+ assertEquals("k", lines.get(0).get(action).get("_id").asText());
+ JsonNode file = upsert ? lines.get(1).get("doc") : lines.get(1);
+ assertEquals("the text", file.get("tk:content").asText(), update.toString());
+ assertTrue(file.get("tk:parsed-by").toString().contains("SomeParser"));
+ assertEquals(2, file.get("tags").size());
+ assertNull(file.get("tk:chunks"), "chunks leave the file document");
+ assertEquals("k-chunk-0", lines.get(2).get(action).get("_id").asText());
+ JsonNode segment = upsert ? lines.get(3).get("doc") : lines.get(3);
+ assertEquals("k", segment.get("file_id").asText());
+ assertEquals(1.0, segment.get("v").get("v").get(0).asDouble(), 1e-6);
+ assertEquals(2.0, segment.get("v").get("a").get(0).asDouble(), 1e-6);
+ if (upsert) {
+ assertTrue(lines.get(3).get("doc_as_upsert").asBoolean());
+ }
+ assertEquals("k-chunk-1", lines.get(4).get(action).get("_id").asText());
+ JsonNode solo = upsert ? lines.get(5).get("doc") : lines.get(5);
+ assertEquals("solo", solo.get("text").asText());
+ }
+ CapturingClient inline = new CapturingClient(
+ config(ESEmitterConfig.UpdateStrategy.OVERWRITE, ESEmitterConfig.ChunkStrategy.INLINE));
+ inline.emitDocument("k", List.of(document(chunks)));
+ List<JsonNode> lines = lines(inline.bodies.get(0));
+ assertEquals(2, lines.size(), "INLINE writes no chunk documents");
+ assertTrue(lines.get(1).get("tk:chunks").isArray());
+ }
+
+ /** A vector that is not base64 loses the vector, not the batch; chunks that are not JSON stay inline. */
+ @Test
+ public void testDocumentsToleratesBadChunks() throws Exception {
+ CapturingClient client = new CapturingClient(
+ config(ESEmitterConfig.UpdateStrategy.OVERWRITE, ESEmitterConfig.ChunkStrategy.DOCUMENTS));
+ client.emitDocument("k", List.of(document(
+ "[{\"text\":\"one\",\"producer\":\"p\",\"vector\":\"not*base64\"},"
+ + "{\"text\":\"two\",\"producer\":\"p\",\"vector\":\"" + vector(1f) + "\"}]")));
+ List<JsonNode> lines = lines(client.bodies.get(0));
+ assertEquals(6, lines.size());
+ assertNull(lines.get(3).get("v"), "the undecodable vector is dropped");
+ assertEquals("one", lines.get(3).get("text").asText());
+ assertEquals(1.0, lines.get(5).get("v").get("p").get(0).asDouble(), 1e-6);
+
+ client = new CapturingClient(
+ config(ESEmitterConfig.UpdateStrategy.OVERWRITE, ESEmitterConfig.ChunkStrategy.DOCUMENTS));
+ client.emitDocument("k", List.of(document("not json")));
+ lines = lines(client.bodies.get(0));
+ assertEquals(2, lines.size(), "nothing to split: the value stays on the document");
+ assertEquals("not json", lines.get(1).get("tk:chunks").asText());
+ }
+
+ /** A missing chunkStrategy is INLINE; DOCUMENTS cannot join a PARENT_CHILD index. */
+ @Test
+ public void testChunkStrategyConfig() throws Exception {
+ String base = "{\"esUrl\": \"http://es/idx\", \"idField\": \"_id\","
+ + " \"updateStrategy\": \"OVERWRITE\", \"attachmentStrategy\": \"%s\"%s}";
+ assertEquals(ESEmitterConfig.ChunkStrategy.INLINE, ESEmitterConfig.load(
+ String.format(Locale.ROOT, base, "PARENT_CHILD", "")).chunkStrategy());
+ assertEquals(ESEmitterConfig.ChunkStrategy.DOCUMENTS, ESEmitterConfig.load(
+ String.format(Locale.ROOT, base, "SEPARATE_DOCUMENTS", ", \"chunkStrategy\": \"DOCUMENTS\""))
+ .chunkStrategy());
+ assertThrows(TikaConfigException.class, () -> ESEmitterConfig.load(
+ String.format(Locale.ROOT, base, "PARENT_CHILD", ", \"chunkStrategy\": \"DOCUMENTS\"")));
+ }
}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/es/ConfigExamplesTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/es/ConfigExamplesTest.java
index ea76e22e15..4ee3fa7fe2 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/es/ConfigExamplesTest.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/es/ConfigExamplesTest.java
@@ -49,6 +49,8 @@ public class ConfigExamplesTest extends AbstractConfigExamplesTest {
config.attachmentStrategy());
assertEquals(ESEmitterConfig.UpdateStrategy.OVERWRITE, config.updateStrategy());
assertEquals(1000, config.commitWithin());
+ assertEquals(ESEmitterConfig.ChunkStrategy.INLINE, config.chunkStrategy(),
+ "the documented example loads");
assertNotNull(config.apiKey());
assertNotNull(config.httpClientConfig());
// The toString override redacts the apiKey value
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/resources/config-examples/es-emitter.json b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/resources/config-examples/es-emitter.json
index e0f341cd6c..4f86ae4188 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/resources/config-examples/es-emitter.json
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/resources/config-examples/es-emitter.json
@@ -8,6 +8,7 @@
"updateStrategy": "OVERWRITE",
"commitWithin": 1000,
"embeddedFileFieldName": "embedded",
+ "chunkStrategy": "INLINE",
"apiKey": "REDACTED_BASE64_ID_AND_KEY",
"httpClientConfig": {
"connectionTimeoutMillis": 10000,
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java
index e02c96a377..66c7609d97 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java
@@ -21,8 +21,14 @@ import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringWriter;
+import java.nio.ByteBuffer;
+import java.nio.FloatBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -30,6 +36,8 @@ import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.CloseableHttpResponse;
@@ -101,6 +109,8 @@ public class OpenSearchClient {
String routing = (config.attachmentStrategy() == OpenSearchEmitterConfig.AttachmentStrategy.PARENT_CHILD) ?
emitKey : null;
+ boolean chunkDocuments =
+ config.chunkStrategy() == OpenSearchEmitterConfig.ChunkStrategy.DOCUMENTS;
for (Metadata metadata : metadataList) {
StringBuilder id = new StringBuilder(emitKey);
if (i > 0) {
@@ -108,16 +118,147 @@ public class OpenSearchClient {
}
String indexJson = metadataToJsonWriter.getBulkJson(id.toString(), routing);
json.append(indexJson).append("\n");
+ JsonNode chunks = chunkDocuments ? chunkArray(metadata.get(CHUNKS_FIELD)) : null;
+ Metadata body = chunks == null ? metadata : without(metadata, CHUNKS_FIELD);
if (i == 0) {
- json.append(metadataToJsonWriter.writeContainer(metadata, config.attachmentStrategy()));
+ json.append(metadataToJsonWriter.writeContainer(body, config.attachmentStrategy()));
} else {
- json.append(metadataToJsonWriter.writeEmbedded(metadata, config.attachmentStrategy(), emitKey, config.embeddedFileFieldName()));
+ json.append(metadataToJsonWriter.writeEmbedded(body, config.attachmentStrategy(), emitKey, config.embeddedFileFieldName()));
}
json.append("\n");
+ if (chunks != null) {
+ appendChunks(id.toString(), emitKey, routing, chunks, json);
+ }
i++;
}
}
+ static final String CHUNKS_FIELD = "tk:chunks";
+
+ /**
+ * Chunks as their own documents, one per unit: chunks sharing a correlator (a segment's
+ * sound and picture, a region's text and crop) merge into one document whose vectors sit
+ * under {@code v.<producer>}, so a search scores the unit and two vectors of one unit add
+ * up on the same hit. Locators are flattened to fields and vectors decoded to float arrays.
+ */
+ /** The chunk array, or null when the value is not one and so stays on the document. */
+ private static JsonNode chunkArray(String chunks) throws IOException {
+ if (chunks == null || !isValidJson(chunks)) {
+ return null;
+ }
+ JsonNode array = VALIDATION_MAPPER.readTree(chunks);
+ return array.isArray() ? array : null;
+ }
+
+ private void appendChunks(String docId, String emitKey, String routing, JsonNode array,
+ StringBuilder json) throws IOException {
+ int n = 0;
+ for (ObjectNode doc : chunkDocuments(array, docId, emitKey)) {
+ json.append(metadataToJsonWriter.getBulkJson(docId + "-chunk-" + n, routing))
+ .append("\n");
+ json.append(metadataToJsonWriter.writeChunk(doc)).append("\n");
+ n++;
+ }
+ }
+
+ /** One document per correlator, in first-seen order; a chunk without one is its own. */
+ static List<ObjectNode> chunkDocuments(JsonNode chunks, String docId, String emitKey) {
+ List<ObjectNode> docs = new ArrayList<>();
+ Map<String, ObjectNode> byCorrelator = new HashMap<>();
+ for (JsonNode chunk : chunks) {
+ String correlator = chunk.hasNonNull("correlator") ? chunk.get("correlator").asText() : null;
+ ObjectNode doc = correlator == null ? null : byCorrelator.get(correlator);
+ if (doc == null) {
+ doc = VALIDATION_MAPPER.createObjectNode();
+ doc.put("file_id", docId);
+ doc.put("container_id", emitKey);
+ if (correlator != null) {
+ doc.put("correlator", correlator);
+ byCorrelator.put(correlator, doc);
+ }
+ docs.add(doc);
+ }
+ mergeChunk(doc, chunk);
+ }
+ for (ObjectNode doc : docs) {
+ if (doc.get("v").isEmpty()) {
+ doc.remove("v");
+ }
+ }
+ return docs;
+ }
+
+ /** Adds what the chunk knows: fields it is the first to set, and its vector under its producer. */
+ private static void mergeChunk(ObjectNode doc, JsonNode chunk) {
+ if (chunk.hasNonNull("text") && !doc.has("text")) {
+ doc.put("text", chunk.get("text").asText());
+ }
+ JsonNode locators = chunk.get("locators");
+ if (locators != null && locators.isObject()) {
+ JsonNode t = first(locators, "temporal");
+ if (t != null) {
+ setIfAbsent(doc, "start_ms", t.get("start_ms"));
+ setIfAbsent(doc, "end_ms", t.get("end_ms"));
+ }
+ JsonNode p = first(locators, "paginated");
+ if (p != null) {
+ setIfAbsent(doc, "page", p.get("page"));
+ setIfAbsent(doc, "bbox", p.get("bbox"));
+ }
+ JsonNode s = first(locators, "spatial");
+ if (s != null) {
+ setIfAbsent(doc, "bbox", s.get("bbox"));
+ }
+ JsonNode x = first(locators, "text");
+ if (x != null) {
+ setIfAbsent(doc, "start_offset", x.get("start_offset"));
+ setIfAbsent(doc, "end_offset", x.get("end_offset"));
+ }
+ JsonNode e = first(locators, "embedded");
+ if (e != null) {
+ setIfAbsent(doc, "embedded_id_path", e.get("id_path"));
+ setIfAbsent(doc, "embedded_name", e.get("name"));
+ }
+ }
+ ObjectNode v = doc.has("v") ? (ObjectNode) doc.get("v") : doc.putObject("v");
+ if (chunk.hasNonNull("vector") && chunk.get("vector").isTextual()) {
+ String key = chunk.hasNonNull("producer") ? chunk.get("producer").asText() : "default";
+ try {
+ floats(v.putArray(key), chunk.get("vector").asText());
+ } catch (IllegalArgumentException e) {
+ // a foreign writer's encoding: the document keeps its other fields
+ LOG.warn("chunk vector for {} is not base64 float32; dropped", key);
+ v.remove(key);
+ }
+ }
+ }
+
+ private static void setIfAbsent(ObjectNode doc, String field, JsonNode value) {
+ if (value != null && !value.isNull() && !doc.has(field)) {
+ doc.set(field, value);
+ }
+ }
+
+ private static JsonNode first(JsonNode locators, String kind) {
+ JsonNode list = locators.get(kind);
+ return list != null && list.isArray() && !list.isEmpty() ? list.get(0) : null;
+ }
+
+ /** Base64 big-endian float32, as tika-inference writes vectors. */
+ private static void floats(ArrayNode into, String base64) {
+ FloatBuffer fb = ByteBuffer.wrap(Base64.getDecoder().decode(base64)).asFloatBuffer();
+ while (fb.hasRemaining()) {
+ into.add(fb.get());
+ }
+ }
+
+ private static Metadata without(Metadata metadata, String field) {
+ Metadata copy = new Metadata();
+ copy.putAll(metadata);
+ copy.remove(field);
+ return copy;
+ }
+
//Only here for testing. These may disappear without notice in the future.
protected static String metadataToJsonContainerInsert(Metadata metadata,
OpenSearchEmitterConfig.AttachmentStrategy attachmentStrategy)
@@ -182,10 +323,17 @@ public class OpenSearchClient {
String emitKey, String embeddedFileFieldName) throws IOException;
String getBulkJson(String id, String routing) throws IOException;
+
+ String writeChunk(ObjectNode chunk) throws IOException;
}
private static class InsertMetadataToJsonWriter implements MetadataToJsonWriter {
+ @Override
+ public String writeChunk(ObjectNode chunk) {
+ return chunk.toString();
+ }
+
@Override
public String writeContainer(Metadata metadata,
OpenSearchEmitterConfig.AttachmentStrategy attachmentStrategy)
@@ -247,6 +395,14 @@ public class OpenSearchClient {
private static class UpsertMetadataToJsonWriter implements MetadataToJsonWriter {
+ @Override
+ public String writeChunk(ObjectNode chunk) {
+ ObjectNode wrapper = VALIDATION_MAPPER.createObjectNode();
+ wrapper.set("doc", chunk);
+ wrapper.put("doc_as_upsert", true);
+ return wrapper.toString();
+ }
+
@Override
public String writeContainer(Metadata metadata, OpenSearchEmitterConfig.AttachmentStrategy attachmentStrategy)
throws IOException {
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java
index 17e25a16ca..877fc8debe 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java
@@ -94,6 +94,7 @@ public class OpenSearchEmitter extends AbstractEmitter {
private void configure() throws TikaConfigException {
ConfigValidator.mustNotBeEmpty("openSearchUrl", config.openSearchUrl());
ConfigValidator.mustNotBeEmpty("idField", config.idField());
+ config.validate();
HttpClientConfig http = config.httpClientConfig();
httpClientFactory.setUserName(http.userName());
httpClientFactory.setPassword(http.password());
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitterConfig.java b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitterConfig.java
index 660885d76d..6684209ac5 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitterConfig.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitterConfig.java
@@ -21,7 +21,26 @@ import org.apache.tika.plugins.PluginJson;
public record OpenSearchEmitterConfig(String openSearchUrl, String idField, AttachmentStrategy attachmentStrategy,
UpdateStrategy updateStrategy, int commitWithin,
- String embeddedFileFieldName, HttpClientConfig httpClientConfig) {
+ String embeddedFileFieldName, HttpClientConfig httpClientConfig,
+ ChunkStrategy chunkStrategy) {
+ /** Where {@code tk:chunks} goes: inside its document, or one document per chunk. */
+ public enum ChunkStrategy {
+ INLINE, DOCUMENTS
+ }
+
+ public OpenSearchEmitterConfig {
+ chunkStrategy = chunkStrategy == null ? ChunkStrategy.INLINE : chunkStrategy;
+ }
+
+ /** The 4.0.0 shape: chunks stay inline. */
+ public OpenSearchEmitterConfig(String openSearchUrl, String idField,
+ AttachmentStrategy attachmentStrategy,
+ UpdateStrategy updateStrategy, int commitWithin,
+ String embeddedFileFieldName, HttpClientConfig httpClientConfig) {
+ this(openSearchUrl, idField, attachmentStrategy, updateStrategy, commitWithin,
+ embeddedFileFieldName, httpClientConfig, ChunkStrategy.INLINE);
+ }
+
public enum AttachmentStrategy {
SEPARATE_DOCUMENTS, PARENT_CHILD,
}
@@ -32,7 +51,18 @@ public record OpenSearchEmitterConfig(String openSearchUrl, String idField, Atta
public static OpenSearchEmitterConfig load(final String json)
throws TikaConfigException {
- return PluginJson.read(json, OpenSearchEmitterConfig.class);
+ OpenSearchEmitterConfig config = PluginJson.read(json, OpenSearchEmitterConfig.class);
+ config.validate();
+ return config;
+ }
+
+ /** Chunk documents carry no join field, so they cannot live in a PARENT_CHILD index. */
+ public void validate() throws TikaConfigException {
+ if (chunkStrategy == ChunkStrategy.DOCUMENTS
+ && attachmentStrategy == AttachmentStrategy.PARENT_CHILD) {
+ throw new TikaConfigException("chunkStrategy DOCUMENTS needs attachmentStrategy "
+ + "SEPARATE_DOCUMENTS: chunk documents have no join field");
+ }
}
}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java
index b097c3bd8a..5b7a924098 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java
@@ -16,13 +16,32 @@
*/
package org.apache.tika.pipes.emitter.opensearch;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Locale;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import org.junit.jupiter.api.Test;
import org.apache.tika.TikaTest;
+import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
public class OpenSearchClientTest extends TikaTest {
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
@Test
public void testSerialization() throws Exception {
Metadata metadata = new Metadata();
@@ -49,4 +68,173 @@ public class OpenSearchClientTest extends TikaTest {
}
}
+
+ /** Chunks sharing a correlator become one document with a vector per producer. */
+ @Test
+ public void testChunkDocumentsGroupByCorrelator() throws Exception {
+ String one = Base64.getEncoder().encodeToString(
+ ByteBuffer.allocate(8).putFloat(1.0f).putFloat(0.5f).array());
+ String two = Base64.getEncoder().encodeToString(
+ ByteBuffer.allocate(4).putFloat(2.0f).array());
+ JsonNode chunks = MAPPER.readTree("["
+ + "{\"correlator\":\"t:25000-55000\",\"producer\":\"jina-video\",\"modality\":\"visual\","
+ + " \"vector\":\"" + one + "\",\"locators\":{\"temporal\":[{\"start_ms\":25000,\"end_ms\":55000}],"
+ + " \"embedded\":[{\"id_path\":\"/1\",\"name\":\"clip.mp4\"}]}},"
+ + "{\"text\":\"solo\",\"vector\":\"" + two + "\",\"locators\":{\"text\":[{\"start_offset\":0,\"end_offset\":4}]}},"
+ + "{\"correlator\":\"t:25000-55000\",\"producer\":\"jina-audio\",\"modality\":\"audio\","
+ + " \"vector\":\"" + two + "\",\"locators\":{\"temporal\":[{\"start_ms\":25000,\"end_ms\":55000}]}}"
+ + "]");
+ List<ObjectNode> docs =
+ OpenSearchClient.chunkDocuments(chunks, "k-abc", "k");
+ assertEquals(2, docs.size());
+ ObjectNode segment = docs.get(0);
+ assertEquals("k-abc", segment.get("file_id").asText());
+ assertEquals("k", segment.get("container_id").asText());
+ assertEquals("t:25000-55000", segment.get("correlator").asText());
+ assertEquals(25000, segment.get("start_ms").asLong());
+ assertEquals("/1", segment.get("embedded_id_path").asText());
+ assertEquals(2, segment.get("v").size());
+ assertEquals(1.0, segment.get("v").get("jina-video").get(0).asDouble(), 1e-6);
+ assertEquals(0.5, segment.get("v").get("jina-video").get(1).asDouble(), 1e-6);
+ assertEquals(2.0, segment.get("v").get("jina-audio").get(0).asDouble(), 1e-6);
+ assertFalse(segment.has("modality"));
+ ObjectNode solo = docs.get(1);
+ assertEquals("solo", solo.get("text").asText());
+ assertEquals(0, solo.get("start_offset").asInt());
+ assertEquals(2.0, solo.get("v").get("default").get(0).asDouble(), 1e-6);
+ assertFalse(solo.has("correlator"));
+ }
+
+ private static void setChunks(Metadata metadata, String json) {
+ metadata.setTrusted(TikaCoreProperties.TIKA_CHUNKS.getName(), json);
+ }
+
+ /** Captures the bulk body instead of posting it. */
+ private static final class CapturingClient extends OpenSearchClient {
+ final List<String> bodies = new ArrayList<>();
+
+ CapturingClient(OpenSearchEmitterConfig config) {
+ super(config, null);
+ }
+
+ @Override
+ public JsonResponse postJson(String url, String json) {
+ bodies.add(json);
+ return new JsonResponse(200, MAPPER.createObjectNode().put("errors", false));
+ }
+ }
+
+ private static OpenSearchEmitterConfig config(OpenSearchEmitterConfig.UpdateStrategy update,
+ OpenSearchEmitterConfig.ChunkStrategy chunks) {
+ return new OpenSearchEmitterConfig("http://os/idx", "_id",
+ OpenSearchEmitterConfig.AttachmentStrategy.SEPARATE_DOCUMENTS, update, 10,
+ "embedded", null, chunks);
+ }
+
+ private static String vector(float... floats) {
+ ByteBuffer bb = ByteBuffer.allocate(4 * floats.length);
+ for (float f : floats) {
+ bb.putFloat(f);
+ }
+ return Base64.getEncoder().encodeToString(bb.array());
+ }
+
+ /** A real Tika document: reserved keys next to the chunks. */
+ private static Metadata document(String chunks) {
+ Metadata metadata = new Metadata();
+ metadata.set(TikaCoreProperties.TIKA_CONTENT, "the text");
+ metadata.add(TikaCoreProperties.TIKA_PARSED_BY, "SomeParser");
+ metadata.add("tags", "a");
+ metadata.add("tags", "b");
+ setChunks(metadata, chunks);
+ return metadata;
+ }
+
+ private static List<JsonNode> lines(String bulk) throws Exception {
+ List<JsonNode> out = new ArrayList<>();
+ for (String line : bulk.split("\n")) {
+ if (!line.isBlank()) {
+ out.add(MAPPER.readTree(line));
+ }
+ }
+ return out;
+ }
+
+ /** The production path: reserved keys survive, tk:chunks leaves the document, chunk docs follow. */
+ @Test
+ public void testDocumentsBulkBody() throws Exception {
+ String chunks = "[{\"correlator\":\"t:0-30000\",\"producer\":\"v\",\"vector\":\""
+ + vector(1f) + "\",\"locators\":{\"temporal\":[{\"start_ms\":0,\"end_ms\":30000}]}},"
+ + "{\"correlator\":\"t:0-30000\",\"producer\":\"a\",\"vector\":\"" + vector(2f)
+ + "\",\"locators\":{\"temporal\":[{\"start_ms\":0,\"end_ms\":30000}]}},"
+ + "{\"text\":\"solo\",\"vector\":\"" + vector(3f) + "\"}]";
+ for (OpenSearchEmitterConfig.UpdateStrategy update : OpenSearchEmitterConfig.UpdateStrategy.values()) {
+ CapturingClient client = new CapturingClient(
+ config(update, OpenSearchEmitterConfig.ChunkStrategy.DOCUMENTS));
+ client.emitDocument("k", List.of(document(chunks)));
+ List<JsonNode> lines = lines(client.bodies.get(0));
+ assertEquals(6, lines.size(), update + ": one action+body per document");
+ boolean upsert = update == OpenSearchEmitterConfig.UpdateStrategy.UPSERT;
+ String action = upsert ? "update" : "index";
+ assertEquals("k", lines.get(0).get(action).get("_id").asText());
+ JsonNode file = upsert ? lines.get(1).get("doc") : lines.get(1);
+ assertEquals("the text", file.get("tk:content").asText(), update.toString());
+ assertTrue(file.get("tk:parsed-by").toString().contains("SomeParser"));
+ assertEquals(2, file.get("tags").size());
+ assertNull(file.get("tk:chunks"), "chunks leave the file document");
+ assertEquals("k-chunk-0", lines.get(2).get(action).get("_id").asText());
+ JsonNode segment = upsert ? lines.get(3).get("doc") : lines.get(3);
+ assertEquals("k", segment.get("file_id").asText());
+ assertEquals(1.0, segment.get("v").get("v").get(0).asDouble(), 1e-6);
+ assertEquals(2.0, segment.get("v").get("a").get(0).asDouble(), 1e-6);
+ if (upsert) {
+ assertTrue(lines.get(3).get("doc_as_upsert").asBoolean());
+ }
+ assertEquals("k-chunk-1", lines.get(4).get(action).get("_id").asText());
+ JsonNode solo = upsert ? lines.get(5).get("doc") : lines.get(5);
+ assertEquals("solo", solo.get("text").asText());
+ }
+ CapturingClient inline = new CapturingClient(
+ config(OpenSearchEmitterConfig.UpdateStrategy.OVERWRITE, OpenSearchEmitterConfig.ChunkStrategy.INLINE));
+ inline.emitDocument("k", List.of(document(chunks)));
+ List<JsonNode> lines = lines(inline.bodies.get(0));
+ assertEquals(2, lines.size(), "INLINE writes no chunk documents");
+ assertTrue(lines.get(1).get("tk:chunks").isArray());
+ }
+
+ /** A vector that is not base64 loses the vector, not the batch; chunks that are not JSON stay inline. */
+ @Test
+ public void testDocumentsToleratesBadChunks() throws Exception {
+ CapturingClient client = new CapturingClient(
+ config(OpenSearchEmitterConfig.UpdateStrategy.OVERWRITE, OpenSearchEmitterConfig.ChunkStrategy.DOCUMENTS));
+ client.emitDocument("k", List.of(document(
+ "[{\"text\":\"one\",\"producer\":\"p\",\"vector\":\"not*base64\"},"
+ + "{\"text\":\"two\",\"producer\":\"p\",\"vector\":\"" + vector(1f) + "\"}]")));
+ List<JsonNode> lines = lines(client.bodies.get(0));
+ assertEquals(6, lines.size());
+ assertNull(lines.get(3).get("v"), "the undecodable vector is dropped");
+ assertEquals("one", lines.get(3).get("text").asText());
+ assertEquals(1.0, lines.get(5).get("v").get("p").get(0).asDouble(), 1e-6);
+
+ client = new CapturingClient(
+ config(OpenSearchEmitterConfig.UpdateStrategy.OVERWRITE, OpenSearchEmitterConfig.ChunkStrategy.DOCUMENTS));
+ client.emitDocument("k", List.of(document("not json")));
+ lines = lines(client.bodies.get(0));
+ assertEquals(2, lines.size(), "nothing to split: the value stays on the document");
+ assertEquals("not json", lines.get(1).get("tk:chunks").asText());
+ }
+
+ /** A missing chunkStrategy is INLINE; DOCUMENTS cannot join a PARENT_CHILD index. */
+ @Test
+ public void testChunkStrategyConfig() throws Exception {
+ String base = "{\"openSearchUrl\": \"http://os/idx\", \"idField\": \"_id\","
+ + " \"updateStrategy\": \"OVERWRITE\", \"attachmentStrategy\": \"%s\"%s}";
+ assertEquals(OpenSearchEmitterConfig.ChunkStrategy.INLINE, OpenSearchEmitterConfig.load(
+ String.format(Locale.ROOT, base, "PARENT_CHILD", "")).chunkStrategy());
+ assertEquals(OpenSearchEmitterConfig.ChunkStrategy.DOCUMENTS, OpenSearchEmitterConfig.load(
+ String.format(Locale.ROOT, base, "SEPARATE_DOCUMENTS", ", \"chunkStrategy\": \"DOCUMENTS\""))
+ .chunkStrategy());
+ assertThrows(TikaConfigException.class, () -> OpenSearchEmitterConfig.load(
+ String.format(Locale.ROOT, base, "PARENT_CHILD", ", \"chunkStrategy\": \"DOCUMENTS\"")));
+ }
}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/opensearch/ConfigExamplesTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/opensearch/ConfigExamplesTest.java
index e673f25b88..faa4f0dc7d 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/opensearch/ConfigExamplesTest.java
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/opensearch/ConfigExamplesTest.java
@@ -48,6 +48,8 @@ public class ConfigExamplesTest extends AbstractConfigExamplesTest {
assertEquals(OpenSearchEmitterConfig.UpdateStrategy.OVERWRITE,
config.updateStrategy());
assertEquals(1000, config.commitWithin());
+ assertEquals(OpenSearchEmitterConfig.ChunkStrategy.INLINE, config.chunkStrategy(),
+ "the documented example loads");
assertNotNull(config.httpClientConfig());
assertEquals("admin", config.httpClientConfig().userName());
}
diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/resources/config-examples/opensearch-emitter.json b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/resources/config-examples/opensearch-emitter.json
index 0221fbfd61..a6c0cda04f 100644
--- a/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/resources/config-examples/opensearch-emitter.json
+++ b/tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/resources/config-examples/opensearch-emitter.json
@@ -8,6 +8,7 @@
"updateStrategy": "OVERWRITE",
"commitWithin": 1000,
"embeddedFileFieldName": "embedded",
+ "chunkStrategy": "INLINE",
"httpClientConfig": {
"userName": "admin",
"password": "REDACTED",
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 a345313430..6bdc0b7d6f 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
@@ -36,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.Modality;
import org.apache.tika.parser.inference.TextChunker;
/**
@@ -47,10 +48,13 @@ 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", "chunker");
+ "maxChunks", "maxBytes", "enabled", "_mime-include", "_mime-exclude", "chunker",
+ "modality");
private static final Pattern LEGAL_ID = Pattern.compile("[A-Za-z0-9._-]+");
- /** Largest unit a binding takes unless it says otherwise. */
+ /** Largest unit a binding takes unless it says otherwise; for MEDIA, one cut segment. */
static final long DEFAULT_MAX_BYTES = 20L * 1024 * 1024;
+ /** Segments a MEDIA binding embeds per document unless it says otherwise: a request may narrow the grid, not widen it past this. */
+ static final int DEFAULT_MEDIA_MAX_CHUNKS = 480;
/** Image types no embedding endpoint takes; a binding's own include list overrides. */
private static final List<String> NON_RASTER = List.of("image/svg+xml", "image/vnd.dwg",
"image/vnd.dxf", "image/x-emf", "image/x-wmf", "image/wmf", "image/emf",
@@ -121,19 +125,30 @@ class InferenceLoader implements ComponentLoader<InferenceDispatcher> {
} else {
taskNames.add("embed");
}
- int maxChunks = bounded(entry, "maxChunks", -1, id).intValue();
+ int maxChunks = bounded(entry, "maxChunks",
+ input == InputKind.MEDIA ? DEFAULT_MEDIA_MAX_CHUNKS : -1, id).intValue();
long maxBytes = bounded(entry, "maxBytes", DEFAULT_MAX_BYTES, id);
Set<MediaType> include = mimeTypes(entry, "_mime-include", id);
Set<MediaType> exclude = mimeTypes(entry, "_mime-exclude", id);
+ if (input == InputKind.MEDIA) {
+ for (MediaType t : include) {
+ if (InputKind.of(t) != InputKind.MEDIA) {
+ throw new TikaConfigException("binding \"" + id + "\": \"_mime-include\" "
+ + t + " is not an audio or video type; a MEDIA binding would never see it");
+ }
+ }
+ }
if (input == InputKind.IMAGES && include.isEmpty() && exclude.isEmpty()) {
for (String t : NON_RASTER) {
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), chunker);
+ Modality modality = modality(entry, id, input);
+ InferenceBinding binding = InferenceBinding.builder(id, engineName, input)
+ .tasks(taskNames).include(include).exclude(exclude).maxChunks(maxChunks)
+ .maxBytes(maxBytes).enabled(entry.path("enabled").asBoolean(true))
+ .chunker(chunker).modality(modality).build();
List<InferenceTask> tasks = new ArrayList<>();
for (String taskName : taskNames) {
InferenceTask task;
@@ -180,6 +195,34 @@ class InferenceLoader implements ComponentLoader<InferenceDispatcher> {
}
}
+ /** Implied by the input except for MEDIA, which must say {@code audio} or {@code visual}. */
+ private static Modality modality(JsonNode entry, String id, InputKind input)
+ throws TikaConfigException {
+ Modality implied = Modality.implied(input);
+ if (!entry.has("modality")) {
+ if (implied == null) {
+ throw new TikaConfigException("binding \"" + id + "\" is on MEDIA and needs "
+ + "\"modality\": \"audio\" or \"visual\"");
+ }
+ return implied;
+ }
+ Modality given;
+ try {
+ given = Modality.valueOf(entry.get("modality").asText().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new TikaConfigException("binding \"" + id + "\": \"modality\" must be one of "
+ + "text, visual, audio");
+ }
+ if (implied != null && given != implied) {
+ throw new TikaConfigException("binding \"" + id + "\": " + input + " is always "
+ + implied.name().toLowerCase(Locale.ROOT));
+ }
+ if (input == InputKind.MEDIA && given == Modality.TEXT) {
+ throw new TikaConfigException("binding \"" + id + "\": MEDIA has no text channel");
+ }
+ return given;
+ }
+
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 a5a4b5fc5b..fe45d88c00 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
@@ -46,6 +46,8 @@ import org.apache.tika.parser.inference.InferenceBinding;
import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.parser.inference.InferenceSelection;
import org.apache.tika.parser.inference.InputKind;
+import org.apache.tika.parser.inference.MediaConfig;
+import org.apache.tika.parser.inference.Modality;
public class InferenceLoaderTest {
@@ -226,4 +228,59 @@ public class InferenceLoaderTest {
assertNotNull(e.getMessage());
}
}
+
+ @Test
+ public void testModality() throws Exception {
+ TikaLoader loader = load("{" + ENGINES + ", \"inference\": ["
+ + " { \"id\": \"a\", \"engine\": \"one\", \"input\": \"MEDIA\","
+ + " \"modality\": \"audio\", \"tasks\": [\"test-task\"] },"
+ + " { \"id\": \"p\", \"engine\": \"one\", \"input\": \"IMAGES\","
+ + " \"tasks\": [\"test-task\"] } ] }");
+ InferenceDispatcher dispatcher = loader.get(InferenceDispatcher.class);
+ assertEquals(Modality.AUDIO, dispatcher.getBound().get(0).binding().getModality());
+ assertEquals(InferenceLoader.DEFAULT_MEDIA_MAX_CHUNKS,
+ dispatcher.getBound().get(0).binding().getMaxChunks(),
+ "a MEDIA binding has a config-side ceiling on segments");
+ assertEquals(-1, dispatcher.getBound().get(1).binding().getMaxChunks());
+ assertEquals(Modality.VISUAL, dispatcher.getBound().get(1).binding().getModality(),
+ "implied by the input");
+ for (String bad : new String[]{
+ "{ \"engine\": \"one\", \"input\": \"MEDIA\", \"tasks\": [\"test-task\"] }",
+ "{ \"engine\": \"one\", \"input\": \"MEDIA\", \"modality\": \"text\","
+ + " \"tasks\": [\"test-task\"] }",
+ "{ \"engine\": \"one\", \"input\": \"IMAGES\", \"modality\": \"audio\","
+ + " \"tasks\": [\"test-task\"] }",
+ "{ \"engine\": \"one\", \"input\": \"MEDIA\", \"modality\": \"smell\","
+ + " \"tasks\": [\"test-task\"] }",
+ "{ \"engine\": \"one\", \"input\": \"MEDIA\", \"modality\": \"audio\","
+ + " \"_mime-include\": [\"image/png\"], \"tasks\": [\"test-task\"] }",
+ "{ \"engine\": \"one\", \"input\": \"MEDIA\", \"modality\": \"audio\","
+ + " \"_mime-include\": [\"audio/x-mpegurl\"], \"tasks\": [\"test-task\"] }"}) {
+ assertThrows(TikaConfigException.class, () -> load("{" + ENGINES
+ + ", \"inference\": [" + bad + "] }").get(InferenceDispatcher.class), bad);
+ }
+ }
+
+ /** The media grid rides the parse-context like the inference selection does. */
+ @Test
+ public void testMediaBlockResolves() throws Exception {
+ TikaLoader loader = load("{" + ENGINES + ", \"inference\": ["
+ + " { \"id\": \"a\", \"engine\": \"one\", \"input\": \"MEDIA\","
+ + " \"modality\": \"audio\", \"tasks\": [\"test-task\"] } ],"
+ + " \"parse-context\": { \"media\": { \"segment\": { \"seconds\": 10,"
+ + " \"overlap\": 2 }, \"maxSegments\": 7 } } }");
+ MediaConfig media = loader.loadParseContext().get(MediaConfig.class);
+ assertNotNull(media, "the parse-context block resolves to the class-keyed DTO");
+ assertEquals(10, media.getSegment().getSeconds());
+ assertEquals(2, media.getSegment().getOverlap());
+ assertEquals(7, media.getMaxSegments());
+ assertEquals(30, new MediaConfig().getSegment().getSeconds(), "default window");
+ assertEquals(5, new MediaConfig().getSegment().getOverlap(), "default overlap");
+ assertThrows(TikaConfigException.class, () -> load("{" + ENGINES + ", \"inference\": ["
+ + " { \"id\": \"a\", \"engine\": \"one\", \"input\": \"MEDIA\","
+ + " \"modality\": \"audio\", \"tasks\": [\"test-task\"] } ],"
+ + " \"parse-context\": { \"media\": { \"segment\": { \"seconds\": 10,"
+ + " \"overlap\": 10 } } } }").loadParseContext(),
+ "a bad grid fails at load, not per document");
+ }
}