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 c309b2a3df TIKA-4901: don't inflate embedded doc twice for digesting
(#3191)
c309b2a3df is described below
commit c309b2a3dfa1323188eddd53adb4632791537883
Author: Tim Allison <[email protected]>
AuthorDate: Thu Sep 17 19:58:04 2026 -0400
TIKA-4901: don't inflate embedded doc twice for digesting (#3191)
---
CHANGES.txt | 3 +
.../apache/tika/digest/InputStreamDigester.java | 3 +
.../java/org/apache/tika/io/ReopenableSource.java | 58 ++++++++--
.../java/org/apache/tika/io/TikaInputSource.java | 16 +++
.../java/org/apache/tika/io/TikaInputStream.java | 17 +++
.../tika/digest/InputStreamDigesterTest.java | 39 +++++++
.../org/apache/tika/io/ReopenableSourceTest.java | 123 +++++++++++++++++++++
7 files changed, 252 insertions(+), 7 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index e72d60114c..010d7fe8b5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,8 @@
Release 4.1.0 - unreleased
+ * Digesting an embedded document no longer makes its source hand out the
+ bytes twice. (TIKA-4901).
+
* Enable audio/video chunking with ffmpeg and inference (TIKA-4900).
* PDF incremental-update scanning reads the file in blocks instead of one
diff --git
a/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
b/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
index ec46af2839..4773a74040 100644
--- a/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
+++ b/tika-core/src/main/java/org/apache/tika/digest/InputStreamDigester.java
@@ -121,6 +121,9 @@ public class InputStreamDigester implements Digester {
CacheMemoryBudget budget =
(parseContext == null) ? null :
parseContext.get(CacheMemoryBudget.class);
tis.enableRewind(budget);
+ // The parse re-reads what we are about to read. Where re-opening the
source means
+ // inflating an embedded entry again, hold it in memory instead; a
no-op for a file.
+ tis.tryRetainInMemory();
MessageDigest messageDigest = newMessageDigest();
byte[] buffer = new byte[8192];
diff --git a/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
b/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
index 61c8dcf181..ea6df9397e 100644
--- a/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
+++ b/tika-core/src/main/java/org/apache/tika/io/ReopenableSource.java
@@ -17,6 +17,7 @@
package org.apache.tika.io;
import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -81,11 +82,28 @@ class ReopenableSource extends InputStream implements
TikaInputSource {
private void ensureOpen() throws IOException {
if (currentStream == null) {
- currentStream = new BufferedInputStream(
- spilledPath != null ? Files.newInputStream(spilledPath) :
opener.get());
+ currentStream = openAt(position);
}
}
+ /**
+ * A fresh stream positioned at {@code at}. Served from the retained
content when there is
+ * some: re-opening this source can mean inflating a zip entry again,
which the page cache
+ * does nothing for.
+ */
+ private InputStream openAt(long at) throws IOException {
+ if (retainedBuffer != null && spilledPath == null) {
+ int from = (int) Math.min(at, retainedLength);
+ return new ByteArrayInputStream(retainedBuffer, from,
retainedLength - from);
+ }
+ InputStream in = new BufferedInputStream(
+ spilledPath != null ? Files.newInputStream(spilledPath) :
opener.get());
+ if (at > 0) {
+ IOUtils.skipFully(in, at);
+ }
+ return in;
+ }
+
@Override
public int read() throws IOException {
ensureOpen();
@@ -128,11 +146,7 @@ class ReopenableSource extends InputStream implements
TikaInputSource {
if (currentStream != null) {
currentStream.close();
}
- currentStream = new BufferedInputStream(
- spilledPath != null ? Files.newInputStream(spilledPath) :
opener.get());
- if (newPosition > 0) {
- IOUtils.skipFully(currentStream, newPosition);
- }
+ currentStream = openAt(newPosition);
this.position = newPosition;
}
@@ -190,6 +204,36 @@ class ReopenableSource extends InputStream implements
TikaInputSource {
}
}
+ /**
+ * Drains into memory up front so the caller's read, and every later one,
is served from
+ * there. Only worth it when the content certainly fits: a doomed attempt
reads as far as
+ * the budget allows and throws it away, and {@link
#getSeekableByteChannel()} would spill
+ * instead, which is what TIKA-4835 removed.
+ */
+ @Override
+ public boolean tryRetainInMemory() throws IOException {
+ if (retainedBuffer != null) {
+ return true;
+ }
+ if (spilledPath != null || closed || length < 0 || length >
MAX_ARRAY_SIZE) {
+ return false;
+ }
+ if (length > IN_MEMORY_FLOOR) {
+ if (budget == null ||
+ budget.getMaxBytes() - budget.getReservedBytes() < length
- IN_MEMORY_FLOOR) {
+ return false;
+ }
+ }
+ if (!tryBufferInMemory()) {
+ return false;
+ }
+ if (currentStream != null) {
+ currentStream.close();
+ currentStream = openAt(position);
+ }
+ return true;
+ }
+
@Override
public SeekableByteChannel getSeekableByteChannel() throws IOException {
if (retainedBuffer != null) {
diff --git a/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
b/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
index 380db7a638..ca70497b2a 100644
--- a/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
+++ b/tika-core/src/main/java/org/apache/tika/io/TikaInputSource.java
@@ -104,4 +104,20 @@ interface TikaInputSource extends Closeable {
* @throws IOException if the source is partially read and cannot be
rewound
*/
SeekableByteChannel getSeekableByteChannel() throws IOException;
+
+ /**
+ * Asks the source to hold its full content in memory, so that later reads
-- sequential
+ * or via {@link #getSeekableByteChannel()} -- are served from there
instead of going
+ * back to the underlying source. Re-opening is cheap for a file (the page
cache serves
+ * it) but not for a zip entry, where it means inflating the entry again.
+ * <p>
+ * Advisory and best effort: a source that would have to spill to disk, or
that cannot
+ * tell whether the content fits, returns false and keeps its current
behaviour. Callers
+ * must work the same either way. Does not change this source's read
position.
+ *
+ * @return true if the full content is now held in memory
+ */
+ default boolean tryRetainInMemory() throws IOException {
+ return false;
+ }
}
diff --git a/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
b/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
index 398bf53778..65ad794795 100644
--- a/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
+++ b/tika-core/src/main/java/org/apache/tika/io/TikaInputStream.java
@@ -593,6 +593,23 @@ public class TikaInputStream extends TaggedInputStream {
return source.getSeekableByteChannel();
}
+ /**
+ * Asks the underlying source to hold its full content in memory, so a
later rewind and
+ * re-read does not go back to the source. Worth calling before a pass
that will be
+ * followed by another (digest then parse) when re-opening is expensive --
a zip entry
+ * has to be inflated again, where a file is served by the page cache.
+ * <p>
+ * Advisory: sources that would have to spill, or cannot tell whether the
content fits,
+ * return false and behave as before. Does not change the read position.
+ *
+ * @return true if the full content is now held in memory
+ * @see TikaInputSource#tryRetainInMemory()
+ */
+ public boolean tryRetainInMemory() throws IOException {
+ TikaInputSource source = inputSource();
+ return source != null && source.tryRetainInMemory();
+ }
+
/**
* Zero-copy, read-only view of the content behind a channel from
* {@link #getSeekableByteChannel()}, or {@code null} when that content is
on disk. Lets a
diff --git
a/tika-core/src/test/java/org/apache/tika/digest/InputStreamDigesterTest.java
b/tika-core/src/test/java/org/apache/tika/digest/InputStreamDigesterTest.java
index be20831746..1d0b5d0d9b 100644
---
a/tika-core/src/test/java/org/apache/tika/digest/InputStreamDigesterTest.java
+++
b/tika-core/src/test/java/org/apache/tika/digest/InputStreamDigesterTest.java
@@ -16,14 +16,22 @@
*/
package org.apache.tika.digest;
+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 java.io.ByteArrayInputStream;
+import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.io.IOUtils;
import org.junit.jupiter.api.Test;
+import org.apache.tika.io.TemporaryResources;
import org.apache.tika.io.TikaInputStream;
+import org.apache.tika.metadata.HttpHeaders;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.Property;
import org.apache.tika.parser.ParseContext;
@@ -43,4 +51,35 @@ public class InputStreamDigesterTest {
assertNotNull(metadata.get(key), "trusted digest write must survive
the reserved-key guard");
assertNotNull(Property.get(key), "digest key must be registered as a
Property");
}
+
+ /**
+ * Digesting a re-openable source used to read it once for the hash and
leave the parse to
+ * read it again. For an embedded entry that second read means inflating
it again, so the
+ * digest now asks the source to hold the content.
+ */
+ @Test
+ public void testDigestOfAReopenableSourceDoesNotForceASecondOpen() throws
Exception {
+ byte[] content = new byte[64 * 1024];
+ for (int i = 0; i < content.length; i++) {
+ content[i] = (byte) (i * 31 + 7);
+ }
+ AtomicInteger opens = new AtomicInteger();
+ Metadata metadata = new Metadata();
+ metadata.set(HttpHeaders.CONTENT_LENGTH,
Integer.toString(content.length));
+ InputStreamDigester digester = new InputStreamDigester("SHA256",
"tk:digest:SHA256",
+ bytes -> Base64.getEncoder().encodeToString(bytes));
+ try (TemporaryResources tmp = new TemporaryResources()) {
+ java.util.function.Supplier<InputStream> raw = () -> {
+ opens.incrementAndGet();
+ return new ByteArrayInputStream(content);
+ };
+ try (TikaInputStream tis = TikaInputStream.get(raw::get, tmp,
metadata)) {
+ digester.digest(tis, metadata, new ParseContext());
+ assertNotNull(metadata.get("tk:digest:SHA256"));
+ // the parse that follows the digest
+ assertArrayEquals(content, IOUtils.toByteArray(tis));
+ }
+ }
+ assertEquals(1, opens.get(), "digest plus parse must open the source
once");
+ }
}
diff --git
a/tika-core/src/test/java/org/apache/tika/io/ReopenableSourceTest.java
b/tika-core/src/test/java/org/apache/tika/io/ReopenableSourceTest.java
index 35ab3027be..49bdc99443 100644
--- a/tika-core/src/test/java/org/apache/tika/io/ReopenableSourceTest.java
+++ b/tika-core/src/test/java/org/apache/tika/io/ReopenableSourceTest.java
@@ -62,6 +62,129 @@ public class ReopenableSourceTest {
};
}
+ private static byte[] readFully(InputStream in) throws IOException {
+ java.io.ByteArrayOutputStream out = new
java.io.ByteArrayOutputStream();
+ byte[] buf = new byte[4096];
+ int n;
+ while ((n = in.read(buf)) != -1) {
+ out.write(buf, 0, n);
+ }
+ return out.toByteArray();
+ }
+
+ @Test
+ public void testRetainInMemoryServesSequentialReads() throws Exception {
+ byte[] data = data(64 * 1024);
+ AtomicInteger opens = new AtomicInteger();
+ ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, ".bin");
+ try {
+ assertTrue(source.tryRetainInMemory());
+ assertEquals(1, opens.get());
+ assertArrayEquals(data, readFully(source));
+ source.seekTo(0);
+ assertArrayEquals(data, readFully(source));
+ assertEquals(1, opens.get(), "reads after retention must not
re-open the source");
+ assertFalse(source.hasPath(), "retention must not spill");
+ } finally {
+ source.close();
+ }
+ }
+
+ @Test
+ public void testRetainInMemoryIsIdempotent() throws Exception {
+ byte[] data = data(8192);
+ AtomicInteger opens = new AtomicInteger();
+ ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, ".bin");
+ try {
+ assertTrue(source.tryRetainInMemory());
+ assertTrue(source.tryRetainInMemory());
+ assertEquals(1, opens.get());
+ } finally {
+ source.close();
+ }
+ }
+
+ @Test
+ public void testRetainInMemoryRefusedWhenLengthUnknown() throws Exception {
+ byte[] data = data(8192);
+ AtomicInteger opens = new AtomicInteger();
+ ReopenableSource source = new ReopenableSource(countingOpener(data,
opens), tmp, -1, ".bin");
+ try {
+ assertFalse(source.tryRetainInMemory());
+ assertEquals(0, opens.get());
+ assertArrayEquals(data, readFully(source));
+ } finally {
+ source.close();
+ }
+ }
+
+ @Test
+ public void testRetainInMemoryRefusedOverFloorWithNoBudget() throws
Exception {
+ byte[] data = data(FLOOR + 4096);
+ AtomicInteger opens = new AtomicInteger();
+ ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, ".bin");
+ try {
+ assertFalse(source.tryRetainInMemory());
+ assertEquals(0, opens.get(), "a doomed attempt must not read the
source");
+ assertFalse(source.hasPath(), "a refusal must not spill");
+ assertArrayEquals(data, readFully(source));
+ } finally {
+ source.close();
+ }
+ }
+
+ @Test
+ public void testRetainInMemoryOverFloorWithBudget() throws Exception {
+ byte[] data = data(FLOOR + 4096);
+ AtomicInteger opens = new AtomicInteger();
+ ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, ".bin");
+ try {
+ source.enableRewind(new CacheMemoryBudget(8 * FLOOR));
+ assertTrue(source.tryRetainInMemory());
+ assertArrayEquals(data, readFully(source));
+ assertEquals(1, opens.get());
+ } finally {
+ source.close();
+ }
+ }
+
+ @Test
+ public void testRetainInMemoryRefusedWhenBudgetTooSmall() throws Exception
{
+ byte[] data = data(4 * FLOOR);
+ AtomicInteger opens = new AtomicInteger();
+ ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, ".bin");
+ try {
+ source.enableRewind(new CacheMemoryBudget(4096));
+ assertFalse(source.tryRetainInMemory());
+ assertEquals(0, opens.get());
+ assertFalse(source.hasPath());
+ assertArrayEquals(data, readFully(source));
+ } finally {
+ source.close();
+ }
+ }
+
+ @Test
+ public void testGetPathAfterRetainUsesRetainedBytes() throws Exception {
+ byte[] data = data(32 * 1024);
+ AtomicInteger opens = new AtomicInteger();
+ ReopenableSource source =
+ new ReopenableSource(countingOpener(data, opens), tmp,
data.length, ".bin");
+ try {
+ assertTrue(source.tryRetainInMemory());
+ Path path = source.getPath(null);
+ assertArrayEquals(data, java.nio.file.Files.readAllBytes(path));
+ assertEquals(1, opens.get());
+ } finally {
+ source.close();
+ }
+ }
+
private static byte[] readFully(SeekableByteChannel channel) throws
IOException {
ByteBuffer buf = ByteBuffer.allocate((int) channel.size());
while (buf.hasRemaining() && channel.read(buf) != -1) {
