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 3a9f5e4da7 TIKA-4989: file-system emitter can gzip its output;
tika-eval finds .… (#3384)
3a9f5e4da7 is described below
commit 3a9f5e4da7181cd37c8fde05b9aa06bd257107fa
Author: Tim Allison <[email protected]>
AuthorDate: Wed Oct 21 14:57:26 2026 -0400
TIKA-4989: file-system emitter can gzip its output; tika-eval finds .…
(#3384)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
CHANGES.txt | 4 +
.../ROOT/pages/pipes/plugins/filesystem.adoc | 4 +
.../org/apache/tika/eval/app/EvalFilePaths.java | 4 +-
.../org/apache/tika/eval/app/ProfilerBase.java | 7 +-
.../org/apache/tika/eval/app/io/ExtractReader.java | 32 ++++++-
.../tika/eval/app/ComparerGzipExtractsTest.java | 97 ++++++++++++++++++++++
.../tika/pipes/emitter/fs/FileSystemEmitter.java | 92 +++++++++++++++++---
.../pipes/emitter/fs/FileSystemEmitterConfig.java | 18 +++-
.../pipes/emitter/fs/FileSystemEmitterTest.java | 90 ++++++++++++++++++++
9 files changed, 328 insertions(+), 20 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 8a15e0a123..554ada5072 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,9 @@
Release 4.2.0 - unreleased
+ * The file-system emitter can gzip its output ("compression": "GZIP"), and
tika-eval
+ finds gzipped extracts (.json.gz) when comparing against a plain extract
set, measuring
+ their extract lengths and -m/-n limits on uncompressed size (TIKA-4989).
+
* tika-server activates catalog presets from the command line (--preset
<name>, repeatable)
and the environment (TIKA_PRESETS, comma-separated), the equivalent of
"<name>": true in
the config's presets, overriding the file's own entry for the name; a
config file needs
diff --git a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
b/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
index c5000a594a..e1ec32d54a 100644
--- a/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
+++ b/docs/modules/ROOT/pages/pipes/plugins/filesystem.adoc
@@ -147,6 +147,10 @@ Writes parsed results as files under `basePath`. The
relative output path is der
|`atomicWrites`
|`true`
|Write each output to a sibling `<name>.<uuid>.tmp` and rename it into place.
A reader of the output directory never sees a partial file, but it does see the
tmp files: crawlers and watchers over the output directory must ignore `*.tmp`.
A worker killed mid-write leaves its `.tmp` behind, and nothing sweeps them.
The `onExists: EXCEPTION`/`SKIP` check runs immediately before the rename and
is best-effort against two workers emitting the same key concurrently (last
writer wins) -- give wo [...]
+
+|`compression`
+|`NONE`
+|`GZIP` gzips each output and appends `.gz` to its name, after `fileExtension`
(`a.pdf.json.gz`). Applies where `fileExtension` does: the JSON metadata list,
and the text of `CONTENT_ONLY`; unpacked files are written as they are.
Extracted text typically shrinks 5-8x, and tika-eval reads `.json.gz` extracts
directly. Init-time only.
|===
[#file-system-iterator]
diff --git
a/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/EvalFilePaths.java
b/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/EvalFilePaths.java
index 8ac28e6ae6..e7e0c81dbe 100644
---
a/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/EvalFilePaths.java
+++
b/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/EvalFilePaths.java
@@ -23,6 +23,8 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Objects;
+import org.apache.tika.eval.app.io.ExtractReader;
+
/**
* Simple struct to keep track of relative path of source file (
* original binary file, e.g. /subdir/document1.doc)
@@ -45,7 +47,7 @@ class EvalFilePaths {
public EvalFilePaths(Path relativeSourceFilePath, Path extractFile) {
if (extractFile != null && Files.isRegularFile(extractFile)) {
try {
- extractFileLength = Files.size(extractFile);
+ extractFileLength = ExtractReader.extractLength(extractFile);
} catch (IOException e) {
//swallow ?
}
diff --git
a/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/ProfilerBase.java
b/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/ProfilerBase.java
index 7e68ad7e49..9ee2e01010 100644
---
a/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/ProfilerBase.java
+++
b/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/ProfilerBase.java
@@ -43,6 +43,7 @@ import org.xml.sax.SAXException;
import org.apache.tika.eval.app.db.ColInfo;
import org.apache.tika.eval.app.db.Cols;
import org.apache.tika.eval.app.db.TableInfo;
+import org.apache.tika.eval.app.io.ExtractReader;
import org.apache.tika.eval.app.io.ExtractReaderException;
import org.apache.tika.eval.app.io.IDBWriter;
import org.apache.tika.eval.app.io.PipesReport;
@@ -93,7 +94,7 @@ public abstract class ProfilerBase {
private static final String CONTAINER_EXCEPTION_1X = "X-TIKA" +
":EXCEPTION:runtime";
private static final Logger LOG =
LoggerFactory.getLogger(ProfilerBase.class);
private static final String[] EXTRACT_EXTENSIONS = {".json", ".txt", ""};
- private static final String[] COMPRESSION_EXTENSIONS = {"", ".bz2",
".gzip", ".zip",};
+ private static final String[] COMPRESSION_EXTENSIONS = {"", ".gz", ".bz2",
".gzip", ".zip",};
private static final String ZERO = "0";
private static final String UNKNOWN_EXTENSION = "unk";
//make this configurable
@@ -111,7 +112,7 @@ public abstract class ProfilerBase {
new ColInfo(Cols.BASE_MIME, Types.VARCHAR, 256),
new ColInfo(Cols.FILE_EXTENSION, Types.VARCHAR, 12));
private static CommonTokenCountManager COMMON_TOKEN_COUNT_MANAGER;
- private static Pattern FILE_NAME_CLEANER =
Pattern.compile("\\.(json|txt)(\\.(bz2|gz|zip))?$");
+ private static Pattern FILE_NAME_CLEANER =
Pattern.compile("\\.(json|txt)(\\.(bz2|gz|gzip|zip))?$");
private static LanguageIDWrapper LANG_ID = new LanguageIDWrapper();
protected IDBWriter writer;
AnalyzerManager analyzerManager;
@@ -867,7 +868,7 @@ public abstract class ProfilerBase {
protected long getFileLength(Path p) {
if (p != null && Files.isRegularFile(p)) {
try {
- return Files.size(p);
+ return ExtractReader.extractLength(p);
} catch (IOException e) {
//swallow
}
diff --git
a/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/io/ExtractReader.java
b/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/io/ExtractReader.java
index cfd2bc3db3..2f7436d76b 100644
---
a/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/io/ExtractReader.java
+++
b/tika-eval/tika-eval-app/src/main/java/org/apache/tika/eval/app/io/ExtractReader.java
@@ -21,9 +21,13 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -93,6 +97,32 @@ public class ExtractReader {
return fileSuffixes;
}
+ /**
+ * Uncompressed length for gzip (from the ISIZE trailer), so plain and
gzipped extract sets
+ * agree on lengths and min/max filtering; on-disk length for everything
else.
+ */
+ public static long extractLength(Path extractFile) throws IOException {
+ long onDisk = Files.size(extractFile);
+ String name =
extractFile.getFileName().toString().toLowerCase(Locale.ROOT);
+ if (onDisk < 18 || !(name.endsWith(".gz") || name.endsWith(".gzip"))) {
+ return onDisk;
+ }
+ ByteBuffer trailer =
ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN);
+ try (FileChannel ch = FileChannel.open(extractFile,
StandardOpenOption.READ)) {
+ while (trailer.hasRemaining()) {
+ if (ch.read(trailer, onDisk - 4 + trailer.position()) < 0) {
+ return onDisk;
+ }
+ }
+ }
+ long isize = trailer.getInt(0) & 0xFFFFFFFFL;
+ // smaller than even stored blocks could yield: wrapped past 4 GiB or
multi-member
+ if (isize + (onDisk >> 10) + 64 < onDisk) {
+ return onDisk;
+ }
+ return isize;
+ }
+
public List<Metadata> loadExtract(Path extractFile) throws
ExtractReaderException {
List<Metadata> metadataList = null;
@@ -112,7 +142,7 @@ public class ExtractReader {
long length = -1L;
try {
- length = Files.size(extractFile);
+ length = extractLength(extractFile);
} catch (IOException e) {
throw new
ExtractReaderException(ExtractReaderException.TYPE.IO_EXCEPTION, e);
}
diff --git
a/tika-eval/tika-eval-app/src/test/java/org/apache/tika/eval/app/ComparerGzipExtractsTest.java
b/tika-eval/tika-eval-app/src/test/java/org/apache/tika/eval/app/ComparerGzipExtractsTest.java
new file mode 100644
index 0000000000..9cb7540c36
--- /dev/null
+++
b/tika-eval/tika-eval-app/src/test/java/org/apache/tika/eval/app/ComparerGzipExtractsTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.eval.app;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+import java.util.zip.GZIPOutputStream;
+
+import org.apache.commons.io.FileUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.eval.app.db.H2Util;
+
+/** A gzipped extract set ({@code .json.gz}, as the file-system emitter writes
it) compares like its plain twin. */
+public class ComparerGzipExtractsTest {
+
+ private static final String[] QUERIES = {
+ "select file_path || ' ' || extract_file_length_a || ' ' ||
extract_file_length_b"
+ + " from containers order by file_path",
+ "select count(1) from profiles_a",
+ "select count(1) from profiles_b",
+ "select count(1) from extract_exceptions_a",
+ "select count(1) from extract_exceptions_b",
+ "select count(1) from content_comparisons"
+ };
+
+ @TempDir
+ Path tmp;
+
+ @Test
+ public void testGzipOnEitherSide() throws Exception {
+ Path testDirs =
Paths.get(getClass().getResource("/test-dirs").toURI());
+ Path a = testDirs.resolve("extractsA");
+ Path b = testDirs.resolve("extractsB");
+ List<String> plain = compare("plain", a, b);
+ assertEquals(plain, compare("gzB", a, gzipCopy(b,
tmp.resolve("bGz"))), "gzipped B");
+ assertEquals(plain, compare("gzA", gzipCopy(a, tmp.resolve("aGz")),
b), "gzipped A");
+ }
+
+ private List<String> compare(String name, Path a, Path b) throws Exception
{
+ Path db = tmp.resolve(name);
+ ExtractComparerRunner.main(new String[]{
+ "-a", a.toString(), "-b", b.toString(), "-d",
db.toAbsolutePath().toString()
+ });
+ List<String> results = new ArrayList<>();
+ try (Connection c = new H2Util(db).getConnection(); Statement st =
c.createStatement()) {
+ for (String q : QUERIES) {
+ try (ResultSet rs = st.executeQuery(q)) {
+ while (rs.next()) {
+ results.add(q + " -> " + rs.getString(1));
+ }
+ }
+ }
+ }
+ return results;
+ }
+
+ private static Path gzipCopy(Path src, Path dest) throws IOException {
+ FileUtils.copyDirectory(src.toFile(), dest.toFile());
+ try (Stream<Path> files = Files.walk(dest)) {
+ for (Path p : files.filter(f ->
f.toString().endsWith(".json")).toList()) {
+ try (OutputStream os = new GZIPOutputStream(
+ Files.newOutputStream(p.resolveSibling(p.getFileName()
+ ".gz")))) {
+ Files.copy(p, os);
+ }
+ Files.delete(p);
+ }
+ }
+ return dest;
+ }
+}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java
index c5e3621113..82099f0205 100644
---
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java
@@ -16,18 +16,23 @@
*/
package org.apache.tika.pipes.emitter.fs;
+import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
import java.io.Writer;
import java.nio.charset.StandardCharsets;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
+import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.UUID;
+import java.util.zip.GZIPOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,6 +63,10 @@ public class FileSystemEmitter extends AbstractStreamEmitter
{
// in-progress writes; crawlers of the output dir should ignore these
static final String TMP_SUFFIX = ".tmp";
+ static final String GZIP_SUFFIX = ".gz";
+
+ private static final int GZIP_BUFFER_SIZE = 64 * 1024;
+
private static final Logger LOG =
LoggerFactory.getLogger(FileSystemEmitter.class);
public static FileSystemEmitter build(ExtensionConfig pluginConfig) throws
TikaConfigException, IOException {
@@ -109,6 +118,9 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
if (!StringUtils.isBlank(config.fileExtension())) {
emitKey += "." + config.fileExtension();
}
+ if (config.gzip()) {
+ emitKey += GZIP_SUFFIX;
+ }
if (config.basePath() != null) {
Path basePath = Paths.get(config.basePath());
@@ -138,8 +150,7 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
}
Path tmp = tmpFor(output);
try {
- try (Writer writer = Files.newBufferedWriter(tmp,
StandardCharsets.UTF_8,
- StandardOpenOption.CREATE_NEW)) {
+ try (Writer writer = newWriter(tmp, config.gzip(),
StandardOpenOption.CREATE_NEW)) {
JsonMetadataList.toJson(metadataList, writer,
config.prettyPrint());
}
publish(tmp, output, config.onExists());
@@ -152,19 +163,47 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
private static void writeInPlace(List<Metadata> metadataList, Path output,
FileSystemEmitterConfig config) throws
IOException {
if (config.onExists() == FileSystemEmitterConfig.ON_EXISTS.EXCEPTION) {
- try (Writer writer = Files.newBufferedWriter(output,
StandardCharsets.UTF_8,
- StandardOpenOption.CREATE_NEW)) {
+ try (Writer writer = newWriter(output, config.gzip(),
StandardOpenOption.CREATE_NEW)) {
JsonMetadataList.toJson(metadataList, writer,
config.prettyPrint());
} catch (FileAlreadyExistsException e) {
throw alreadyExistsException(output);
}
} else {
- try (Writer writer = Files.newBufferedWriter(output,
StandardCharsets.UTF_8)) {
+ try (Writer writer = newWriter(output, config.gzip())) {
JsonMetadataList.toJson(metadataList, writer,
config.prettyPrint());
}
}
}
+ private static Writer newWriter(Path path, boolean gzip, OpenOption...
options)
+ throws IOException {
+ return new BufferedWriter(new OutputStreamWriter(newOutputStream(path,
gzip, options),
+ StandardCharsets.UTF_8));
+ }
+
+ private static OutputStream newOutputStream(Path path, boolean gzip,
OpenOption... options)
+ throws IOException {
+ OutputStream os = Files.newOutputStream(path, options);
+ return gzip ? gzip(os) : os;
+ }
+
+ static OutputStream gzip(OutputStream os) throws IOException {
+ try {
+ return new GZIPOutputStream(os, GZIP_BUFFER_SIZE) {
+ // JDK 17's close() skips closing os when finish() throws
(e.g. disk full)
+ @Override
+ public void close() throws IOException {
+ try (os) {
+ super.close();
+ }
+ }
+ };
+ } catch (IOException e) {
+ os.close();
+ throw e;
+ }
+ }
+
private static Path tmpFor(Path output) {
// sibling so the rename stays on one filesystem (and therefore atomic)
return output.resolveSibling(output.getFileName() + "." +
UUID.randomUUID() + TMP_SUFFIX);
@@ -209,9 +248,16 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
// CONTENT_ONLY emit keys are the input's own name, so without the
extension the output
// collides with the input. UNPACK keys already carry the embedded
file's own name and
// extension, so appending there would rename every unpacked file.
- if (parseContext.get(ParseMode.class) == ParseMode.CONTENT_ONLY
- && !StringUtils.isBlank(config.fileExtension())) {
- emitKey += "." + config.fileExtension();
+ // compression follows the same rule: never gzip an unpacked file
under its own name
+ boolean gzip = false;
+ if (parseContext.get(ParseMode.class) == ParseMode.CONTENT_ONLY) {
+ if (!StringUtils.isBlank(config.fileExtension())) {
+ emitKey += "." + config.fileExtension();
+ }
+ if (config.gzip()) {
+ emitKey += GZIP_SUFFIX;
+ gzip = true;
+ }
}
if (config.basePath() != null) {
@@ -232,12 +278,18 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
return;
}
if (!config.atomicWrites()) {
- copyInPlace(inputStream, output, config.onExists());
+ copyInPlace(inputStream, output, config.onExists(), gzip);
return;
}
Path tmp = tmpFor(output);
try {
- Files.copy(inputStream, tmp);
+ if (gzip) {
+ try (OutputStream os = newOutputStream(tmp, true,
StandardOpenOption.CREATE_NEW)) {
+ inputStream.transferTo(os);
+ }
+ } else {
+ Files.copy(inputStream, tmp);
+ }
publish(tmp, output, config.onExists());
} finally {
Files.deleteIfExists(tmp);
@@ -245,7 +297,20 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
}
private static void copyInPlace(InputStream inputStream, Path output,
- FileSystemEmitterConfig.ON_EXISTS
onExists) throws IOException {
+ FileSystemEmitterConfig.ON_EXISTS
onExists, boolean gzip)
+ throws IOException {
+ if (gzip) {
+ OpenOption[] options = onExists ==
FileSystemEmitterConfig.ON_EXISTS.REPLACE
+ ? new OpenOption[0] : new
OpenOption[]{StandardOpenOption.CREATE_NEW};
+ try (OutputStream os = newOutputStream(output, true, options)) {
+ inputStream.transferTo(os);
+ } catch (FileAlreadyExistsException e) {
+ if (onExists == FileSystemEmitterConfig.ON_EXISTS.EXCEPTION) {
+ throw alreadyExistsException(output);
+ }
+ }
+ return;
+ }
if (onExists == FileSystemEmitterConfig.ON_EXISTS.REPLACE) {
Files.copy(inputStream, output,
StandardCopyOption.REPLACE_EXISTING);
return;
@@ -275,7 +340,8 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
if (parseContext.hasJsonConfig(configKey)) {
JsonConfig configJson = parseContext.getJsonConfig(configKey);
if (configJson != null) {
- // basePath, allowAbsolutePaths and atomicWrites are not
runtime keys; naming one fails
+ // basePath, allowAbsolutePaths, atomicWrites and compression
are not runtime keys;
+ // naming one fails
FileSystemEmitterRuntimeConfig runtimeConfig =
PluginJson.read(configJson,
FileSystemEmitterRuntimeConfig.class);
@@ -283,7 +349,7 @@ public class FileSystemEmitter extends
AbstractStreamEmitter {
// init-time allowAbsolutePaths -- neither may be changed at
runtime.
config = new
FileSystemEmitterConfig(fileSystemEmitterConfig.basePath(),
runtimeConfig.getFileExtension(), runtimeConfig.getOnExists(),
runtimeConfig.isPrettyPrint(),
fileSystemEmitterConfig.allowAbsolutePaths(),
- fileSystemEmitterConfig.atomicWrites());
+ fileSystemEmitterConfig.atomicWrites(),
fileSystemEmitterConfig.compression());
checkConfig(config);
}
}
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
index 5365c0e62c..26a62a2b92 100644
---
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java
@@ -20,13 +20,20 @@ import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.plugins.PluginJson;
public record FileSystemEmitterConfig(String basePath, String fileExtension,
ON_EXISTS onExists, boolean prettyPrint, boolean allowAbsolutePaths,
- Boolean atomicWrites) {
+ Boolean atomicWrites, COMPRESSION compression) {
enum ON_EXISTS {
SKIP, EXCEPTION, REPLACE
}
- /** onExists absent means EXCEPTION, atomicWrites absent means true -- the
documented defaults. */
+ enum COMPRESSION {
+ NONE, GZIP
+ }
+
+ /**
+ * onExists absent means EXCEPTION, atomicWrites absent means true,
compression absent means
+ * NONE -- the documented defaults.
+ */
public FileSystemEmitterConfig {
if (onExists == null) {
onExists = ON_EXISTS.EXCEPTION;
@@ -34,6 +41,13 @@ public record FileSystemEmitterConfig(String basePath,
String fileExtension, ON_
if (atomicWrites == null) {
atomicWrites = Boolean.TRUE;
}
+ if (compression == null) {
+ compression = COMPRESSION.NONE;
+ }
+ }
+
+ boolean gzip() {
+ return compression == COMPRESSION.GZIP;
}
public static FileSystemEmitterConfig load(final String json)
diff --git
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java
index 21500b40ce..668dbd8dd1 100644
---
a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java
+++
b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java
@@ -23,11 +23,16 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.ByteArrayInputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.Reader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.stream.Stream;
+import java.util.zip.GZIPInputStream;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -37,9 +42,11 @@ import org.junit.jupiter.api.io.TempDir;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.parser.ParseContext;
+import org.apache.tika.pipes.api.ParseMode;
import org.apache.tika.pipes.api.emitter.Emitter;
import org.apache.tika.pipes.api.emitter.StreamEmitter;
import org.apache.tika.plugins.ExtensionConfig;
+import org.apache.tika.serialization.JsonMetadataList;
public class FileSystemEmitterTest {
@@ -212,4 +219,87 @@ public class FileSystemEmitterTest {
assertTrue(Files.isRegularFile(basePath.resolve("b.json")));
assertEquals(0, tmpFiles(basePath));
}
+
+ private StreamEmitter createGzipEmitter(Path basePath, String onExists,
boolean atomicWrites)
+ throws TikaConfigException, IOException {
+ ObjectNode config = MAPPER.createObjectNode();
+ config.put("basePath", basePath.toAbsolutePath().toString());
+ config.put("fileExtension", "json");
+ config.put("onExists", onExists);
+ config.put("atomicWrites", atomicWrites);
+ config.put("compression", "GZIP");
+ ExtensionConfig pluginConfig = new ExtensionConfig("test", "test",
config.toString());
+ return (StreamEmitter) new
FileSystemEmitterFactory().buildExtension(pluginConfig);
+ }
+
+ private static String gunzip(Path p) throws IOException {
+ try (InputStream is = new GZIPInputStream(Files.newInputStream(p))) {
+ return new String(is.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ }
+
+ @Test
+ public void testGzipMetadataList() throws Exception {
+ for (boolean atomic : new boolean[]{true, false}) {
+ Path basePath = tempDir.resolve("gz-" + atomic);
+ StreamEmitter emitter = createGzipEmitter(basePath, "EXCEPTION",
atomic);
+ Metadata m = new Metadata();
+ m.set("k", "v");
+ emitter.emit("dir/a.pdf", List.of(m), new ParseContext());
+ Path out = basePath.resolve("dir/a.pdf.json.gz");
+ assertTrue(Files.isRegularFile(out), "atomicWrites=" + atomic);
+ assertFalse(Files.exists(basePath.resolve("dir/a.pdf.json")));
+ try (Reader r = new InputStreamReader(new
GZIPInputStream(Files.newInputStream(out)),
+ StandardCharsets.UTF_8)) {
+ assertEquals("v",
JsonMetadataList.fromJson(r).get(0).get("k"));
+ }
+ // onExists checks the name actually written, .gz included
+ assertThrows(IOException.class,
+ () -> emitter.emit("dir/a.pdf", List.of(m), new
ParseContext()));
+ assertEquals(0, tmpFiles(basePath.resolve("dir")));
+ }
+ }
+
+ @Test
+ public void testGzipContentOnlyButNotUnpack() throws Exception {
+ for (boolean atomic : new boolean[]{true, false}) {
+ Path basePath = tempDir.resolve("gz-stream-" + atomic);
+ StreamEmitter emitter = createGzipEmitter(basePath, "REPLACE",
atomic);
+ ParseContext contentOnly = new ParseContext();
+ contentOnly.set(ParseMode.class, ParseMode.CONTENT_ONLY);
+ emitter.emit("a.pdf", new
ByteArrayInputStream("text".getBytes(StandardCharsets.UTF_8)),
+ new Metadata(), contentOnly);
+ assertEquals("text", gunzip(basePath.resolve("a.pdf.json.gz")),
"atomicWrites=" + atomic);
+
+ // an unpacked file keeps its own name and bytes
+ emitter.emit("unpacked/image.png", new ByteArrayInputStream(new
byte[]{1, 2, 3}),
+ new Metadata(), new ParseContext());
+ assertEquals(3,
Files.size(basePath.resolve("unpacked/image.png")));
+ assertEquals(0, tmpFiles(basePath));
+ }
+ }
+
+ @Test
+ public void testGzipClosesFileWhenFinishFails() throws Exception {
+ boolean[] closed = {false};
+ // room for the 10-byte header only, so finish() hits the full disk
+ OutputStream full = new OutputStream() {
+ int written;
+
+ @Override
+ public void write(int b) throws IOException {
+ if (++written > 10) {
+ throw new IOException("No space left on device");
+ }
+ }
+
+ @Override
+ public void close() {
+ closed[0] = true;
+ }
+ };
+ OutputStream gz = FileSystemEmitter.gzip(full);
+ assertThrows(IOException.class, gz::close);
+ assertTrue(closed[0]);
+ }
}
