chenboat commented on code in PR #12980:
URL: https://github.com/apache/pinot/pull/12980#discussion_r1585809421
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java:
##########
@@ -48,21 +51,156 @@ public static InputStream
getDescriptorFileInputStream(String descriptorFilePath
PinotFS pinotFS = PinotFSFactory.create(scheme);
Path localTmpDir = Files.createTempDirectory(TMP_DIR_PREFIX +
System.currentTimeMillis());
File protoDescriptorLocalFile = createLocalFile(descriptorFileURI,
localTmpDir.toFile());
- LOGGER.info("Copying protocol buffer descriptor file from source: {} to
dst: {}", descriptorFilePath,
+ LOGGER.info("Copying protocol buffer jar/descriptor file from source: {}
to dst: {}", filePath,
protoDescriptorLocalFile.getAbsolutePath());
pinotFS.copyToLocalFile(descriptorFileURI, protoDescriptorLocalFile);
- return new FileInputStream(protoDescriptorLocalFile);
+ return protoDescriptorLocalFile;
} else {
throw new RuntimeException(String.format("Scheme: %s not supported in
PinotFSFactory"
- + " for protocol buffer descriptor file: %s.", scheme,
descriptorFilePath));
+ + " for protocol buffer jar/descriptor file: %s.", scheme,
filePath));
}
}
+ public static InputStream getDescriptorFileInputStream(String
descriptorFilePath)
+ throws Exception {
+ return new FileInputStream(getFileCopiedToLocal(descriptorFilePath));
+ }
+
public static File createLocalFile(URI srcURI, File dstDir) {
String sourceURIPath = srcURI.getPath();
File dstFile = new File(dstDir, new File(sourceURIPath).getName());
LOGGER.debug("Created empty local temporary file {} to copy protocol "
+ "buffer descriptor {}", dstFile.getAbsolutePath(), srcURI);
return dstFile;
}
+
+ // This is needed since the generated class name is not always the same as
the proto file name.
+ // The descriptor that we get from the jar drops the first prefix of the
proto class name.
+ // For example, insead of com.data.example.ExampleProto we get
data.example.ExampleProto.
+ // Copied from Flink codebase.
Review Comment:
should we import/use Flink method instead of copy? Does Flink allow such
copy of their code?
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
