gortiz commented on code in PR #19898:
URL: https://github.com/apache/pinot/pull/19898#discussion_r4311439301
##########
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:
I agree failing fast is better than silently uploading whichever archive
comes last. Still, this is a backward-incompatible change. Jobs whose output
dir has the same archive name in two recursively listed subdirectories pushed
fine before and now fail. Could you add the `backward-incompat` label and call
it out at the top of the PR description, so it shows up clearly in the release
notes?
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/SegmentPushUtilsTest.java:
##########
@@ -503,8 +605,8 @@ public void testGetSegmentUriToTarPathMap()
String[] segmentFiles = new String[]{
outputDirURI.resolve("segment.tar.gz").toString(),
outputDirURI.resolve("stats_202201.tar.gz").toString(),
- outputDirURI.resolve("/2022/segment.tar.gz").toString(),
- outputDirURI.resolve("/2022/stats_202201.tar.gz").toString()
+ outputDirURI.resolve("/2022/segment_2022.tar.gz").toString(),
Review Comment:
This test used to cover `segment.tar.gz` and `2022/segment.tar.gz` side by
side, and that is the exact layout the new validation rejects. Rather than
renaming the files to dodge it, could you keep a case with the old layout and
assert that it now throws? That documents the behavior change in the test suite.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-standalone/src/test/java/org/apache/pinot/plugin/ingestion/batch/standalone/SegmentGenerationJobRunnerTest.java:
##########
@@ -227,6 +356,34 @@ public boolean accept(File dir, String name) {
}
}
+ /// Exercises the same sibling-writing unpacker as the Parquet and ORC
readers without another plugin dependency.
+ public static class UnpackingCSVRecordReader extends CSVRecordReader {
+ @Override
+ public void init(File dataFile, @Nullable Set<String> fieldsToRead,
@Nullable RecordReaderConfig config)
+ throws IOException {
+ super.init(RecordReaderUtils.unpackIfRequired(dataFile, "csv"),
fieldsToRead, config);
+ }
+ }
+
+ /// Records copies to verify that plain local inputs are borrowed.
+ public static class TrackingLocalPinotFS extends LocalPinotFS {
+ private static TrackingLocalPinotFS _instance;
Review Comment:
The static `_instance`, together with re-registering `"file"` in
`PinotFSFactory` in the `finally` block at line 104, mutates global state. If
another test in the same JVM uses `PinotFSFactory` concurrently, it can pick up
`TrackingLocalPinotFS` or lose its own registration. Could you inject the
tracking FS without the global registry, or at least mark this test
`singleThreaded`?
##########
README.md:
##########
@@ -58,6 +58,19 @@ For getting started guides, deployment recipes, tutorials,
and more, please visi
<img src="https://pinot.apache.org/static/images/hero_diagram.svg";
align="center" alt="Apache Pinot"/>
+Batch ingestion jobs can opt into Zstandard segment archives in standalone,
Spark, and Hadoop runners:
Review Comment:
The root README is the project landing page, so usage guides and upgrade
notes are easy to miss here and go stale quickly. Could you move this to
pinot-docs, under the batch ingestion job spec reference, and keep the upgrade
notes in the PR description so they land in the release notes?
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/TarCompressionUtils.java:
##########
@@ -123,8 +126,10 @@ public static void createCompressedTarFile(File[]
inputFiles, File outputFile, S
throws IOException {
try (OutputStream fileOut = Files.newOutputStream(outputFile.toPath());
BufferedOutputStream bufferedOut = new BufferedOutputStream(fileOut);
- OutputStream compressorOut =
COMPRESSOR_STREAM_FACTORY.createCompressorOutputStream(compressorName,
- bufferedOut); TarArchiveOutputStream tarOut = new
TarArchiveOutputStream(compressorOut)) {
+ OutputStream compressorOut =
CompressorStreamFactory.LZ4_FRAMED.equalsIgnoreCase(compressorName)
Review Comment:
This changes the writer for every LZ4 archive Pinot creates, not only batch
ingestion output. Any deployment that creates `.tar.lz4` archives, or uses
`.tar.compressed` with LZ4 as the default compressor, gets the new encoder. The
format stays readable by the Commons reader on all versions, and the interop
test covers that. But per the PR description the archives are about 22% larger,
which existing LZ4 users would pay in deep store size without opting in.
I think this PR bundles four independent changes: the zstd codec option,
this LZ4 writer swap, borrowing local inputs in the standalone runner, and
duplicate segment-name validation on push. Could you split at least the LZ4
change and the duplicate validation into their own PRs? That would let each get
its own release note and be reverted on its own if needed.
--
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]
![https://pinot.apache.org/static/images/hero_diagram.svg"](https://pinot.apache.org/static/images/hero_diagram.svg"){kind=link}