Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[VL] Row based sort shuffle implementation #6475

Merged
merged 9 commits into from
Jul 24, 2024
Merged

[VL] Row based sort shuffle implementation #6475

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
920fd08
convert to row and sort
marin-ma Jul 10, 2024
d922a36
separate sort shuffle and rss_sort shuffle
marin-ma Jul 22, 2024
97cd9b7
refine metrics
marin-ma Jul 22, 2024
b351561
fix style
marin-ma Jul 22, 2024
016a6fb
fix
marin-ma Jul 22, 2024
4705462
fix
Jul 22, 2024
cdae405
fix coredump
marin-ma Jul 23, 2024
83b6123
cleanup
marin-ma Jul 23, 2024
b869910
format
marin-ma Jul 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,27 +258,33 @@ class VeloxMetricsApi extends MetricsApi with Logging {
sparkContext: SparkContext,
isSort: Boolean): Map[String, SQLMetric] = {
val baseMetrics = Map(
"dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size"),
"numPartitions" -> SQLMetrics.createMetric(sparkContext, "number of partitions"),
"dataSize" -> SQLMetrics.createSizeMetric(sparkContext, "data size"),
"bytesSpilled" -> SQLMetrics.createSizeMetric(sparkContext, "shuffle bytes spilled"),
"splitBufferSize" -> SQLMetrics.createSizeMetric(sparkContext, "split buffer size"),
"splitTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to split"),
"spillTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to spill"),
"deserializeTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to deserialize"),
"avgReadBatchNumRows" -> SQLMetrics
.createAverageMetric(sparkContext, "avg read batch num rows"),
"numInputRows" -> SQLMetrics.createMetric(sparkContext, "number of input rows"),
"numOutputRows" -> SQLMetrics
.createMetric(sparkContext, "number of output rows"),
"inputBatches" -> SQLMetrics
.createMetric(sparkContext, "number of input batches")
.createMetric(sparkContext, "number of input batches"),
"spillTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to spill"),
"compressTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to compress"),
"decompressTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to decompress"),
"deserializeTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to deserialize"),
// For hash shuffle writer, the peak bytes represents the maximum split buffer size.
// For sort shuffle writer, the peak bytes represents the maximum
// row buffer + sort buffer size.
"peakBytes" -> SQLMetrics.createSizeMetric(sparkContext, "peak bytes allocated")
)
if (isSort) {
baseMetrics
baseMetrics ++ Map(
"sortTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to shuffle sort"),
"c2rTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to shuffle c2r")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand this is newly added for sort based shuffle, it's not available in hash based shuffle, however this may bring some trouble for analysis script

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

however this may bring some trouble for analysis script

Checked with the analysis script. Except for the get_shuffle_stat, other analysis (sar, emon, time breakdown, app comparison) goes well.
Perhaps we can introduce a new metrics shuffle write wall time to calculate the real time for both hash and sort shuffle, and modify the analysis script to use this one.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps we can introduce a new metrics shuffle write wall time to calculate the real time for both hash and sort shuffle, and modify the analysis script to use this one.

Will address in follow-up PR.

)
} else {
baseMetrics ++ Map(
"compressTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to compress"),
"decompressTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to decompress")
"splitTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time to split")
)
}
}
Expand Down
39 changes: 21 additions & 18 deletions ...ends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,24 +349,30 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
}

def maybeAddAppendBatchesExec(plan: SparkPlan): SparkPlan = {
if (GlutenConfig.getConf.veloxCoalesceBatchesBeforeShuffle) {
VeloxAppendBatchesExec(plan, GlutenConfig.getConf.veloxMinBatchSizeForShuffle)
} else {
plan
plan match {
case shuffle: ColumnarShuffleExchangeExec
if !shuffle.useSortBasedShuffle &&
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this due to sort will do the coalesce behavior?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No matter coalesced or not, sort will convert all inputs into rows, so it won't benefit from coalesce. The test result shows the coalesce op only brings extra overhead.

GlutenConfig.getConf.veloxCoalesceBatchesBeforeShuffle =>
val appendBatches =
VeloxAppendBatchesExec(shuffle.child, GlutenConfig.getConf.veloxMinBatchSizeForShuffle)
shuffle.withNewChildren(Seq(appendBatches))
case _ => plan
}
}

val child = shuffle.child

shuffle.outputPartitioning match {
val newShuffle = shuffle.outputPartitioning match {
case HashPartitioning(exprs, _) =>
val hashExpr = new Murmur3Hash(exprs)
val projectList = Seq(Alias(hashExpr, "hash_partition_key")()) ++ child.output
val projectTransformer = ProjectExecTransformer(projectList, child)
val validationResult = projectTransformer.doValidate()
if (validationResult.ok()) {
val newChild = maybeAddAppendBatchesExec(projectTransformer)
ColumnarShuffleExchangeExec(shuffle, newChild, newChild.output.drop(1))
ColumnarShuffleExchangeExec(
shuffle,
projectTransformer,
projectTransformer.output.drop(1))
} else {
FallbackTags.add(shuffle, validationResult)
shuffle.withNewChildren(child :: Nil)
Expand All @@ -382,8 +388,7 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
// null type since the value always be null.
val columnsForHash = child.output.filterNot(_.dataType == NullType)
if (columnsForHash.isEmpty) {
val newChild = maybeAddAppendBatchesExec(child)
ColumnarShuffleExchangeExec(shuffle, newChild, newChild.output)
ColumnarShuffleExchangeExec(shuffle, child, child.output)
} else {
val hashExpr = new Murmur3Hash(columnsForHash)
val projectList = Seq(Alias(hashExpr, "hash_partition_key")()) ++ child.output
Expand All @@ -404,18 +409,20 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
ProjectExecTransformer(projectList.drop(1), sortByHashCode)
val validationResult = dropSortColumnTransformer.doValidate()
if (validationResult.ok()) {
val newChild = maybeAddAppendBatchesExec(dropSortColumnTransformer)
ColumnarShuffleExchangeExec(shuffle, newChild, newChild.output)
ColumnarShuffleExchangeExec(
shuffle,
dropSortColumnTransformer,
dropSortColumnTransformer.output)
} else {
FallbackTags.add(shuffle, validationResult)
shuffle.withNewChildren(child :: Nil)
}
}
}
case _ =>
val newChild = maybeAddAppendBatchesExec(child)
ColumnarShuffleExchangeExec(shuffle, newChild, null)
ColumnarShuffleExchangeExec(shuffle, child, null)
}
maybeAddAppendBatchesExec(newShuffle)
}

/** Generate ShuffledHashJoinExecTransformer. */
Expand Down Expand Up @@ -572,11 +579,7 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi {
val numOutputRows = metrics("numOutputRows")
val deserializeTime = metrics("deserializeTime")
val readBatchNumRows = metrics("avgReadBatchNumRows")
val decompressTime: Option[SQLMetric] = if (!isSort) {
Some(metrics("decompressTime"))
} else {
None
}
val decompressTime = metrics("decompressTime")
if (GlutenConfig.getConf.isUseCelebornShuffleManager) {
val clazz = ClassUtils.getClass("org.apache.spark.shuffle.CelebornColumnarBatchSerializer")
val constructor =
Expand Down
1 change: 1 addition & 0 deletions cpp/core/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,7 @@ set(SPARK_COLUMNAR_PLUGIN_SRCS
shuffle/RoundRobinPartitioner.cc
shuffle/ShuffleMemoryPool.cc
shuffle/ShuffleReader.cc
shuffle/ShuffleWriter.cc
shuffle/SinglePartitioner.cc
shuffle/Spill.cc
shuffle/Utils.cc
Expand Down
19 changes: 6 additions & 13 deletions cpp/core/jni/JniWrapper.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,7 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) {
jniByteInputStreamClose = getMethodIdOrError(env, jniByteInputStreamClass, "close", "()V");

splitResultClass = createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/vectorized/GlutenSplitResult;");
splitResultConstructor = getMethodIdOrError(env, splitResultClass, "<init>", "(JJJJJJJ[J[J)V");
splitResultConstructor = getMethodIdOrError(env, splitResultClass, "<init>", "(JJJJJJJJJ[J[J)V");

columnarBatchSerializeResultClass =
createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/vectorized/ColumnarBatchSerializeResult;");
Expand Down Expand Up @@ -780,14 +780,7 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_ShuffleWriterJniWrappe
.partitioning = gluten::toPartitioning(jStringToCString(env, partitioningNameJstr)),
.taskAttemptId = (int64_t)taskAttemptId,
.startPartitionId = startPartitionId,
};
auto shuffleWriterTypeC = env->GetStringUTFChars(shuffleWriterTypeJstr, JNI_FALSE);
auto shuffleWriterType = std::string(shuffleWriterTypeC);
env->ReleaseStringUTFChars(shuffleWriterTypeJstr, shuffleWriterTypeC);

if (shuffleWriterType == "sort") {
shuffleWriterOptions.shuffleWriterType = kSortShuffle;
}
.shuffleWriterType = gluten::ShuffleWriter::stringToType(jStringToCString(env, shuffleWriterTypeJstr))};

// Build PartitionWriterOptions.
auto partitionWriterOptions = PartitionWriterOptions{
Expand Down Expand Up @@ -945,9 +938,11 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_vectorized_ShuffleWriterJniWrap
shuffleWriter->totalWriteTime(),
shuffleWriter->totalEvictTime(),
shuffleWriter->totalCompressTime(),
shuffleWriter->totalSortTime(),
shuffleWriter->totalC2RTime(),
shuffleWriter->totalBytesWritten(),
shuffleWriter->totalBytesEvicted(),
shuffleWriter->maxPartitionBufferSize(),
shuffleWriter->peakBytesAllocated(),
partitionLengthArr,
rawPartitionLengthArr);

Expand Down Expand Up @@ -996,9 +991,7 @@ JNIEXPORT jlong JNICALL Java_org_apache_gluten_vectorized_ShuffleReaderJniWrappe
options.batchSize = batchSize;
// TODO: Add coalesce option and maximum coalesced size.

if (jStringToCString(env, shuffleWriterType) == "sort") {
options.shuffleWriterType = kSortShuffle;
}
options.shuffleWriterType = gluten::ShuffleWriter::stringToType(jStringToCString(env, shuffleWriterType));
std::shared_ptr<arrow::Schema> schema =
gluten::arrowGetOrThrow(arrow::ImportSchema(reinterpret_cast<struct ArrowSchema*>(cSchema)));

Expand Down
2 changes: 1 addition & 1 deletion cpp/core/operators/serializer/ColumnarBatchSerializer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ namespace gluten {

class ColumnarBatchSerializer {
public:
ColumnarBatchSerializer(arrow::MemoryPool* arrowPool, struct ArrowSchema* cSchema) : arrowPool_(arrowPool) {}
ColumnarBatchSerializer(arrow::MemoryPool* arrowPool) : arrowPool_(arrowPool) {}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good cleanup


virtual ~ColumnarBatchSerializer() = default;

Expand Down
5 changes: 1 addition & 4 deletions cpp/core/shuffle/FallbackRangePartitioner.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,18 +23,15 @@ namespace gluten {
arrow::Status gluten::FallbackRangePartitioner::compute(
const int32_t* pidArr,
const int64_t numRows,
std::vector<uint32_t>& row2Partition,
std::vector<uint32_t>& partition2RowCount) {
std::vector<uint32_t>& row2Partition) {
row2Partition.resize(numRows);
std::fill(std::begin(partition2RowCount), std::end(partition2RowCount), 0);
for (auto i = 0; i < numRows; ++i) {
auto pid = pidArr[i];
if (pid >= numPartitions_) {
return arrow::Status::Invalid(
"Partition id ", std::to_string(pid), " is equal or greater than ", std::to_string(numPartitions_));
}
row2Partition[i] = pid;
partition2RowCount[pid]++;
}
return arrow::Status::OK();
}
Expand Down
6 changes: 1 addition & 5 deletions cpp/core/shuffle/FallbackRangePartitioner.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,7 @@ class FallbackRangePartitioner final : public Partitioner {
public:
FallbackRangePartitioner(int32_t numPartitions) : Partitioner(numPartitions, true) {}

arrow::Status compute(
const int32_t* pidArr,
const int64_t numRows,
std::vector<uint32_t>& row2partition,
std::vector<uint32_t>& partition2RowCount) override;
arrow::Status compute(const int32_t* pidArr, const int64_t numRows, std::vector<uint32_t>& row2partition) override;

arrow::Status compute(
const int32_t* pidArr,
Expand Down
14 changes: 2 additions & 12 deletions cpp/core/shuffle/HashPartitioner.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,23 +37,13 @@ int32_t computePid(const int32_t* pidArr, int64_t i, int32_t numPartitions) {
return pid;
}

arrow::Status gluten::HashPartitioner::compute(
const int32_t* pidArr,
const int64_t numRows,
std::vector<uint32_t>& row2partition,
std::vector<uint32_t>& partition2RowCount) {
arrow::Status
gluten::HashPartitioner::compute(const int32_t* pidArr, const int64_t numRows, std::vector<uint32_t>& row2partition) {
row2partition.resize(numRows);
std::fill(std::begin(partition2RowCount), std::end(partition2RowCount), 0);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it seems this fill operation is all removed, will this impact the performance?

Copy link
Contributor Author

@marin-ma marin-ma Jul 23, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This logic is moved into Hash shuffle writer https://github.com/apache/incubator-gluten/pull/6475/files#diff-0ea9c330c6637a7c2814fd738102f2928bca7b74a99c9f15163a6dce3f3c1e5fR278, because the partition2RowCount computation is not used in sort shuffle writer. I will push some cleanup to completely remove this parameter. It won't have performance impact to the hash shuffle writer.


for (auto i = 0; i < numRows; ++i) {
auto pid = computePid(pidArr, i, numPartitions_);
row2partition[i] = pid;
}

for (auto& pid : row2partition) {
partition2RowCount[pid]++;
}

return arrow::Status::OK();
}

Expand Down
6 changes: 1 addition & 5 deletions cpp/core/shuffle/HashPartitioner.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,7 @@ class HashPartitioner final : public Partitioner {
public:
HashPartitioner(int32_t numPartitions) : Partitioner(numPartitions, true) {}

arrow::Status compute(
const int32_t* pidArr,
const int64_t numRows,
std::vector<uint32_t>& row2partition,
std::vector<uint32_t>& partition2RowCount) override;
arrow::Status compute(const int32_t* pidArr, const int64_t numRows, std::vector<uint32_t>& row2partition) override;

arrow::Status compute(
const int32_t* pidArr,
Expand Down
Loading
Loading