xiangfu0 commented on code in PR #19898:
URL: https://github.com/apache/pinot/pull/19898#discussion_r4288772858
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-standalone/src/main/java/org/apache/pinot/plugin/ingestion/batch/standalone/SegmentGenerationJobRunner.java:
##########
@@ -230,9 +232,17 @@ private void submitSegmentGenTask(File localTempDir, URI
inputFileURI, int seqId
File localOutputTempDir = new File(localTempDir, "output");
FileUtils.forceMkdir(localOutputTempDir);
- //copy input path to local
- File localInputDataFile = createLocalInputDateFile(inputFileURI,
localInputTempDir);
- _inputDirFS.copyToLocalFile(inputFileURI, localInputDataFile);
+ // Gzip readers can unpack beside the input, so keep those writes inside
an owned staging directory.
+ File inputFile = new File(URIUtils.decode(inputFileURI.getRawPath()));
+ boolean borrowedInputFile =
PinotFSFactory.isFileSystemInstanceOf(_inputDirFS, LocalPinotFS.class)
+ && !RecordReaderUtils.isGZippedFile(inputFile);
Review Comment:
Small robustness idea: `isGZippedFile` opens the file here, so if an input
disappears between listing and this call it throws a raw
`FileNotFoundException` instead of the clearer per-file failure the old copy
path gave. Optional — wrapping to attach the input URI would make that edge
easier to diagnose.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/spec/SegmentGenerationJobSpec.java:
##########
@@ -228,6 +232,17 @@ public void setSegmentCreationJobParallelism(int
segmentCreationJobParallelism)
_segmentCreationJobParallelism = segmentCreationJobParallelism;
}
+ public String getSegmentTarCompressionCodec() {
+ return _segmentTarCompressionCodec == null ? "gzip" :
_segmentTarCompressionCodec;
+ }
+
+ /// Selects the archive codec for standalone, Spark, and Hadoop segment
generation jobs.
+ public void setSegmentTarCompressionCodec(String segmentTarCompressionCodec)
{
+ Preconditions.checkArgument("gzip".equals(segmentTarCompressionCodec) ||
"zstd".equals(segmentTarCompressionCodec),
Review Comment:
Minor: the setter only allows `gzip`/`zstd`, but
`Constants.SEGMENT_TAR_FILE_EXTENSIONS` also maps `lz4`. That looks intentional
(consume existing LZ4 archives, don't produce them), so it's not a bug — just a
small asymmetry worth a one-line comment so a future reader doesn't try adding
`lz4` here expecting generation to work.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentPushUtils.java:
##########
@@ -462,15 +461,41 @@ private static File
getSegmentMetadataFile(SegmentGenerationJobSpec spec, PinotF
return generateSegmentMetadataFile(fileSystem, URI.create(tarFilePath));
}
- /// Segments stored in the deep store do not have the .tar.gz extension.
- private static String getSegmentName(String tarFilePath) {
+ /// Resolves and validates the segment name from a supported gzip, LZ4, or
Zstandard archive path.
+ public static String getSegmentNameFromTarFile(String tarFilePath) {
String fileName = new File(tarFilePath).getName();
- String segmentName = fileName.endsWith(Constants.TAR_GZ_FILE_EXT)
- ? fileName.substring(0, fileName.length() -
Constants.TAR_GZ_FILE_EXT.length()) : fileName;
+ Preconditions.checkArgument(Constants.isSegmentTarFile(fileName),
"Unsupported segment archive: %s", fileName);
+ String extension =
Constants.SEGMENT_TAR_FILE_EXTENSIONS.values().stream().filter(fileName::endsWith)
+ .findFirst().orElseThrow();
+ String segmentName = fileName.substring(0, fileName.length() -
extension.length());
SegmentNameUtils.validatePartialOrFullSegmentName(segmentName);
return segmentName;
}
+ /// Rejects ambiguous uploads before HTTP or segment replacement starts.
Changing archive codecs in a reused output
+ /// directory can otherwise upload an old and a new archive of the same
segment in an arbitrary order.
+ /// Extensionless deep-store segment paths remain supported for metadata
push.
+ public static void validateUniqueSegmentNames(Collection<String>
segmentPaths) {
Review Comment:
Nice guard. Just flagging it's a behavior change: push sets where two
archives resolve to the same segment name (e.g. same-named segments in
different sub-dirs of a recursively-listed output dir) now hard-fail where they
previously uploaded in arbitrary order. That old behavior was nondeterministic
so failing fast is the right call — worth a line in the release notes so an
upgrade doesn't surprise anyone.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/spec/SegmentGenerationJobSpec.java:
##########
@@ -58,6 +59,9 @@ public class SegmentGenerationJobSpec implements Serializable
{
/// Segment creation job parallelism.
private int _segmentCreationJobParallelism;
+ /// Codec for segment archives: `gzip` (default) or `zstd`. Metadata
sidecars always use gzip.
+ private String _segmentTarCompressionCodec = "gzip";
Review Comment:
Nit, not from this PR: this class is `Serializable` and reaches Spark
executors via Java serialization, yet none of the spec classes declare a
`serialVersionUID`. Since you're adding a field, declaring an explicit
`serialVersionUID` here would make mixed-version Spark driver/executor
deployments tolerant of this and future field additions.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/batch/spec/Constants.java:
##########
@@ -18,11 +18,23 @@
*/
package org.apache.pinot.spi.ingestion.batch.spec;
+import java.util.Map;
+
public class Constants {
private Constants() {
}
/// By default Pinot segments are compressed in 'tar.gz' format then pushed
to controller.
public static final String TAR_GZ_FILE_EXT = ".tar.gz";
public static final String METADATA_TAR_GZ_FILE_EXT = ".metadata.tar.gz";
+
+ /// Archive extensions understood by segment upload and metadata extraction,
including existing LZ4 archives.
+ public static final Map<String, String> SEGMENT_TAR_FILE_EXTENSIONS =
Review Comment:
Since `zstd` is opt-in and the default stays `gzip`, exposure is
operator-controlled, which is good. Worth a release-note caveat though: don't
emit `.tar.zst` until every server/controller/minion that untars these segments
is on a build with zstd support, otherwise extraction fails with a
commons-compress signature error.
--
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]
