liuneng1994 commented on code in PR #13377:
URL: https://github.com/apache/gluten/pull/13377#discussion_r4345083563
##########
backends-bolt/src-celeborn/main/scala/org/apache/spark/shuffle/BoltCelebornColumnarBatchSerializer.scala:
##########
@@ -112,10 +112,10 @@ private class CelebornColumnarBatchSerializerInstance(
if (conf.getBoolean(SHUFFLE_COMPRESS.key,
SHUFFLE_COMPRESS.defaultValue.get)) {
GlutenShuffleUtils.getCompressionCodec(conf)
} else {
- null // uncompressed
+ "" // uncompressed
Review Comment:
Uncompressed is encoded as `""` here, while the Uniffle writer passes
`"uncompressed"` (`BoltUniffleColumnarShuffleWriter.java:144`). Both are RSS
paths for the same state, so please confirm the native reader/writer accept
both, or share one constant. A mismatch would only show up at runtime when
`spark.shuffle.compress=false`.
##########
backends-bolt/src-uniffle/main/java/org/apache/spark/shuffle/writer/BoltUniffleColumnarShuffleWriter.java:
##########
@@ -123,20 +120,50 @@ public BoltUniffleColumnarShuffleWriter(
sparkConf.getSizeAsBytes(
RssSparkConfig.RSS_WRITER_BUFFER_SIZE.key(),
RssSparkConfig.RSS_WRITER_BUFFER_SIZE.defaultValue().get());
- this.diskWriteBufferSize =
- (int) (long)
sparkConf.get(package$.MODULE$.SHUFFLE_DISK_WRITE_BUFFER_SIZE());
if ((boolean) sparkConf.get(package$.MODULE$.SHUFFLE_COMPRESS())) {
compressionCodec = GlutenShuffleUtils.getCompressionCodec(sparkConf);
compressionLevel = GlutenShuffleUtils.getCompressionLevel(sparkConf,
compressionCodec);
- compressionBufferSize =
- GlutenShuffleUtils.getCompressionBufferSize(sparkConf,
compressionCodec);
Option<String> codecBackend =
GlutenConfig.get().columnarShuffleCodecBackend();
if (codecBackend.isDefined()) {
this.codecBackend = codecBackend.get();
}
}
}
+ private ShuffleWriterInfo getShuffleWriterInfo() {
+ BoltConfig config = BoltConfig.get();
+ return ShuffleWriterInfo.newBuilder()
+ .setPartitioningName(columnarDep.nativePartitioning().getShortName())
+ .setNumPartitions(numPartitions)
+ .setStartPartitionId(
+
GlutenShuffleUtils.getStartPartitionId(columnarDep.nativePartitioning(),
partitionId))
+ .setTaskAttemptId(TaskContext.get().taskAttemptId())
+ .setBufferSize(nativeBufferSize)
+ .setReallocThreshold(reallocThreshold)
+ .setMemLimit(availableOffHeapPerTask())
+ .setCompressionCodec(compressionCodec == null ? "uncompressed" :
compressionCodec)
+ .setCompressionBackend(codecBackend == null ? "none" : codecBackend)
+ .setCompressionLevel(compressionLevel)
+ .setCompressionThreshold(config.columnarShuffleCompressionThreshold())
+ .setCompressionMode(config.columnarShuffleCompressionMode())
+ .setPushBufferMaxSize(bufferSize)
+ .setShuffleBatchByteSize(config.maxShuffleBatchByteSize())
+ // Bolt names its transport-neutral RSS partition writer "celeborn".
+ // The supplied pusher still sends every block through Uniffle.
+ .setWriterType("celeborn")
Review Comment:
`setWriterType("celeborn")` is an implicit dependency on a Bolt-internal
writer name from the Uniffle integration. If the native side renames this
transport-neutral writer, the failure will be silent (or a confusing runtime
error). Consider hoisting it into a named constant in the JNI/config layer so
the contract is visible.
##########
backends-bolt/src/main/scala/org/apache/gluten/backendsapi/bolt/BoltBackend.scala:
##########
@@ -572,4 +572,6 @@ object BoltBackendSettings extends BackendSettingsApi {
override def supportOverwriteByExpression(): Boolean =
enableEnhancedFeatures()
override def supportOverwritePartitionsDynamic(): Boolean =
enableEnhancedFeatures()
+
+ override def supportEmptySchemaColumnarShuffle(): Boolean = false
Review Comment:
`BoltValidatorApi.doColumnarShuffleExchangeExecValidate` already rejects
empty output/input schemas unconditionally, so this setting does not gate
shuffle validation. Its only current effect is `CollectLimitTransformerRule`.
If the intent is only to make CollectLimit fall back, please say that in the PR
description; otherwise this override is a no-op for the stated purpose.
##########
backends-bolt/src-uniffle/main/java/org/apache/spark/shuffle/writer/BoltUniffleColumnarShuffleWriter.java:
##########
@@ -224,29 +222,36 @@ public long spill(MemoryTarget self, Spiller.Phase phase,
long size) {
return;
}
long startTime = System.nanoTime();
- GlutenSplitResult splitResult;
+ BoltSplitResult splitResult;
try {
splitResult = shuffleWriterJniWrapper.stop(nativeShuffleWriter);
} catch (IOException e) {
throw new RssException(e);
}
columnarDep.metrics().get("shuffleWallTime").get().add(System.nanoTime() -
startTime);
+ // AQE needs the uncompressed size, including batches flushed by stop().
+ columnarDep
+ .metrics()
+ .get("dataSize")
+ .get()
+ .add(Arrays.stream(splitResult.getRawPartitionLengths()).sum());
Review Comment:
Metric semantics change: `dataSize` now comes from `rawPartitionLengths`
after `stop()` instead of the per-batch return value of `write()`. That is
arguably more correct for AQE, but please assert the exact value in the new
suite so a future change in native accounting does not silently alter it.
--
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]
