krickert commented on code in PR #2961:
URL: https://github.com/apache/tika/pull/2961#discussion_r4355353006
##########
tika-grpc-api/README.md:
##########
@@ -38,7 +38,10 @@ metadata as a `fields` map of strings, is defined in the
`tika-grpc` module.
Tika metadata key can hold several values. The value type follows the
`Property`
type Tika declares for the key (integer, number, boolean or timestamp); keys
without a declared type are strings. For example, `pdf:charsPerPage` arrives
as
- one int64 per page. New or renamed keys need no change to the proto.
+ one int64 per page. Keys declared only in a parser module, such as
`csv:num-rows`,
+ get integer, number and boolean types only when the configuration sets
+ `"metadata-list": {"typedValues": true}`, which also types the JSON that
emitters
+ write; their timestamps stay strings. New or renamed keys need no change to
the proto.
Review Comment:
gRPC people would find this part very useful.
##########
tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java:
##########
@@ -360,65 +383,120 @@ private ParseContext
buildRequestParseContext(FetchAndParseRequest request,
return null;
}
}
- String additionalFetchConfigJson =
request.getAdditionalFetchConfigJson();
if (StringUtils.isNotBlank(additionalFetchConfigJson)) {
// The fork reads this jsonConfig by the fetcher's registered
component name
// (e.g. "http-fetcher"). An unregistered key fails the fork's
resolveAll and a
// wire-blocked one is refused by its restricted tuple
deserialization -- both
// exit the worker. Enforce here: a 400, not a JVM restart per
request.
- var info =
ComponentNameResolver.getComponentInfo(request.getFetcherId());
+ var info = ComponentNameResolver.getComponentInfo(fetcherId);
if (info.isEmpty() || ComponentNameResolver.isWireBlocked(
ComponentNameResolver.determineContextKey(info.get()))) {
responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT
.withDescription("additional_fetch_config_json
requires fetcher_id to "
+ "be a registered, wire-allowed component
name; got '"
- + request.getFetcherId() + "'")
+ + fetcherId + "'")
.asRuntimeException());
return null;
}
- parseContext.setJsonConfig(request.getFetcherId(),
additionalFetchConfigJson);
+ parseContext.setJsonConfig(fetcherId, additionalFetchConfigJson);
}
return parseContext;
}
private void fetchAndParseImpl(FetchAndParseRequest request, ParseContext
parseContext,
StreamObserver<FetchAndParseReply>
responseObserver) {
+ FetchParseOutcome outcome = runFetchAndParse(
+ request.getFetcherId(), request.getFetchKey(), parseContext);
+ if (outcome == null) {
Review Comment:
Same issue as the comment at line 410, fixed in ffe0add7.
##########
tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/transform/MetadataTagger.java:
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.grpc.mapper.transform;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+import com.google.protobuf.Timestamp;
+
+import org.apache.tika.grpc.v2.BooleanValues;
+import org.apache.tika.grpc.v2.Document;
+import org.apache.tika.grpc.v2.IntegerValues;
+import org.apache.tika.grpc.v2.MetadataField;
+import org.apache.tika.grpc.v2.MetadataValue;
+import org.apache.tika.grpc.v2.NumberValues;
+import org.apache.tika.grpc.v2.StringValues;
+import org.apache.tika.grpc.v2.TimestampValues;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.Property;
+import org.apache.tika.serialization.MetadataValueTypes;
+import org.apache.tika.utils.DateUtils;
+
+/**
+ * Builds the tagged metadata tail. A Tika metadata entry is an array of
values under
+ * one key ({@link Metadata} is String[]-backed), so every tagged value is a
typed
+ * ARRAY: the tag comes from Tika's <em>declared</em> {@link Property} value
type and
+ * applies to every element -- a declared integer sequence like {@code
pdf:charsPerPage}
+ * arrives as int64s, one per page, not as strings. Keys with no declared type
are
+ * emitted as strings, never guessed -- so we never turn "8 8 8 8" into a
broken
+ * integer. If any element refuses its declared type, the whole entry falls
back to
+ * strings: lossless beats typed when the two conflict.
+ */
+public final class MetadataTagger {
+
+ /**
+ * {@link Property#get(String)} can only see Property constants whose
declaring class
+ * has already been initialized by the JVM. In the gRPC server the parse
runs in a
+ * forked pipes process, so nothing in the server JVM necessarily touches
e.g.
+ * {@code org.apache.tika.metadata.PDF} before mapping -- and declared
types
+ * ({@code pdf:encrypted} is a boolean, ...) would silently degrade to
plain strings
+ * depending on class-loading order. Force-initialize the tika-core
property
+ * vocabularies once so tagging is deterministic.
+ */
+ private static final Class<?>[] PROPERTY_VOCABULARIES = {
+ org.apache.tika.metadata.AccessPermissions.class,
+ org.apache.tika.metadata.ClimateForecast.class,
+ org.apache.tika.metadata.Database.class,
+ org.apache.tika.metadata.DublinCore.class,
+ org.apache.tika.metadata.DWG.class,
+ org.apache.tika.metadata.Epub.class,
+ org.apache.tika.metadata.ExternalProcess.class,
+ org.apache.tika.metadata.FileSystem.class,
+ org.apache.tika.metadata.Font.class,
+ org.apache.tika.metadata.Geographic.class,
+ org.apache.tika.metadata.HTML.class,
+ org.apache.tika.metadata.HttpHeaders.class,
+ org.apache.tika.metadata.IPTC.class,
+ org.apache.tika.metadata.MachineMetadata.class,
+ org.apache.tika.metadata.MAPI.class,
+ org.apache.tika.metadata.Message.class,
+ org.apache.tika.metadata.Office.class,
+ org.apache.tika.metadata.OfficeOpenXMLCore.class,
+ org.apache.tika.metadata.OfficeOpenXMLExtended.class,
+ org.apache.tika.metadata.PageAnchoring.class,
+ org.apache.tika.metadata.PagedText.class,
+ org.apache.tika.metadata.PDF.class,
+ org.apache.tika.metadata.Photoshop.class,
+ org.apache.tika.metadata.PST.class,
+ org.apache.tika.metadata.QuattroPro.class,
+ org.apache.tika.metadata.Rendering.class,
+ org.apache.tika.metadata.RTFMetadata.class,
+ org.apache.tika.metadata.TIFF.class,
+ org.apache.tika.metadata.TikaCoreProperties.class,
+ org.apache.tika.metadata.TikaPagedText.class,
+ org.apache.tika.metadata.WARC.class,
+ org.apache.tika.metadata.WordPerfect.class,
+ org.apache.tika.metadata.XMP.class,
+ org.apache.tika.metadata.XMPDC.class,
+ org.apache.tika.metadata.XMPDM.class,
+ org.apache.tika.metadata.XMPIdq.class,
+ org.apache.tika.metadata.XMPMM.class,
+ org.apache.tika.metadata.XMPPDF.class,
+ org.apache.tika.metadata.XMPRights.class,
+ org.apache.tika.metadata.Zip.class,
+ };
+
+ static {
+ for (Class<?> vocabulary : PROPERTY_VOCABULARIES) {
+ try {
+ // A class literal alone does NOT initialize the class;
forName(…, true, …) does.
+ Class.forName(vocabulary.getName(), true,
vocabulary.getClassLoader());
+ } catch (ClassNotFoundException e) {
+ throw new ExceptionInInitializerError(e);
+ }
+ }
+ }
+
+ // The same date formats Metadata.getDate uses; DateUtils is not
thread-safe, so
+ // access is synchronized exactly as tika-core does (see TIKA-495).
+ private static final DateUtils DATE_UTILS = new DateUtils();
+
+ private MetadataTagger() {
+ }
+
+ /** Append every key not already consumed by a typed field into {@code
Document.extra}. */
+ public static void appendTail(Metadata tika, Set<String> consumed,
Document.Builder document) {
+ for (String key : tika.names()) {
+ if (consumed.contains(key)) {
+ continue;
+ }
+ MetadataValue value = tag(tika, key);
+ if (value != null) {
+
document.addExtra(MetadataField.newBuilder().setKey(key).setValue(value).build());
+ }
+ }
+ }
+
+ /**
+ * Tag one key's value array by its declared Property element type, or by
the type a pipes
+ * fork reported for a key declared only there; fall back to strings for
keys of unknown
+ * type or elements the type cannot parse.
+ */
+ public static MetadataValue tag(Metadata tika, String key) {
+ String[] raw = tika.getValues(key);
+ if (raw == null || raw.length == 0) {
+ return null;
+ }
+ List<String> values = new ArrayList<>(raw.length);
+ for (String v : raw) {
+ if (v != null && !v.trim().isEmpty()) {
+ values.add(v.trim());
+ }
+ }
+ if (values.isEmpty()) {
+ return null;
+ }
+ Property property = Property.get(key);
+ // a composite's own type is PROPERTY; tag by its primary's type
+ Property.ValueType type = property != null
+ ? property.getPrimaryProperty().getValueType()
+ : MetadataValueTypes.get(key);
+ if (type != null) {
+ MetadataValue typed = tagByDeclaredType(type, values);
+ if (typed != null) {
+ return typed;
+ }
+ }
+ StringValues.Builder strings = StringValues.newBuilder();
+ strings.addAllValues(values);
+ return MetadataValue.newBuilder().setStrings(strings).build();
+ }
+
+ private static MetadataValue tagByDeclaredType(Property.ValueType type,
List<String> values) {
+ switch (type) {
+ case INTEGER: {
+ IntegerValues.Builder integers = IntegerValues.newBuilder();
+ for (String v : values) {
+ try {
+ integers.addValues(Long.parseLong(v));
+ } catch (NumberFormatException ignored) {
+ return null;
+ }
+ }
+ return
MetadataValue.newBuilder().setIntegers(integers).build();
+ }
+ case REAL:
+ case RATIONAL: {
+ NumberValues.Builder numbers = NumberValues.newBuilder();
+ for (String v : values) {
+ try {
+ numbers.addValues(Double.parseDouble(v));
+ } catch (NumberFormatException ignored) {
+ return null;
+ }
+ }
+ return MetadataValue.newBuilder().setNumbers(numbers).build();
+ }
+ case BOOLEAN: {
+ BooleanValues.Builder booleans = BooleanValues.newBuilder();
+ for (String v : values) {
+ booleans.addValues(parseBoolean(v));
+ }
+ return
MetadataValue.newBuilder().setBooleans(booleans).build();
+ }
Review Comment:
Same issue as the comment at line 202, fixed in ffe0add7.
##########
tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java:
##########
@@ -360,65 +383,120 @@ private ParseContext
buildRequestParseContext(FetchAndParseRequest request,
return null;
}
}
- String additionalFetchConfigJson =
request.getAdditionalFetchConfigJson();
if (StringUtils.isNotBlank(additionalFetchConfigJson)) {
// The fork reads this jsonConfig by the fetcher's registered
component name
// (e.g. "http-fetcher"). An unregistered key fails the fork's
resolveAll and a
// wire-blocked one is refused by its restricted tuple
deserialization -- both
// exit the worker. Enforce here: a 400, not a JVM restart per
request.
- var info =
ComponentNameResolver.getComponentInfo(request.getFetcherId());
+ var info = ComponentNameResolver.getComponentInfo(fetcherId);
if (info.isEmpty() || ComponentNameResolver.isWireBlocked(
ComponentNameResolver.determineContextKey(info.get()))) {
responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT
.withDescription("additional_fetch_config_json
requires fetcher_id to "
+ "be a registered, wire-allowed component
name; got '"
- + request.getFetcherId() + "'")
+ + fetcherId + "'")
.asRuntimeException());
return null;
}
- parseContext.setJsonConfig(request.getFetcherId(),
additionalFetchConfigJson);
+ parseContext.setJsonConfig(fetcherId, additionalFetchConfigJson);
}
return parseContext;
}
private void fetchAndParseImpl(FetchAndParseRequest request, ParseContext
parseContext,
StreamObserver<FetchAndParseReply>
responseObserver) {
+ FetchParseOutcome outcome = runFetchAndParse(
+ request.getFetcherId(), request.getFetchKey(), parseContext);
+ if (outcome == null) {
+ return;
+ }
+ FetchAndParseReply.Builder fetchReplyBuilder =
+ FetchAndParseReply.newBuilder()
+ .setFetchKey(outcome.fetchKey())
+ .setStatus(outcome.status())
+ .putAllFields(outcome.fields());
+ if (outcome.errorMessage() != null) {
+ fetchReplyBuilder.setErrorMessage(outcome.errorMessage());
+ }
+ responseObserver.onNext(fetchReplyBuilder.build());
+ }
+
+ /**
+ * Shared pipes round-trip used by the v1 {@code fields}-map reply and the
v2 typed
+ * {@code Document} reply. {@code parseContext} is the per-request context
already
+ * validated by {@link #buildRequestParseContext}. Returns primary
metadata as
+ * {@code null} when the pipes result carried no metadata list (so the v2
builder can
+ * distinguish empty output from an empty {@link Metadata} object).
+ */
+ FetchParseOutcome runFetchAndParse(String fetcherId, String fetchKey,
+ ParseContext parseContext) {
Fetcher fetcher;
try {
- fetcher = fetcherManager.getFetcher(request.getFetcherId());
+ fetcher = fetcherManager.getFetcher(fetcherId);
} catch (TikaException | IOException e) {
- throw new RuntimeException("Could not find fetcher with name " +
request.getFetcherId(), e);
+ throw new RuntimeException("Could not find fetcher with name " +
fetcherId, e);
}
Metadata tikaMetadata = new Metadata();
+ // Times the whole pipesParser.parse() round trip: fetch and parse
both happen
+ // inside the forked pipes worker, so this is fetch+parse latency, not
parse-only.
+ long fetchParseStart = System.nanoTime();
try {
- PipesResult pipesResult = pipesParser.parse(new
FetchEmitTuple(request.getFetchKey(), new
FetchKey(fetcher.getExtensionConfig().id(), request.getFetchKey()),
- new EmitKey(), tikaMetadata, parseContext,
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
- FetchAndParseReply.Builder fetchReplyBuilder =
- FetchAndParseReply.newBuilder()
- .setFetchKey(request.getFetchKey())
- .setStatus(pipesResult.status().name());
- if
(pipesResult.status().equals(PipesResult.RESULT_STATUS.FETCH_EXCEPTION)) {
- fetchReplyBuilder.setErrorMessage(pipesResult.message());
- }
+ PipesResult pipesResult = pipesParser.parse(new FetchEmitTuple(
+ fetchKey,
+ new FetchKey(fetcher.getExtensionConfig().id(), fetchKey),
+ new EmitKey(),
+ tikaMetadata,
+ parseContext,
+ FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
+ long fetchParseTimeMs = (System.nanoTime() - fetchParseStart) /
1_000_000L;
+ Map<String, String> fields = new LinkedHashMap<>();
+ Metadata primary = null;
if (pipesResult.emitData() != null &&
pipesResult.emitData().getMetadataList() != null) {
for (Metadata metadata :
pipesResult.emitData().getMetadataList()) {
for (String name : metadata.names()) {
String value = metadata.get(name);
if (value != null) {
- fetchReplyBuilder.putFields(name, value);
+ fields.put(name, value);
}
}
}
+ if (!pipesResult.emitData().getMetadataList().isEmpty()) {
+ primary = pipesResult.emitData().getMetadataList().get(0);
+ }
}
- responseObserver.onNext(fetchReplyBuilder.build());
+ String errorMessage = null;
+ if
(pipesResult.status().equals(PipesResult.RESULT_STATUS.FETCH_EXCEPTION)) {
+ errorMessage = pipesResult.message();
+ }
+ return new FetchParseOutcome(
+ fetchKey,
+ pipesResult.status().name(),
+ errorMessage,
+ fields,
+ primary,
+ fetchParseTimeMs);
} catch (IOException | PipesException e) {
throw new RuntimeException(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
+ return null;
}
}
+ /**
+ * Outcome of a single fetch+parse pipes round trip, shared by v1 and v2
reply builders.
+ * {@code null} means the calling thread was interrupted before a reply
could be built.
Review Comment:
Same issue as the comment at line 410, fixed in ffe0add7.
##########
tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java:
##########
@@ -638,6 +771,78 @@ public void onCompleted() {
}
}
+ /**
+ * The v2 server-streaming variant must close the call like the v1 one
above: one typed
+ * reply, no error, exactly one completion. Same in-process transport and
direct
+ * executor, so the counts are final when the stub call returns.
+ */
+ @Test
+ public void testV2ServerSideStreamingSendsTerminalSignal(Resources
resources) throws Exception {
+ String serverName = InProcessServerBuilder.generateName();
+ TikaGrpcServerImpl serviceImpl = newService(tikaConfigUnlocked);
+ Server server = InProcessServerBuilder
+ .forName(serverName)
+ .directExecutor()
+ .addService(serviceImpl)
+ .addService(new TikaGrpcV2ServerImpl(serviceImpl))
+ .build()
+ .start();
+ resources.register(server, Duration.ofSeconds(10));
+
+ ManagedChannel channel = InProcessChannelBuilder
+ .forName(serverName)
+ .directExecutor()
+ .build();
+ resources.register(channel, Duration.ofSeconds(10));
+ TikaV2Grpc.TikaV2Stub v2 = TikaV2Grpc.newStub(channel);
+
+ String fetcherId = createFetcherId(1);
+ String fetchKey = "tika4766-stream-" + UUID.randomUUID() + ".html";
+ File testFile = new File("target", fetchKey);
+ FileUtils.writeStringToFile(testFile,
+ "<html><body>terminal signal</body></html>",
StandardCharsets.UTF_8);
+
+ List<org.apache.tika.grpc.v2.FetchAndParseReply> replies =
+ Collections.synchronizedList(new ArrayList<>());
+ AtomicInteger errors = new AtomicInteger();
+ AtomicInteger completions = new AtomicInteger();
+ StreamObserver<org.apache.tika.grpc.v2.FetchAndParseReply> observer =
+ new StreamObserver<>() {
+ @Override
+ public void
onNext(org.apache.tika.grpc.v2.FetchAndParseReply reply) {
+ replies.add(reply);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ errors.incrementAndGet();
+ }
+
+ @Override
+ public void onCompleted() {
+ completions.incrementAndGet();
+ }
+ };
+
+ try {
+
v2.fetchAndParseServerSideStreaming(org.apache.tika.grpc.v2.FetchAndParseRequest
+ .newBuilder()
+ .setFetcherId(fetcherId)
+ .setFetchKey(fetchKey)
+ .build(), observer);
+
+ assertEquals(1, replies.size(), "one reply for one fetch key");
+ assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(),
+ replies.get(0).getDocument().getStatus().getPipesStatus(),
+ "the fixture must actually parse, or this test proves
nothing");
+ assertEquals(0, errors.get(), "no error on the happy path");
+ assertEquals(1, completions.get(),
+ "server streaming must send a terminal signal");
Review Comment:
This is a false positive - no change needed. The in-process server and
channel both use `directExecutor()`, so the handler and the observer callbacks
run inside the stub call and the counts are final when it returns.
`testServerSideStreamingSendsTerminalSignal` on main (TIKA-4804) relies on
the same thing.
##########
tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/DocumentBuilder.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * 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.grpc.mapper;
+
+import java.time.Instant;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.tika.Tika;
+import org.apache.tika.digest.DigestDef;
+import org.apache.tika.grpc.mapper.transform.DocumentTransformers;
+import org.apache.tika.grpc.v2.Document;
+import org.apache.tika.grpc.v2.ParseStatus;
+import org.apache.tika.grpc.v2.SourceOrigin;
+import org.apache.tika.metadata.HttpHeaders;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+
+/**
+ * Builds a {@link Document} from Tika's parse output: the envelope (content
type,
+ * origin, status), the typed Dublin Core metadata via {@link
DocumentTransformers},
+ * and the lossless tagged tail for everything else.
+ */
+public final class DocumentBuilder {
+
+ private static final DocumentTransformers TRANSFORMERS =
DocumentTransformers.defaults();
+
+ // The key the parse pipeline records the source digest under when a
digester is
+ // configured. Derived from DigestDef so it cannot drift from the
producer: a
+ // hand-rolled "...SHA256" copy of it did, silently, when the algorithm
keys moved
+ // to their Java names ("SHA-256").
+ private static final String SHA256_DIGEST_KEY =
+ new DigestDef(DigestDef.Algorithm.SHA256,
DigestDef.Encoding.HEX).metadataKey();
+
+ private DocumentBuilder() {
+ }
+
+ /**
+ * Maps Tika parse output to {@link Document}. Pass {@code primary ==
null} when the
+ * pipes result carried no metadata at all (e.g. a fetch failure, or a
crash before
+ * the parse produced anything): the returned Document then has only an id
and a
+ * status. The status still derives from {@code pipesStatus} -- {@code
EMPTY_OUTPUT}
+ * with no metadata is a legitimate success -- and an explicit error is
recorded for
+ * the non-success cases.
+ */
+ public static Document build(Metadata primary, String docId, String
pipesStatus,
+ long fetchParseTimeMs) {
+ ParseStatus.Builder status = ParseStatus.newBuilder()
+ .setStatus(mapPipesStatus(pipesStatus))
+ .setFetchParseTimeMs(fetchParseTimeMs)
+ .setTikaVersion(Tika.getString());
+ if (pipesStatus != null && !pipesStatus.isEmpty()) {
+ status.setPipesStatus(pipesStatus);
+ }
+
+ Document.Builder document = Document.newBuilder();
+ if (docId != null && !docId.isEmpty()) {
+ document.setId(docId);
+ }
+
+ if (primary == null) {
+ if (status.getStatus() != ParseStatus.Status.SUCCESS) {
+ status.addErrors("No metadata returned from parse");
+ }
+ return document.setStatus(status.build()).build();
+ }
+
+ // Keys the envelope maps below are consumed up front so the tagged
tail never
+ // carries them a second time. tk:content is consumed without a typed
home
+ // yet: the reply's fields map still carries the flat content, and the
structured
+ // content tree is a planned additive follow-up -- duplicating the
whole body
+ // into `extra` as a string would defeat both.
+ Set<String> consumed = new HashSet<>();
+ consumed.add(TikaCoreProperties.TIKA_CONTENT.getName());
+
+ String contentType = primary.get(HttpHeaders.CONTENT_TYPE);
+ if (contentType != null && !contentType.isBlank()) {
+ document.setContentType(contentType.trim());
+ consumed.add(HttpHeaders.CONTENT_TYPE.getName());
+ }
+
+ Instant now = Instant.now();
+ document.setParsedAt(com.google.protobuf.Timestamp.newBuilder()
+ .setSeconds(now.getEpochSecond())
+ .setNanos(now.getNano())
+ .build());
+
+ SourceOrigin.Builder origin = SourceOrigin.newBuilder();
+ String resourceName =
primary.get(TikaCoreProperties.RESOURCE_NAME_KEY);
+ if (resourceName != null && !resourceName.isBlank()) {
+ origin.setFilename(resourceName.trim());
+ consumed.add(TikaCoreProperties.RESOURCE_NAME_KEY.getName());
+ }
+ String contentLength = primary.get(HttpHeaders.CONTENT_LENGTH);
+ if (contentLength != null && !contentLength.isBlank()) {
+ try {
+ origin.setByteSize(Long.parseLong(contentLength.trim()));
+ consumed.add(HttpHeaders.CONTENT_LENGTH.getName());
+ } catch (NumberFormatException ignored) {
+ // leave unconsumed; falls through to the tagged tail
+ }
+ }
+ String parserClass = primary.get(TikaCoreProperties.TIKA_PARSED_BY);
+ if (parserClass != null && !parserClass.isBlank()) {
+ origin.setParser(parserClass.trim());
+ }
+ String sha256 = primary.get(SHA256_DIGEST_KEY);
+ if (sha256 != null && !sha256.isBlank()) {
+ origin.setSha256(sha256.trim());
+ consumed.add(SHA256_DIGEST_KEY);
+ }
+ document.setOrigin(origin.build());
+
+ String[] parsedByFull =
primary.getValues(TikaCoreProperties.TIKA_PARSED_BY_FULL_SET);
+ if (parsedByFull == null || parsedByFull.length == 0) {
+ parsedByFull =
primary.getValues(TikaCoreProperties.TIKA_PARSED_BY);
+ }
+ for (String parser : parsedByFull) {
+ if (parser != null && !parser.isBlank()) {
+ status.addParsersUsed(parser.trim());
+ }
+ }
+ consumed.add(TikaCoreProperties.TIKA_PARSED_BY.getName());
+ consumed.add(TikaCoreProperties.TIKA_PARSED_BY_FULL_SET.getName());
+ document.setStatus(status.build());
+
+ TRANSFORMERS.transform(primary, document, consumed);
+
+ return document.build();
+ }
+
+ /**
+ * Maps {@code org.apache.tika.pipes.api.PipesResult.RESULT_STATUS.name()}
to
+ * {@link ParseStatus.Status}. Values are named after the real enum
constants (not
+ * guessed) since this module intentionally has no compile dependency on
tika-pipes-api:
+ * a clean success (e.g. {@code PARSE_SUCCESS}, {@code EMIT_SUCCESS}) is
{@code SUCCESS};
+ * a success that happened alongside a caught exception (e.g.
+ * {@code PARSE_SUCCESS_WITH_EXCEPTION}) is {@code PARTIAL}; everything
else -- including
+ * process crashes like {@code TIMEOUT} and {@code OOM}, which are not
partial successes --
+ * is {@code FAILED}.
+ */
+ private static ParseStatus.Status mapPipesStatus(String pipesStatus) {
+ if (pipesStatus == null || pipesStatus.isEmpty()) {
+ return ParseStatus.Status.UNSPECIFIED;
+ }
+ return switch (pipesStatus) {
+ case "EMPTY_OUTPUT", "PARSE_SUCCESS", "EMIT_SUCCESS",
"EMIT_SUCCESS_PASSBACK" ->
+ ParseStatus.Status.SUCCESS;
+ case "PARSE_SUCCESS_WITH_EXCEPTION", "PARSE_EXCEPTION_NO_EMIT",
"EMIT_SUCCESS_PARSE_EXCEPTION" ->
+ ParseStatus.Status.PARTIAL;
+ default -> ParseStatus.Status.FAILED;
+ };
Review Comment:
@dpol1 found it fixed too
##########
tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/transform/MetadataTagger.java:
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.grpc.mapper.transform;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+import com.google.protobuf.Timestamp;
+
+import org.apache.tika.grpc.v2.BooleanValues;
+import org.apache.tika.grpc.v2.Document;
+import org.apache.tika.grpc.v2.IntegerValues;
+import org.apache.tika.grpc.v2.MetadataField;
+import org.apache.tika.grpc.v2.MetadataValue;
+import org.apache.tika.grpc.v2.NumberValues;
+import org.apache.tika.grpc.v2.StringValues;
+import org.apache.tika.grpc.v2.TimestampValues;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.Property;
+import org.apache.tika.serialization.MetadataValueTypes;
+import org.apache.tika.utils.DateUtils;
+
+/**
+ * Builds the tagged metadata tail. A Tika metadata entry is an array of
values under
+ * one key ({@link Metadata} is String[]-backed), so every tagged value is a
typed
+ * ARRAY: the tag comes from Tika's <em>declared</em> {@link Property} value
type and
+ * applies to every element -- a declared integer sequence like {@code
pdf:charsPerPage}
+ * arrives as int64s, one per page, not as strings. Keys with no declared type
are
+ * emitted as strings, never guessed -- so we never turn "8 8 8 8" into a
broken
+ * integer. If any element refuses its declared type, the whole entry falls
back to
+ * strings: lossless beats typed when the two conflict.
+ */
+public final class MetadataTagger {
+
+ /**
+ * {@link Property#get(String)} can only see Property constants whose
declaring class
+ * has already been initialized by the JVM. In the gRPC server the parse
runs in a
+ * forked pipes process, so nothing in the server JVM necessarily touches
e.g.
+ * {@code org.apache.tika.metadata.PDF} before mapping -- and declared
types
+ * ({@code pdf:encrypted} is a boolean, ...) would silently degrade to
plain strings
+ * depending on class-loading order. Force-initialize the tika-core
property
+ * vocabularies once so tagging is deterministic.
+ */
+ private static final Class<?>[] PROPERTY_VOCABULARIES = {
+ org.apache.tika.metadata.AccessPermissions.class,
+ org.apache.tika.metadata.ClimateForecast.class,
+ org.apache.tika.metadata.Database.class,
+ org.apache.tika.metadata.DublinCore.class,
+ org.apache.tika.metadata.DWG.class,
+ org.apache.tika.metadata.Epub.class,
+ org.apache.tika.metadata.ExternalProcess.class,
+ org.apache.tika.metadata.FileSystem.class,
+ org.apache.tika.metadata.Font.class,
+ org.apache.tika.metadata.Geographic.class,
+ org.apache.tika.metadata.HTML.class,
+ org.apache.tika.metadata.HttpHeaders.class,
+ org.apache.tika.metadata.IPTC.class,
+ org.apache.tika.metadata.MachineMetadata.class,
+ org.apache.tika.metadata.MAPI.class,
+ org.apache.tika.metadata.Message.class,
+ org.apache.tika.metadata.Office.class,
+ org.apache.tika.metadata.OfficeOpenXMLCore.class,
+ org.apache.tika.metadata.OfficeOpenXMLExtended.class,
+ org.apache.tika.metadata.PageAnchoring.class,
+ org.apache.tika.metadata.PagedText.class,
+ org.apache.tika.metadata.PDF.class,
+ org.apache.tika.metadata.Photoshop.class,
+ org.apache.tika.metadata.PST.class,
+ org.apache.tika.metadata.QuattroPro.class,
+ org.apache.tika.metadata.Rendering.class,
+ org.apache.tika.metadata.RTFMetadata.class,
+ org.apache.tika.metadata.TIFF.class,
+ org.apache.tika.metadata.TikaCoreProperties.class,
+ org.apache.tika.metadata.TikaPagedText.class,
+ org.apache.tika.metadata.WARC.class,
+ org.apache.tika.metadata.WordPerfect.class,
+ org.apache.tika.metadata.XMP.class,
+ org.apache.tika.metadata.XMPDC.class,
+ org.apache.tika.metadata.XMPDM.class,
+ org.apache.tika.metadata.XMPIdq.class,
+ org.apache.tika.metadata.XMPMM.class,
+ org.apache.tika.metadata.XMPPDF.class,
+ org.apache.tika.metadata.XMPRights.class,
+ org.apache.tika.metadata.Zip.class,
+ };
+
+ static {
+ for (Class<?> vocabulary : PROPERTY_VOCABULARIES) {
+ try {
+ // A class literal alone does NOT initialize the class;
forName(…, true, …) does.
+ Class.forName(vocabulary.getName(), true,
vocabulary.getClassLoader());
+ } catch (ClassNotFoundException e) {
+ throw new ExceptionInInitializerError(e);
+ }
+ }
+ }
+
+ // The same date formats Metadata.getDate uses; DateUtils is not
thread-safe, so
+ // access is synchronized exactly as tika-core does (see TIKA-495).
+ private static final DateUtils DATE_UTILS = new DateUtils();
+
+ private MetadataTagger() {
+ }
+
+ /** Append every key not already consumed by a typed field into {@code
Document.extra}. */
+ public static void appendTail(Metadata tika, Set<String> consumed,
Document.Builder document) {
+ for (String key : tika.names()) {
+ if (consumed.contains(key)) {
+ continue;
+ }
+ MetadataValue value = tag(tika, key);
+ if (value != null) {
+
document.addExtra(MetadataField.newBuilder().setKey(key).setValue(value).build());
+ }
+ }
+ }
+
+ /**
+ * Tag one key's value array by its declared Property element type, or by
the type a pipes
+ * fork reported for a key declared only there; fall back to strings for
keys of unknown
+ * type or elements the type cannot parse.
+ */
+ public static MetadataValue tag(Metadata tika, String key) {
+ String[] raw = tika.getValues(key);
+ if (raw == null || raw.length == 0) {
+ return null;
+ }
+ List<String> values = new ArrayList<>(raw.length);
+ for (String v : raw) {
+ if (v != null && !v.trim().isEmpty()) {
+ values.add(v.trim());
+ }
+ }
+ if (values.isEmpty()) {
+ return null;
+ }
+ Property property = Property.get(key);
+ // a composite's own type is PROPERTY; tag by its primary's type
+ Property.ValueType type = property != null
+ ? property.getPrimaryProperty().getValueType()
+ : MetadataValueTypes.get(key);
+ if (type != null) {
+ MetadataValue typed = tagByDeclaredType(type, values);
+ if (typed != null) {
+ return typed;
+ }
+ }
+ StringValues.Builder strings = StringValues.newBuilder();
+ strings.addAllValues(values);
+ return MetadataValue.newBuilder().setStrings(strings).build();
+ }
+
+ private static MetadataValue tagByDeclaredType(Property.ValueType type,
List<String> values) {
+ switch (type) {
+ case INTEGER: {
+ IntegerValues.Builder integers = IntegerValues.newBuilder();
+ for (String v : values) {
+ try {
+ integers.addValues(Long.parseLong(v));
+ } catch (NumberFormatException ignored) {
+ return null;
+ }
+ }
+ return
MetadataValue.newBuilder().setIntegers(integers).build();
+ }
+ case REAL:
+ case RATIONAL: {
+ NumberValues.Builder numbers = NumberValues.newBuilder();
+ for (String v : values) {
+ try {
+ numbers.addValues(Double.parseDouble(v));
+ } catch (NumberFormatException ignored) {
+ return null;
+ }
+ }
+ return MetadataValue.newBuilder().setNumbers(numbers).build();
+ }
+ case BOOLEAN: {
+ BooleanValues.Builder booleans = BooleanValues.newBuilder();
+ for (String v : values) {
+ booleans.addValues(parseBoolean(v));
+ }
+ return
MetadataValue.newBuilder().setBooleans(booleans).build();
+ }
+ case DATE: {
+ TimestampValues.Builder timestamps =
TimestampValues.newBuilder();
+ for (String v : values) {
+ Date date = parseDate(v);
+ if (date == null) {
+ return null;
+ }
+ timestamps.addValues(toTimestamp(date));
+ }
+ return
MetadataValue.newBuilder().setTimestamps(timestamps).build();
+ }
+ default:
+ return null; // not a declared scalar type -> caller falls
back to strings
+ }
+ }
+
+ private static synchronized Date parseDate(String value) {
+ return DATE_UTILS.tryToParse(value);
+ }
+
+ static boolean parseBoolean(String v) {
+ String s = v.trim().toLowerCase(Locale.ROOT);
+ return "true".equals(s) || "yes".equals(s) || "1".equals(s);
+ }
Review Comment:
Same issue as the comment at line 202, fixed in ffe0add7.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
