This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4868-performance-improvements in repository https://gitbox.apache.org/repos/asf/tika.git
commit 25a2f6059e1cc532cc9091e5941606e8f6c0e21d Author: tallison <[email protected]> AuthorDate: Wed Sep 2 05:51:26 2026 -0400 side channel content bytes for fewer encode/decode in tika-server --- CHANGES.txt | 9 ++++ .../apache/tika/pipes/api/emitter/EmitData.java | 9 ++++ .../apache/tika/pipes/core/ContentBytesConfig.java | 44 ++++++++++++++++ .../tika/pipes/core/emitter/EmitDataImpl.java | 22 +++++++- .../core/serialization/EmitDataDeserializer.java | 7 ++- .../core/serialization/EmitDataSerializer.java | 6 +++ .../apache/tika/pipes/core/server/EmitHandler.java | 32 ++++++++++++ .../serialization/EmitDataContentBytesTest.java | 58 ++++++++++++++++++++++ .../server/core/resource/PipesParsingHelper.java | 32 +++++++++++- .../tika/server/core/resource/TikaResource.java | 44 ++++++++++------ 10 files changed, 244 insertions(+), 19 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 8ef82bedf1..d5beba927b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,14 @@ Release 4.1.0 - unreleased + * tika-server's raw-output endpoints (/tika, /tika/text, ...) carry the + extracted content as raw UTF-8 bytes from the pipes worker to the HTTP + response instead of a Smile-encoded string: one encode in the worker + replaces a string transcode on both sides of the IPC plus a re-encode + at the HTTP layer (9MB text: 115ms -> 75ms end-to-end). Opt-in via + the new content-bytes-config parse-context component, which moves + CONTENT_ONLY passback content out of tk:content into + EmitData.getContentBytes() (TIKA-4868). + * Detection hot-path cleanups: MagicMatch resolves its detector via double-checked locking instead of a synchronized method per eval; glob patterns are compiled once at registration instead of per diff --git a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/emitter/EmitData.java b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/emitter/EmitData.java index ec6266ed23..0a337360ba 100644 --- a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/emitter/EmitData.java +++ b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/emitter/EmitData.java @@ -28,6 +28,15 @@ public interface EmitData { String getContainerStackTrace(); + /** + * Raw UTF-8 content bytes, set only when the request opted in via + * {@code content-bytes-config} under {@code ParseMode.CONTENT_ONLY}; + * the metadata then no longer carries {@code TIKA_CONTENT}. Null otherwise. + */ + default byte[] getContentBytes() { + return null; + } + long getEstimatedSizeBytes(); /** diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ContentBytesConfig.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ContentBytesConfig.java new file mode 100644 index 0000000000..5d00bbd6ff --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/ContentBytesConfig.java @@ -0,0 +1,44 @@ +/* + * 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; + +import java.io.Serializable; + +import org.apache.tika.annotation.TikaComponent; + +/** + * Opt-in for CONTENT_ONLY passback: the worker moves the extracted content out of + * {@code TIKA_CONTENT} into raw UTF-8 bytes on the {@code EmitData}, which travel + * as binary over the IPC instead of a Smile-encoded string. A caller who sets this + * must read {@code EmitData.getContentBytes()}; the metadata no longer carries the + * content. tika-server sets it for the raw-output endpoints. + */ +@TikaComponent(name = "content-bytes-config") +public class ContentBytesConfig implements Serializable { + + private static final long serialVersionUID = 1L; + + private boolean enabled = true; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } +} diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java index 5fec421209..ebe53de666 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/emitter/EmitDataImpl.java @@ -31,6 +31,9 @@ public class EmitDataImpl implements EmitData { // ParseContext is not serialized - it's set by PipesClient after deserialization private ParseContext parseContext; + // Raw UTF-8 content under content-bytes-config; rides the IPC as binary + private byte[] contentBytes; + public EmitDataImpl(String emitKey, List<Metadata> metadataList) { this(emitKey, metadataList, StringUtils.EMPTY); } @@ -54,8 +57,21 @@ public class EmitDataImpl implements EmitData { return containerStackTrace; } + @Override + public byte[] getContentBytes() { + return contentBytes; + } + + public void setContentBytes(byte[] contentBytes) { + this.contentBytes = contentBytes; + } + public long getEstimatedSizeBytes() { - return estimateSizeInBytes(getEmitKey(), getMetadataList(), containerStackTrace); + long sz = estimateSizeInBytes(getEmitKey(), getMetadataList(), containerStackTrace); + if (contentBytes != null) { + sz += 36 + contentBytes.length; + } + return sz; } /** @@ -97,6 +113,8 @@ public class EmitDataImpl implements EmitData { @Override public String toString() { return "EmitData{" + "emitKey=" + emitKey + ", metadataList=" + metadataList + - ", containerStackTrace='" + containerStackTrace + '\'' + '}'; + ", containerStackTrace='" + containerStackTrace + '\'' + + ", contentBytes.length=" + (contentBytes == null ? "null" : contentBytes.length) + + '}'; } } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataDeserializer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataDeserializer.java index 8d8c4e303d..1f63fa9325 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataDeserializer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataDeserializer.java @@ -46,7 +46,12 @@ public class EmitDataDeserializer extends JsonDeserializer<EmitDataImpl> { String containerStackTrace = readString(CONTAINER_STACK_TRACE, root, StringUtils.EMPTY, false); // ParseContext is NOT deserialized - it's restored by PipesClient from the original FetchEmitTuple - return new EmitDataImpl(emitKey, metadataList, containerStackTrace); + EmitDataImpl emitData = new EmitDataImpl(emitKey, metadataList, containerStackTrace); + JsonNode contentBytesNode = root.get(EmitDataSerializer.CONTENT_BYTES); + if (contentBytesNode != null && !contentBytesNode.isNull()) { + emitData.setContentBytes(contentBytesNode.binaryValue()); + } + return emitData; } private static List<Metadata> readMetadataList(JsonNode root, ObjectMapper mapper) throws IOException { diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataSerializer.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataSerializer.java index 5687088e55..5f7b87b397 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataSerializer.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/serialization/EmitDataSerializer.java @@ -30,6 +30,7 @@ public class EmitDataSerializer extends JsonSerializer<EmitData> { public static final String EMIT_KEY = "emitKey"; public static final String METADATA_LIST = "metadataList"; public static final String CONTAINER_STACK_TRACE = "containerStackTrace"; + public static final String CONTENT_BYTES = "contentBytes"; @Override public void serialize(EmitData emitData, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException { @@ -39,6 +40,11 @@ public class EmitDataSerializer extends JsonSerializer<EmitData> { if (!StringUtils.isBlank(emitData.getContainerStackTrace())) { jsonGenerator.writeStringField(CONTAINER_STACK_TRACE, emitData.getContainerStackTrace()); } + if (emitData.getContentBytes() != null) { + // Smile writes this as raw binary (7-bit encoding is disabled), so multi-MB + // content costs a copy, not a string transcode + jsonGenerator.writeBinaryField(CONTENT_BYTES, emitData.getContentBytes()); + } // ParseContext is NOT serialized - it's restored by PipesClient from the original FetchEmitTuple jsonGenerator.writeEndObject(); } diff --git a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java index f1fe8a7a8a..57bac19e86 100644 --- a/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java +++ b/tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java @@ -43,6 +43,7 @@ import org.apache.tika.pipes.api.PipesResult; import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.api.emitter.Emitter; import org.apache.tika.pipes.api.emitter.StreamEmitter; +import org.apache.tika.pipes.core.ContentBytesConfig; import org.apache.tika.pipes.core.EmitStrategy; import org.apache.tika.pipes.core.EmitStrategyConfig; import org.apache.tika.pipes.core.PassbackFilter; @@ -133,6 +134,9 @@ class EmitHandler { && emitterId != null && emitterManager.getAllIds().contains(emitterId); boolean willEmit = forceEmit || shouldEmit(parseMode, parseData, emitDataTuple, parseContext); + if (!willEmit) { + maybeMoveContentToBytes(parseMode, emitDataTuple, parseContext); + } if (willEmit) { return emit(t.getId(), emitKey, parseMode == ParseMode.UNPACK, parseData, stack, parseContext); @@ -148,6 +152,34 @@ class EmitHandler { } } + /** + * Under {@code content-bytes-config} + CONTENT_ONLY, moves the extracted content out of + * {@code TIKA_CONTENT} into raw UTF-8 bytes on the passback {@code EmitData}. One encode + * here replaces a Smile string encode, the client-side string decode, and the consumer's + * re-encode. Passback only -- the direct-emit path streams the string itself. + */ + private static void maybeMoveContentToBytes(ParseMode parseMode, EmitDataImpl emitData, + ParseContext parseContext) { + if (parseMode != ParseMode.CONTENT_ONLY) { + return; + } + ContentBytesConfig config = parseContext.get(ContentBytesConfig.class); + if (config == null || !config.isEnabled()) { + return; + } + List<Metadata> metadataList = emitData.getMetadataList(); + if (metadataList == null || metadataList.isEmpty()) { + return; + } + Metadata m = metadataList.get(0); + String content = m.get(TikaCoreProperties.TIKA_CONTENT); + if (content == null) { + return; + } + emitData.setContentBytes(content.getBytes(StandardCharsets.UTF_8)); + m.remove(TikaCoreProperties.TIKA_CONTENT); + } + private PipesResult emit(String taskId, EmitKey emitKey, boolean isExtractEmbeddedBytes, MetadataListAndEmbeddedBytes parseData, String parseExceptionStack, ParseContext parseContext) { diff --git a/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/EmitDataContentBytesTest.java b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/EmitDataContentBytesTest.java new file mode 100644 index 0000000000..1acb0bb035 --- /dev/null +++ b/tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/serialization/EmitDataContentBytesTest.java @@ -0,0 +1,58 @@ +/* + * 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.assertNull; + +import java.nio.charset.StandardCharsets; +import java.util.List; + +import org.junit.jupiter.api.Test; + +import org.apache.tika.metadata.Metadata; +import org.apache.tika.pipes.api.PipesResult; +import org.apache.tika.pipes.core.emitter.EmitDataImpl; + +public class EmitDataContentBytesTest { + + @Test + public void testContentBytesRoundTrip() throws Exception { + Metadata m = new Metadata(); + m.set("k", "v"); + EmitDataImpl emitData = new EmitDataImpl("key", List.of(m)); + byte[] content = "the quick UTF-8 café 中文".getBytes(StandardCharsets.UTF_8); + emitData.setContentBytes(content); + PipesResult result = new PipesResult(PipesResult.RESULT_STATUS.PARSE_SUCCESS, emitData); + + byte[] wire = JsonPipesIpc.toBytes(result); + PipesResult back = JsonPipesIpc.fromBytes(wire, PipesResult.class); + + assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS, back.status()); + assertArrayEquals(content, back.emitData().getContentBytes()); + assertEquals("v", back.emitData().getMetadataList().get(0).get("k")); + } + + @Test + public void testAbsentContentBytesStaysNull() throws Exception { + EmitDataImpl emitData = new EmitDataImpl("key", List.of(new Metadata())); + PipesResult result = new PipesResult(PipesResult.RESULT_STATUS.PARSE_SUCCESS, emitData); + PipesResult back = JsonPipesIpc.fromBytes(JsonPipesIpc.toBytes(result), PipesResult.class); + assertNull(back.emitData().getContentBytes()); + } +} diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java index b88cdc0bd9..967aabbca7 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/PipesParsingHelper.java @@ -45,6 +45,7 @@ import org.apache.tika.pipes.api.PipesResult; import org.apache.tika.pipes.api.emitter.EmitData; import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.api.fetcher.FetchKey; +import org.apache.tika.pipes.core.ContentBytesConfig; import org.apache.tika.pipes.core.EmitStrategy; import org.apache.tika.pipes.core.EmitStrategyConfig; import org.apache.tika.pipes.core.PipesConfig; @@ -179,6 +180,30 @@ public class PipesParsingHelper { */ public List<Metadata> parse(TikaInputStream tis, Metadata metadata, ParseContext parseContext, ParseMode parseMode) throws IOException { + return parseInternal(tis, metadata, parseContext, parseMode, false).metadataList(); + } + + /** + * The metadata plus, when requested via {@code content-bytes-config}, the extracted + * content as raw UTF-8 -- {@code TIKA_CONTENT} is then absent from the metadata. + */ + public record ParseOutput(List<Metadata> metadataList, byte[] contentBytes) { + } + + /** + * Like {@link #parse} but asks the worker for the content as raw UTF-8 bytes, which + * travel as binary over the IPC instead of a Smile-encoded string -- the win is one + * UTF-8 encode in the worker instead of a string transcode on both sides plus a + * re-encode at the HTTP layer. CONTENT_ONLY only. + */ + public ParseOutput parseContentOnlyToBytes(TikaInputStream tis, Metadata metadata, + ParseContext parseContext) throws IOException { + return parseInternal(tis, metadata, parseContext, ParseMode.CONTENT_ONLY, true); + } + + private ParseOutput parseInternal(TikaInputStream tis, Metadata metadata, + ParseContext parseContext, ParseMode parseMode, + boolean contentAsBytes) throws IOException { String requestId = UUID.randomUUID().toString(); PayloadRouter.Routed routed = null; String callerSuppliedName = metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY); @@ -209,6 +234,9 @@ public class PipesParsingHelper { // Set parse mode in context parseContext.set(ParseMode.class, parseMode); + if (contentAsBytes) { + parseContext.set(ContentBytesConfig.class, new ContentBytesConfig()); + } // This parser is shared with /pipes, whose own default is EMIT_ALL. No // emitter is configured for /tika/rmeta/unpack requests (EmitKey.NO_EMIT @@ -235,10 +263,12 @@ public class PipesParsingHelper { if (relativeName != null) { stripSpoolIdentity(metadataList, relativeName, callerSuppliedName); } + byte[] contentBytes = (contentAsBytes && result.emitData() != null) + ? result.emitData().getContentBytes() : null; postNanos = System.nanoTime() - postStart; logTiming(requestId, routed.route().name(), routeNanos, pipesNanos, postNanos, System.nanoTime() - entryNanos); - return metadataList; + return new ParseOutput(metadataList, contentBytes); } catch (InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaResource.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaResource.java index c590a183d6..eca7b938c4 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaResource.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaResource.java @@ -22,8 +22,6 @@ import static org.apache.tika.server.core.resource.RecursiveMetadataResource.HAN import java.io.IOException; import java.io.InputStream; -import java.io.OutputStreamWriter; -import java.io.Writer; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; @@ -404,6 +402,20 @@ public class TikaResource { * @return list of metadata objects from parsing * @throws IOException if parsing fails */ + private PipesParsingHelper.ParseOutput parseWithPipesRaw(TikaInputStream tis, + Metadata metadata, ParseContext parseContext) throws IOException { + if (pipesParsingHelper == null) { + throw new IllegalStateException("Pipes-based parsing is not enabled"); + } + String fileName = metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY); + long taskId = serverStatus.start(ServerStatus.TASK.PARSE, fileName); + try { + return pipesParsingHelper.parseContentOnlyToBytes(tis, metadata, parseContext); + } finally { + serverStatus.complete(taskId); + } + } + public List<Metadata> parseWithPipes(TikaInputStream tis, Metadata metadata, ParseContext parseContext, ParseMode parseMode) throws IOException { @@ -715,22 +727,26 @@ public class TikaResource { handlerTypeName, context.get(ContentHandlerFactory.class)); // Parse with pipes using CONTENT_ONLY mode - the metadata filter in - // EmitHandler will strip everything except tk:content - List<Metadata> metadataList = - parseWithPipes(tis, metadata, context, ParseMode.CONTENT_ONLY); + // EmitHandler will strip everything except tk:content, and the content comes + // back as raw UTF-8 bytes rather than a Smile-encoded string + PipesParsingHelper.ParseOutput parsed = + parseWithPipesRaw(tis, metadata, context); + List<Metadata> metadataList = parsed.metadataList(); LOG.debug("produceRawOutput: parseWithPipes returned {} metadata objects", metadataList.size()); // Extract content before checking for an exception -- content must not be // discarded just because a container-level exception also occurred. - String content = ""; + byte[] content = parsed.contentBytes(); boolean hasException = false; String exceptionMessage = null; if (!metadataList.isEmpty()) { - String extracted = metadataList.get(0).get(TikaCoreProperties.TIKA_CONTENT); - LOG.debug("produceRawOutput: TIKA_CONTENT length={}", extracted != null ? extracted.length() : 0); - if (extracted != null) { - content = extracted; + if (content == null) { + // fallback: results built without the byte path (crash/error metadata) + String extracted = metadataList.get(0).get(TikaCoreProperties.TIKA_CONTENT); + if (extracted != null) { + content = extracted.getBytes(UTF_8); + } } exceptionMessage = metadataList.get(0).get(TikaCoreProperties.CONTAINER_EXCEPTION); hasException = exceptionMessage != null && !exceptionMessage.isEmpty(); @@ -742,13 +758,11 @@ public class TikaResource { // 422 status signals the partial parse and the body carries the extracted content // only -- never the server-side exception/stack trace. Clients that need the // container exception should use /rmeta. - final String finalContent = content; + final byte[] finalContent = content == null ? new byte[0] : content; StreamingOutput streamingOutput = outputStream -> { - try (Writer writer = new OutputStreamWriter(outputStream, UTF_8)) { - writer.write(finalContent); - writer.flush(); - } + outputStream.write(finalContent); + outputStream.flush(); }; return Response.status(hasException ? 422 : Response.Status.OK.getStatusCode()) .entity(streamingOutput)
