Copilot commented on code in PR #3699:
URL: https://github.com/apache/celeborn/pull/3699#discussion_r3386207518
##########
common/src/main/java/org/apache/celeborn/common/meta/ReduceFileMeta.java:
##########
@@ -43,6 +44,11 @@ public ReduceFileMeta(List<Long> chunkOffsets, long
chunkSize) {
this.chunkSize = chunkSize;
}
+ public ReduceFileMeta(List<Long> chunkOffsets, List<Boolean>
chunkCompressed) {
+ this.chunkOffsets = chunkOffsets;
+ this.chunkCompressed = chunkCompressed;
+ }
Review Comment:
This new constructor does not initialize `chunkSize` (or `nextBoundary`).
Callers that rely on `ReduceFileMeta` having a meaningful chunk size (e.g.,
code paths that use `addChunkOffset`, or any logic keyed off `chunkSize`) will
get the default `0`, which is inconsistent with other constructors. A concrete
fix is to either (a) add a constructor that accepts `chunkSize` alongside
`chunkCompressed`, or (b) delegate to an existing constructor and set
`chunkSize` appropriately so metadata remains complete.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -234,6 +234,15 @@ public FileInfo getSortedFileInfo(
targetBuffer);
} else {
DiskFileInfo diskFileInfo = ((DiskFileInfo) fileInfo);
+ if (diskFileInfo.isChunkCompressionEnabled()) {
+ // TODO this is yet to be implemented
+ // We can read the file one chunk at a time and store chunkid +
uncompressed offsets before
+ // writing
+ throw new IOException(
+ "Chunk compressed shuffle file is not supported to sort, file
path: "
+ + diskFileInfo.getFilePath()
+ + ". Disable celeborn.chunk.compression.enabled when using
with AQE skew join enabled");
+ }
Review Comment:
The new error message is actionable but a bit unclear/awkwardly phrased and
mixes responsibilities (worker-side sorter mentioning Spark AQE). Consider
rewording to (1) fix grammar ('not supported for sorting'), (2) clearly state
the limitation ('sorting requires random/sliced reads which are unsupported for
chunk-compressed files'), and (3) provide concise remediation ('set
celeborn.chunk.compression.enabled=false' and/or disable the feature that
triggers sorting) without tying it to a specific Spark feature name in the core
error.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/file/chunk/compressed/ChunkCompressedFileChannelWriter.java:
##########
@@ -0,0 +1,173 @@
+/*
+ * 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.celeborn.service.deploy.worker.file.chunk.compressed;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.github.luben.zstd.Zstd;
+import com.google.common.annotations.VisibleForTesting;
+import io.netty.buffer.CompositeByteBuf;
+
+import org.apache.celeborn.common.meta.DiskFileInfo;
+import org.apache.celeborn.common.meta.ReduceFileMeta;
+import org.apache.celeborn.common.util.FileChannelUtils;
+import org.apache.celeborn.service.deploy.worker.file.FileChannelWriter;
+
+public class ChunkCompressedFileChannelWriter extends FileChannelWriter {
+ private final FileChannel channel;
+ private final DiskFileInfo diskFileInfo;
+ private final int compressionLevel;
+ private final ChunkBufferPool chunkBufferPool;
+ private final ChunkBufferPool.BufferPair bufferPair;
+ private ByteBuffer chunkBuffer;
+ private ByteBuffer compressedChunkBuffer;
+ private final List<Long> chunkOffsets;
+ private final List<Boolean> chunkCompressed;
+ private final long chunkSize;
+ private boolean closed = false;
+
+ public ChunkCompressedFileChannelWriter(
+ DiskFileInfo diskFileInfo,
+ long chunkSize,
+ int compressionLevel,
+ ChunkBufferPool chunkBufferPool)
+ throws IOException {
+ this.diskFileInfo = diskFileInfo;
+ this.chunkSize = chunkSize;
+ channel =
FileChannelUtils.createWritableFileChannel(diskFileInfo.getFilePath());
+ this.compressionLevel = compressionLevel;
+ this.chunkBufferPool = chunkBufferPool;
+ bufferPair = chunkBufferPool.acquire(chunkSize);
+ chunkBuffer = bufferPair.chunkBuffer;
+ compressedChunkBuffer = bufferPair.compressedBuffer;
+ chunkOffsets = new ArrayList<>();
+ chunkOffsets.add(0L);
+ chunkCompressed = new ArrayList<>();
+ }
+
+ @Override
+ public void write(CompositeByteBuf buffer, boolean gatherApiEnabled) throws
IOException {
+ if (buffer.readableBytes() > chunkSize) {
+ // Flush any pending accumulated data before writing the large record so
file offsets
+ // remain consistent.
+ compressAndFlush();
+ flushLargeRecord(buffer);
+ return;
+ }
+
+ if (buffer.readableBytes() > chunkBuffer.remaining()) {
+ compressAndFlush();
+ }
+
+ ByteBuffer[] buffers = buffer.nioBuffers();
+ for (ByteBuffer byteBuffer : buffers) {
+ while (byteBuffer.hasRemaining()) {
+ chunkBuffer.put(byteBuffer);
+ }
+ }
Review Comment:
The inner `while (byteBuffer.hasRemaining())` loop is redundant because
`chunkBuffer.put(byteBuffer)` already consumes the source buffer until either
the source is exhausted or the destination is full. Since you already ensure
the total readable bytes fit before this loop, this can be simplified to a
single `chunkBuffer.put(byteBuffer)` per component, reducing overhead and
making the intent clearer.
##########
common/src/main/java/org/apache/celeborn/common/network/buffer/FileChunkBuffers.java:
##########
@@ -27,15 +27,25 @@
public class FileChunkBuffers extends ChunkBuffers {
private final File file;
private final TransportConf conf;
+ private final boolean isChunkCompressed;
public FileChunkBuffers(DiskFileInfo fileInfo, TransportConf conf) {
super(fileInfo.getReduceFileMeta());
+ isChunkCompressed = fileInfo.isChunkCompressionEnabled();
file = fileInfo.getFile();
this.conf = conf;
}
@Override
public ManagedBuffer chunk(int chunkIndex, int offset, int len) {
+ if (isChunkCompressed && (offset != 0 || len != Integer.MAX_VALUE)) {
+ throw new IllegalArgumentException(
+ "Sliced reads (offset="
+ + offset
+ + ", len="
+ + len
+ + ") are not supported for chunk-compressed files");
+ }
Review Comment:
This introduces a hard runtime failure for sliced reads on chunk-compressed
shuffle files. Since this can surface from normal reader behavior (range reads
/ retries / partial fetch), consider making the failure mode more
API-appropriate and easier to handle (e.g., a dedicated exception type or an
`IOException`/`UnsupportedOperationException` message that the client can map
to a clear user-facing error). Additionally, it would be more robust to prevent
issuing sliced reads in the first place when `chunkCompressionConfig.enabled`
is true (so the failure happens at request planning rather than deep in the
fetch path).
--
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]
