This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4829-carry-inline-bytes in repository https://gitbox.apache.org/repos/asf/tika.git
commit 8ef20cac5cfee11012b7fb9b95968b35621f9a00 Author: tallison <[email protected]> AuthorDate: Thu Aug 20 17:17:31 2026 -0400 TIKA-4829 -- handle byte passing more cleanly --- CHANGES.txt | 5 + .../tika/pipes/core/fetcher/InlineBytes.java | 6 +- .../serialization/FetchEmitTupleDeserializer.java | 16 ++- .../serialization/FetchEmitTupleSerializer.java | 19 +++- .../pipes/core/serialization/JsonPipesIpc.java | 9 +- .../core/serialization/InlineBytesWireTest.java | 121 +++++++++++++++++++++ 6 files changed, 169 insertions(+), 7 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 3b26ecb220..070b739474 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,3 +1,8 @@ +Release 4.1.0 - unreleased + + * TIKA-4829 -- pipes IPC: carry inline document bytes as a raw binary field + of the tuple and disable Smile's 7-bit binary encoding. + Release 4.0.0 - 8/18/2026 This section is the complete delta from 3.x. It includes everything first diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java index 2981847f49..eb915f3713 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/fetcher/InlineBytes.java @@ -27,8 +27,10 @@ import org.apache.tika.annotation.TikaComponent; * hand it to the forked worker. * <p> * Read by {@link BytesFetcher}, which the tuple selects with fetcher id - * {@link BytesFetcher#FETCHER_ID}. The IPC is Smile, so this rides as native binary rather than - * base64; it counts against {@code maxIpcPayloadBytes} like any other part of the request. + * {@link BytesFetcher#FETCHER_ID}. On the IPC wire this travels as a dedicated raw-binary + * field of the tuple, bypassing the parse-context config machinery (whose text-JSON round + * trip would base64 it); it counts against {@code maxIpcPayloadBytes} like any other part + * of the request. */ @TikaComponent(name = "inline-bytes", spi = false) public class InlineBytes implements Serializable { diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java index 07bc0c6f09..67b9cf2aff 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleDeserializer.java @@ -24,6 +24,7 @@ import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer. import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.FETCH_RANGE_END; import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.FETCH_RANGE_START; import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.ID; +import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.INLINE_BYTES; import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.METADATA_KEY; import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.ON_PARSE_EXCEPTION; import static org.apache.tika.serialization.serdes.ParseContextSerializer.PARSE_CONTEXT; @@ -45,13 +46,14 @@ import org.apache.tika.pipes.api.ComponentIds; import org.apache.tika.pipes.api.FetchEmitTuple; import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.api.fetcher.FetchKey; +import org.apache.tika.pipes.core.fetcher.InlineBytes; import org.apache.tika.serialization.serdes.ParseContextDeserializer; public class FetchEmitTupleDeserializer extends JsonDeserializer<FetchEmitTuple> { private static final Set<String> KNOWN_KEYS = Set.of( ID, FETCHER, FETCH_KEY, EMITTER, EMIT_KEY, FETCH_RANGE_START, FETCH_RANGE_END, - METADATA_KEY, PARSE_CONTEXT, ON_PARSE_EXCEPTION); + METADATA_KEY, PARSE_CONTEXT, ON_PARSE_EXCEPTION, INLINE_BYTES); private final boolean restricted; @@ -81,6 +83,12 @@ public class FetchEmitTupleDeserializer extends JsonDeserializer<FetchEmitTuple> public FetchEmitTuple deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException, JacksonException { JsonNode root = jsonParser.readValueAsTree(); rejectUnknownKeys(root); + if (restricted && root.has(INLINE_BYTES)) { + // Same stance as the reserved __bytes fetcher id: inline content is how the + // host hands a payload to its own child, never something a request supplies. + throw new IOException("'" + INLINE_BYTES + + "' is reserved for the host's IPC and may not be supplied by a request"); + } String id = readVal(ID, root, null, true); String fetcherId = normalizeId(readVal(FETCHER, root, null, true), "fetcher"); @@ -95,6 +103,12 @@ public class FetchEmitTupleDeserializer extends JsonDeserializer<FetchEmitTuple> // its parseContext so it cannot introduce wire-blocked components (parsers, detectors, ...). ParseContext parseContext = parseContextNode == null ? new ParseContext() : ParseContextDeserializer.readParseContext(parseContextNode, true); + JsonNode inlineBytesNode = root.get(INLINE_BYTES); + if (inlineBytesNode != null) { + // Set typed, not as a lazy json config: resolveAll never sees the payload, and + // the child's context merge copies typed entries as-is. + parseContext.set(InlineBytes.class, new InlineBytes(inlineBytesNode.binaryValue())); + } FetchEmitTuple.ON_PARSE_EXCEPTION onParseException = readOnParseException(root); return new FetchEmitTuple(id, new FetchKey(fetcherId, fetchKey, fetchRangeStart, fetchRangeEnd), diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java index 9dc06b560c..adf26226a3 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/FetchEmitTupleSerializer.java @@ -25,7 +25,9 @@ import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.JsonSerializer; import com.fasterxml.jackson.databind.SerializerProvider; +import org.apache.tika.parser.ParseContext; import org.apache.tika.pipes.api.FetchEmitTuple; +import org.apache.tika.pipes.core.fetcher.InlineBytes; import org.apache.tika.utils.StringUtils; public class FetchEmitTupleSerializer extends JsonSerializer<FetchEmitTuple> { @@ -38,6 +40,7 @@ public class FetchEmitTupleSerializer extends JsonSerializer<FetchEmitTuple> { public static final String EMIT_KEY = "emitKey"; public static final String METADATA_KEY = "metadata"; public static final String ON_PARSE_EXCEPTION = "onParseException"; + public static final String INLINE_BYTES = "inlineBytes"; public void serialize(FetchEmitTuple t, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException { @@ -57,8 +60,20 @@ public class FetchEmitTupleSerializer extends JsonSerializer<FetchEmitTuple> { jsonGenerator.writeObjectField(METADATA_KEY, t.getMetadata()); } jsonGenerator.writeStringField(ON_PARSE_EXCEPTION, t.getOnParseException().name().toLowerCase(Locale.US)); - if (!t.getParseContext().isEmpty()) { - jsonGenerator.writeObjectField(PARSE_CONTEXT, t.getParseContext()); + // The document payload is data, not config: written as a top-level binary field so it + // never enters the parse-context lazy-config machinery, whose per-entry text-JSON + // round trip would base64 it (see ParseContextDeserializer.readParseContext). + ParseContext parseContext = t.getParseContext(); + InlineBytes inlineBytes = parseContext.get(InlineBytes.class); + if (inlineBytes != null && inlineBytes.getBytes() != null) { + jsonGenerator.writeBinaryField(INLINE_BYTES, inlineBytes.getBytes()); + ParseContext copy = new ParseContext(); + copy.copyFrom(parseContext); + copy.set(InlineBytes.class, null); + parseContext = copy; + } + if (!parseContext.isEmpty()) { + jsonGenerator.writeObjectField(PARSE_CONTEXT, parseContext); } jsonGenerator.writeEndObject(); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java index d35f45a66d..b946e85f85 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/JsonPipesIpc.java @@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.StreamReadConstraints; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.module.SimpleModule; import com.fasterxml.jackson.dataformat.smile.SmileFactory; +import com.fasterxml.jackson.dataformat.smile.SmileGenerator; import org.apache.tika.config.loader.TikaObjectMapperFactory; import org.apache.tika.pipes.api.FetchEmitTuple; @@ -42,8 +43,12 @@ public class JsonPipesIpc { private static final ObjectMapper OBJECT_MAPPER; static { - // Use SmileFactory for binary format - more compact and faster than text JSON - SmileFactory smileFactory = new SmileFactory(); + // Use SmileFactory for binary format - more compact and faster than text JSON. + // 7-bit binary encoding (the Smile default) costs +14% size and a transcode pass + // on every binary payload; this is a private same-version channel, so write raw. + SmileFactory smileFactory = SmileFactory.builder() + .disable(SmileGenerator.Feature.ENCODE_BINARY_AS_7BIT) + .build(); // Configure stream constraints for large content (e.g., 30MB+ documents) // Default Jackson limit is 20MB which is too small for IPC with large documents diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/InlineBytesWireTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/InlineBytesWireTest.java new file mode 100644 index 0000000000..329da5b09d --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/InlineBytesWireTest.java @@ -0,0 +1,121 @@ +/* + * 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.pipes.core.serialization; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.StringReader; +import java.util.Random; + +import org.junit.jupiter.api.Test; + +import org.apache.tika.io.TikaInputStream; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.metadata.TikaCoreProperties; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.pipes.api.FetchEmitTuple; +import org.apache.tika.pipes.api.ParseMode; +import org.apache.tika.pipes.api.emitter.EmitKey; +import org.apache.tika.pipes.api.fetcher.FetchKey; +import org.apache.tika.pipes.core.fetcher.BytesFetcher; +import org.apache.tika.pipes.core.fetcher.InlineBytes; +import org.apache.tika.serialization.ParseContextUtils; + +/** + * The inline payload travels as a dedicated raw-binary field of the tuple, not through the + * parse-context config machinery whose text-JSON round trip would base64 it. + */ +public class InlineBytesWireTest { + + private static byte[] payload(int size) { + byte[] b = new byte[size]; + new Random(17).nextBytes(b); + return b; + } + + private static FetchEmitTuple tuple(byte[] payload) { + ParseContext ctx = new ParseContext(); + ctx.set(InlineBytes.class, new InlineBytes(payload)); + ctx.set(ParseMode.class, ParseMode.RMETA); + return new FetchEmitTuple("t", new FetchKey(BytesFetcher.FETCHER_ID, "doc.bin"), + EmitKey.NO_EMIT, new Metadata(), ctx); + } + + @Test + public void ipcRoundTripPreservesPayload() throws Exception { + byte[] payload = payload(100_000); + FetchEmitTuple t = tuple(payload); + byte[] wire = JsonPipesIpc.toBytes(t); + // serializing must not strip the payload from the caller's live context + assertNotNull(t.getParseContext().get(InlineBytes.class)); + + FetchEmitTuple back = JsonPipesIpc.fromBytes(wire, FetchEmitTuple.class); + // the child's path: merge, resolve, fetch + ParseContext merged = new ParseContext(); + merged.copyFrom(back.getParseContext()); + ParseContextUtils.resolveAll(merged, getClass().getClassLoader()); + assertEquals(ParseMode.RMETA, merged.get(ParseMode.class)); + + Metadata metadata = new Metadata(); + try (TikaInputStream tis = new BytesFetcher().fetch("doc.bin", metadata, merged)) { + assertArrayEquals(payload, tis.readAllBytes()); + } + assertEquals("doc.bin", metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY)); + } + + @Test + public void wireCarriesRawBinary() throws Exception { + byte[] payload = payload(1_000_000); + byte[] wire = JsonPipesIpc.toBytes(tuple(payload)); + // raw binary: no base64 (+33%) and no Smile 7-bit encoding (+14%) + assertTrue(wire.length >= payload.length, "wire shorter than payload?"); + assertTrue(wire.length < payload.length + 1024, + "payload not raw on the wire: " + wire.length + " bytes for " + payload.length); + } + + @Test + public void payloadBypassesParseContextConfigs() throws Exception { + byte[] wire = JsonPipesIpc.toBytes(tuple(payload(1000))); + FetchEmitTuple back = JsonPipesIpc.fromBytes(wire, FetchEmitTuple.class); + assertFalse(back.getParseContext().hasJsonConfig("inline-bytes"), + "payload leaked into the lazy-config path"); + assertNotNull(back.getParseContext().get(InlineBytes.class)); + } + + @Test + public void requestBodyRejectsInlineBytes() { + String json = "{\"id\":\"t\",\"fetcher\":\"f\",\"fetchKey\":\"k\"," + + "\"emitter\":\"e\",\"inlineBytes\":\"QUJD\"}"; + Exception e = assertThrows(Exception.class, + () -> JsonFetchEmitTuple.fromJson(new StringReader(json))); + assertTrue(root(e).contains("reserved for the host's IPC"), + "expected inlineBytes rejection, got: " + root(e)); + } + + private static String root(Throwable t) { + Throwable r = t; + while (r.getCause() != null && r.getCause() != r) { + r = r.getCause(); + } + return String.valueOf(r.getMessage()); + } +}
