mayankshriv commented on code in PR #19378:
URL: https://github.com/apache/pinot/pull/19378#discussion_r3936932773
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java:
##########
@@ -20,59 +20,36 @@
import com.google.protobuf.Descriptors;
import com.google.protobuf.ProtobufInternalUtils;
-import java.io.File;
-import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;
-import java.nio.file.Files;
-import java.nio.file.Path;
import org.apache.pinot.spi.filesystem.PinotFS;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class ProtoBufUtils {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ProtoBufUtils.class);
public static final String TMP_DIR_PREFIX = "pinot-protobuf";
public static final String PB_OUTER_CLASS_SUFFIX = "OuterClass";
private ProtoBufUtils() {
}
- public static File getFileCopiedToLocal(String filePath)
+ /// Reads the contents of a descriptor file (local or remote) into a byte
array. The file is read via
+ /// [PinotFS#open] and the stream is closed before returning - no temporary
files are created.
+ ///
+ /// @param descriptorFilePath URI string pointing to a `.desc` protobuf
descriptor file
+ /// @return the raw bytes of the descriptor file
+ public static byte[] readDescriptorFileBytes(String descriptorFilePath)
throws Exception {
- URI fileURI = URI.create(filePath);
+ URI fileURI = URI.create(descriptorFilePath);
String scheme = fileURI.getScheme();
if (scheme == null) {
scheme = PinotFSFactory.LOCAL_PINOT_FS_SCHEME;
}
- if (PinotFSFactory.isSchemeSupported(scheme)) {
- PinotFS pinotFS = PinotFSFactory.create(scheme);
- Path localTmpDir = Files.createTempDirectory(TMP_DIR_PREFIX +
System.currentTimeMillis());
- File localFile = createLocalFile(fileURI, localTmpDir.toFile());
- LOGGER.info("Copying protocol buffer jar/descriptor file from source: {}
to dst: {}", filePath,
- localFile.getAbsolutePath());
- pinotFS.copyToLocalFile(fileURI, localFile);
- return localFile;
- } else {
- throw new RuntimeException(String.format("Scheme: %s not supported in
PinotFSFactory"
- + " for protocol buffer jar/descriptor file: %s.", scheme,
filePath));
+ PinotFS pinotFS = PinotFSFactory.create(scheme);
+ try (InputStream in = pinotFS.open(fileURI)) {
+ return in.readAllBytes();
Review Comment:
Reopening this - I implemented it as suggested, but it turns out to be
unsafe on one of the two paths, so I need your call before changing it back.
`com.github.os72:protobuf-dynamic:1.0.1`, which `ProtoBufMessageDecoder`
uses:
```java
public static DynamicSchema parseFrom(InputStream schemaDescIn) throws ... {
try {
int len;
byte[] buf = new byte[4096];
ByteArrayOutputStream baos = new ByteArrayOutputStream();
while ((len = schemaDescIn.read(buf)) > 0) baos.write(buf, 0, len);
return parseFrom(baos.toByteArray());
} finally { schemaDescIn.close(); }
}
```
Two problems:
1. **The loop cannot distinguish a 0-return from EOF**, so it stops early
and parses a truncated descriptor. Truncation on a `FileDescriptorSet` message
boundary still parses successfully, yielding a descriptor missing message types
- after which `getMessageTypes().toArray()[0]` can select the wrong type and we
silently extract the wrong columns. Before this PR the stream was always a
`FileInputStream` over a local copy, which never returns 0, so the loop was
safe by construction. Reading via `PinotFS.open()` removes that guarantee (e.g.
GCS returns `Channels.newInputStream(...)`, and the JDK's `ChannelInputStream`
propagates a channel's 0 verbatim).
2. **It does not actually stream** - it accumulates the whole descriptor
into a `ByteArrayOutputStream` before parsing, so there is no allocation saving
on this path, and `ByteArrayOutputStream` doubling makes the transient
footprint worse than `readAllBytes()`.
`readAllBytes()` is not exposed to (1): it delegates to `readNBytes`, whose
outer `do { ... } while (n >= 0 && remaining > 0)` re-enters on a 0-return
rather than treating it as EOF. Measured with a stream that returns 0 once and
then 10000 bytes:
```
DynamicSchema loop = 0 bytes -> silently truncated
readAllBytes() = 10000 bytes -> correct
```
`ProtoBufRecordReader` is unaffected:
`FileDescriptorSet.parseFrom(InputStream)` genuinely streams via
`CodedInputStream` and rejects a 0-return outright (`#read(byte[]) returned
invalid result:`). Your suggestion is the right call there, so I would keep it.
There is also a knock-on effect for your other comment about proving the
consumers close the stream: because `DynamicSchema.parseFrom` closes the stream
itself, that assertion cannot distinguish "our try-with-resources closed it"
from "the library closed it". I confirmed by mutation that deleting the
decoder's try-with-resources leaves the test green. So on the decoder path,
parsing from the stream and proving the caller closes it are mutually exclusive.
Proposed: `readAllBytes()` + `DynamicSchema.parseFrom(byte[])` for the
decoder, keep streaming for the record reader, and note the asymmetry in a
comment.
Alternative if you would rather preserve streaming: wrap the stream in a
filter that retries on a 0-return. That fixes the truncation but leaves the
close assertion unprovable, and it is extra machinery to work around a
third-party loop. Note `protobuf-dynamic` 1.0.1 is the latest release and the
project is dormant, so there is no upstream fix to wait for.
Which would you prefer?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
