You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "marin-ma (via GitHub)" <gi...@apache.org> on 2024/07/16 08:21:04 UTC
[PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
marin-ma opened a new pull request, #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475
(no comment)
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2246814041
TPCDS perf on jenkins
| Query | log/native_6475_time.csv | log/native_master_07_23_2024_time.csv | Difference | Percentage |
|-------|---------------------------|---------------------------------------|------------|------------|
| q1 | 9.47 | 9.25 | -0.217 | 97.71% |
| q2 | 7.19 | 7.19 | -0.007 | 99.90% |
| q3 | 2.62 | 2.59 | -0.021 | 99.18% |
| q4 | 47.04 | 47.26 | 0.217 | 100.46% |
| q5 | 5.80 | 5.88 | 0.076 | 101.32% |
| q6 | 2.73 | 1.55 | -1.180 | 56.70% |
| q7 | 4.09 | 4.10 | 0.006 | 100.15% |
| q8 | 2.12 | 2.88 | 0.758 | 135.71% |
| q9 | 13.26 | 12.97 | -0.288 | 97.83% |
| q10 | 8.31 | 8.49 | 0.184 | 102.21% |
| q11 | 25.40 | 24.49 | -0.909 | 96.42% |
| q12 | 0.97 | 1.96 | 0.986 | 201.41% |
| q13 | 3.80 | 4.46 | 0.660 | 117.38% |
| q14a | 31.55 | 31.97 | 0.417 | 101.32% |
| q14b | 29.04 | 28.89 | -0.148 | 99.49% |
| q15 | 2.62 | 2.94 | 0.318 | 112.13% |
| q16 | 5.44 | 6.15 | 0.710 | 113.05% |
| q17 | 3.90 | 3.84 | -0.052 | 98.66% |
| q18 | 5.94 | 5.77 | -0.169 | 97.15% |
| q19 | 1.92 | 1.94 | 0.023 | 101.20% |
| q20 | 2.48 | 1.00 | -1.473 | 40.50% |
| q21 | 0.66 | 0.60 | -0.068 | 89.77% |
| q22 | 3.00 | 2.98 | -0.021 | 99.32% |
| q23a | 57.16 | 56.35 | -0.815 | 98.57% |
| q23b | 70.90 | 69.86 | -1.042 | 98.53% |
| q24a | 65.02 | 65.96 | 0.947 | 101.46% |
| q24b | 64.10 | 63.05 | -1.047 | 98.37% |
| q25 | 3.09 | 4.25 | 1.161 | 137.54% |
| q26 | 1.72 | 1.80 | 0.084 | 104.87% |
| q27 | 2.75 | 2.66 | -0.095 | 96.56% |
| q28 | 16.29 | 16.28 | -0.008 | 99.95% |
| q29 | 5.90 | 5.87 | -0.032 | 99.47% |
| q30 | 3.92 | 3.89 | -0.032 | 99.18% |
| q31 | 4.72 | 4.73 | 0.012 | 100.25% |
| q32 | 0.74 | 0.76 | 0.020 | 102.70% |
| q33 | 1.52 | 1.59 | 0.074 | 104.85% |
| q34 | 2.68 | 2.07 | -0.610 | 77.21% |
| q35 | 4.98 | 5.64 | 0.664 | 113.34% |
| q36 | 1.84 | 1.72 | -0.124 | 93.27% |
| q37 | 2.58 | 2.49 | -0.091 | 96.47% |
| q38 | 9.60 | 9.53 | -0.063 | 99.34% |
| q39a | 2.67 | 2.88 | 0.213 | 108.00% |
| q39b | 2.27 | 2.40 | 0.134 | 105.92% |
| q40 | 2.32 | 2.44 | 0.122 | 105.27% |
| q41 | 1.07 | 0.34 | -0.733 | 31.56% |
| q42 | 0.39 | 0.37 | -0.023 | 94.12% |
| q43 | 1.74 | 1.65 | -0.092 | 94.69% |
| q44 | 5.10 | 5.16 | 0.061 | 101.19% |
| q45 | 2.20 | 2.15 | -0.045 | 97.95% |
| q46 | 2.17 | 2.27 | 0.094 | 104.33% |
| q47 | 9.50 | 10.21 | 0.707 | 107.44% |
| q48 | 2.58 | 2.64 | 0.066 | 102.55% |
| q49 | 3.37 | 3.35 | -0.013 | 99.61% |
| q50 | 16.79 | 16.72 | -0.062 | 99.63% |
| q51 | 6.36 | 5.56 | -0.809 | 87.29% |
| q52 | 0.70 | 1.29 | 0.596 | 185.61% |
| q53 | 1.23 | 1.27 | 0.037 | 103.00% |
| q54 | 2.63 | 2.48 | -0.153 | 94.17% |
| q55 | 0.67 | 0.66 | -0.013 | 98.05% |
| q56 | 2.20 | 1.41 | -0.787 | 64.18% |
| q57 | 6.03 | 6.00 | -0.030 | 99.50% |
| q58 | 1.64 | 1.83 | 0.198 | 112.10% |
| q59 | 3.19 | 4.07 | 0.886 | 127.82% |
| q60 | 2.02 | 2.70 | 0.675 | 133.36% |
| q61 | 1.77 | 1.71 | -0.061 | 96.53% |
| q62 | 2.52 | 2.51 | -0.018 | 99.30% |
| q63 | 1.07 | 1.09 | 0.026 | 102.46% |
| q64 | 27.15 | 25.90 | -1.245 | 95.42% |
| q65 | 10.92 | 10.22 | -0.697 | 93.61% |
| q66 | 2.24 | 1.99 | -0.247 | 88.97% |
| q67 | 62.16 | 62.97 | 0.808 | 101.30% |
| q68 | 2.51 | 2.09 | -0.422 | 83.19% |
| q69 | 4.93 | 4.66 | -0.265 | 94.61% |
| q70 | 4.62 | 4.47 | -0.155 | 96.65% |
| q71 | 1.81 | 2.38 | 0.574 | 131.80% |
| q72 | 18.41 | 18.91 | 0.508 | 102.76% |
| q73 | 1.54 | 1.72 | 0.182 | 111.78% |
| q74 | 16.41 | 16.07 | -0.339 | 97.93% |
| q75 | 19.10 | 19.20 | 0.099 | 100.52% |
| q76 | 5.91 | 5.93 | 0.023 | 100.40% |
| q77 | 1.44 | 1.34 | -0.091 | 93.69% |
| q78 | 32.88 | 32.67 | -0.211 | 99.36% |
| q79 | 2.75 | 2.64 | -0.104 | 96.22% |
| q80 | 7.84 | 7.54 | -0.294 | 96.25% |
| q81 | 4.32 | 4.27 | -0.050 | 98.85% |
| q82 | 5.30 | 5.02 | -0.278 | 94.75% |
| q83 | 1.05 | 1.00 | -0.047 | 95.53% |
| q84 | 1.90 | 2.04 | 0.136 | 107.16% |
| q85 | 4.55 | 4.44 | -0.113 | 97.52% |
| q86 | 1.76 | 1.76 | 0.003 | 100.17% |
| q87 | 9.85 | 10.10 | 0.254 | 102.58% |
| q88 | 11.77 | 11.55 | -0.223 | 98.10% |
| q89 | 1.76 | 1.65 | -0.102 | 94.17% |
| q90 | 1.74 | 1.89 | 0.150 | 108.67% |
| q91 | 1.60 | 1.70 | 0.096 | 105.97% |
| q92 | 0.76 | 0.84 | 0.089 | 111.76% |
| q93 | 23.46 | 23.85 | 0.393 | 101.67% |
| q94 | 8.38 | 7.84 | -0.540 | 93.56% |
| q95 | 55.65 | 54.44 | -1.205 | 97.83% |
| q96 | 1.65 | 2.03 | 0.375 | 122.74% |
| q97 | 10.12 | 9.49 | -0.625 | 93.82% |
| q98 | 1.55 | 1.30 | -0.249 | 83.93% |
| q99 | 5.77 | 5.12 | -0.649 | 88.73% |
| total | 998.85 | 994.22 | -4.632 | 99.54% |
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1687659792
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala:
##########
@@ -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")
Review Comment:
> 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.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1688973227
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala:
##########
@@ -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")
Review Comment:
> 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.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implementation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r2228842835
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
Review Comment:
The sort shuffle writer convert the inputs from column to row, sort the rows on the writer side ,and convert the rows back to column on the reader side. For single partition, there's no sort operation. We directly write the input without any other operations.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2230305956
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implementation [incubator-gluten]
Posted by "mcdull-zhang (via GitHub)" <gi...@apache.org>.
mcdull-zhang commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r2228427714
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
Review Comment:
Excuse me, why can't SinglePartition use SortShuffleWriter?
In principle, it is supported, but the performance is not as good as Hash. Is this correct?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2230305488
<!--
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/LICEGJ-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.
-->
Thanks for opening a pull request!
Could you open an issue for this pull request on Github Issues?
https://github.com/apache/incubator-gluten/issues
Then could you also rename ***commit message*** and ***pull request title*** in the following format?
[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}
See also:
* [Other pull requests](https://github.com/apache/incubator-gluten/pulls/)
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2238189029
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "zhouyuan (via GitHub)" <gi...@apache.org>.
zhouyuan commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1687310066
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala:
##########
@@ -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 &&
Review Comment:
Is this due to sort will do the coalesce behavior?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2242459040
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2230378842
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2242525716
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1687481779
##########
cpp/core/shuffle/HashPartitioner.cc:
##########
@@ -43,7 +43,6 @@ arrow::Status gluten::HashPartitioner::compute(
std::vector<uint32_t>& row2partition,
std::vector<uint32_t>& partition2RowCount) {
row2partition.resize(numRows);
- std::fill(std::begin(partition2RowCount), std::end(partition2RowCount), 0);
Review Comment:
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.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "zhouyuan (via GitHub)" <gi...@apache.org>.
zhouyuan commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1687370806
##########
cpp/core/operators/serializer/ColumnarBatchSerializer.h:
##########
@@ -25,7 +25,7 @@ namespace gluten {
class ColumnarBatchSerializer {
public:
- ColumnarBatchSerializer(arrow::MemoryPool* arrowPool, struct ArrowSchema* cSchema) : arrowPool_(arrowPool) {}
+ ColumnarBatchSerializer(arrow::MemoryPool* arrowPool) : arrowPool_(arrowPool) {}
Review Comment:
good cleanup
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
+ partition2RowCount_.resize(numPartitions_, 0);
+ array_.resize(initialSize_);
+ return arrow::Status::OK();
+}
+
+void VeloxSortShuffleWriter::initRowType(const facebook::velox::RowVectorPtr& rv) {
+ if (UNLIKELY(!rowType_)) {
+ rowType_ = facebook::velox::asRowType(rv->type());
+ fixedRowSize_ = facebook::velox::row::CompactRow::fixedRowSize(rowType_);
+ }
+}
+
+arrow::Result<facebook::velox::RowVectorPtr> VeloxSortShuffleWriter::getPeeledRowVector(
+ const std::shared_ptr<ColumnarBatch>& cb) {
+ if (options_.partitioning == Partitioning::kRange) {
+ auto compositeBatch = std::dynamic_pointer_cast<CompositeColumnarBatch>(cb);
+ VELOX_CHECK_NOT_NULL(compositeBatch);
+ auto batches = compositeBatch->getBatches();
+ VELOX_CHECK_EQ(batches.size(), 2);
+
+ auto pidBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[0]);
+ auto pidArr = getFirstColumn(*(pidBatch->getRowVector()));
+ RETURN_NOT_OK(partitioner_->compute(pidArr, pidBatch->numRows(), row2Partition_, partition2RowCount_));
+
+ auto rvBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[1]);
+ return rvBatch->getFlattenedRowVector();
+ } else {
+ auto veloxColumnBatch = VeloxColumnarBatch::from(veloxPool_.get(), cb);
+ VELOX_CHECK_NOT_NULL(veloxColumnBatch);
+ auto rv = veloxColumnBatch->getFlattenedRowVector();
+ if (partitioner_->hasPid()) {
+ auto pidArr = getFirstColumn(*rv);
+ RETURN_NOT_OK(partitioner_->compute(pidArr, rv->size(), row2Partition_, partition2RowCount_));
+ return getStrippedRowVector(*rv);
+ } else {
+ RETURN_NOT_OK(partitioner_->compute(nullptr, rv->size(), row2Partition_, partition2RowCount_));
+ return rv;
+ }
+ }
+}
+
+arrow::Status VeloxSortShuffleWriter::insert(const facebook::velox::RowVectorPtr& vector, int64_t memLimit) {
+ ScopedTimer timer(&c2rTime_);
+ auto inputRows = vector->size();
+ VELOX_DCHECK_GT(inputRows, 0);
+
+ facebook::velox::row::CompactRow row(vector);
+
+ if (!fixedRowSize_) {
+ rowSizes_.resize(inputRows + 1);
+ rowSizes_[0] = 0;
+ for (auto i = 0; i < inputRows; ++i) {
+ rowSizes_[i + 1] = rowSizes_[i] + row.rowSize(i);
+ }
+ }
+
+ uint32_t rowOffset = 0;
+ while (rowOffset < inputRows) {
+ auto remainingRows = inputRows - rowOffset;
+ auto rows = maxRowsToInsert(rowOffset, remainingRows);
+ if (rows == 0) {
+ auto minSizeRequired = fixedRowSize_ ? fixedRowSize_.value() : rowSizes_[rowOffset + 1] - rowSizes_[rowOffset];
+ acquireNewBuffer(memLimit, minSizeRequired);
+ rows = maxRowsToInsert(rowOffset, remainingRows);
+ ARROW_RETURN_IF(
+ rows == 0, arrow::Status::Invalid("Failed to insert rows. Remaining rows: " + std::to_string(remainingRows)));
+ }
+ RETURN_NOT_OK(maybeSpill(rows));
+ insertRows(row, rowOffset, rows);
+ rowOffset += rows;
+ }
+ return arrow::Status::OK();
+}
+
+void VeloxSortShuffleWriter::insertRows(facebook::velox::row::CompactRow& row, uint32_t offset, uint32_t rows) {
+ // Allocate newArray can trigger spill.
+ growArrayIfNecessary(rows);
+ for (auto i = offset; i < offset + rows; ++i) {
+ auto size = row.serialize(i, currentPage_ + pageCursor_);
+ array_[offset_++] = {toCompactRowId(row2Partition_[i], pageNumber_, pageCursor_), size};
+ pageCursor_ += size;
+ }
+}
+
+arrow::Status VeloxSortShuffleWriter::maybeSpill(int32_t nextRows) {
+ if ((uint64_t)offset_ + nextRows > std::numeric_limits<uint32_t>::max()) {
+ EvictGuard evictGuard{evictState_};
+ RETURN_NOT_OK(evictAllPartitions());
+ }
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::evictAllPartitions() {
+ {
+ ScopedTimer timer(&sortTime_);
+ // TODO: Add radix sort to align with Spark.
+ std::sort(array_.begin(), array_.begin() + offset_);
Review Comment:
can we use the prefix sort in Velox here?
##########
cpp/core/shuffle/HashPartitioner.cc:
##########
@@ -43,7 +43,6 @@ arrow::Status gluten::HashPartitioner::compute(
std::vector<uint32_t>& row2partition,
std::vector<uint32_t>& partition2RowCount) {
row2partition.resize(numRows);
- std::fill(std::begin(partition2RowCount), std::end(partition2RowCount), 0);
Review Comment:
it seems this fill operation is all removed, will this impact the performance?
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala:
##########
@@ -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")
Review Comment:
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
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2246811660
TPCH perf on jenkins
| query | log/native_6475_time.csv | log/native_master_07_23_2024_time.csv | difference | percentage |
|-------|---------------------------|---------------------------------------|------------|------------|
| q1 | 24.34 | 25.32 | 0.985 | 104.05% |
| q2 | 22.34 | 22.18 | -0.165 | 99.26% |
| q3 | 36.98 | 36.78 | -0.204 | 99.45% |
| q4 | 26.37 | 26.18 | -0.195 | 99.26% |
| q5 | 64.72 | 65.08 | 0.359 | 100.55% |
| q6 | 10.71 | 11.36 | 0.650 | 106.07% |
| q7 | 72.83 | 71.56 | -1.271 | 98.25% |
| q8 | 70.20 | 73.17 | 2.960 | 104.22% |
| q9 | 98.08 | 111.75 | 13.670 | 113.94% |
| q10 | 49.69 | 47.90 | -1.787 | 96.40% |
| q11 | 15.51 | 18.05 | 2.536 | 116.35% |
| q12 | 18.22 | 16.70 | -1.521 | 91.65% |
| q13 | 25.73 | 26.52 | 0.799 | 103.10% |
| q14 | 13.57 | 14.21 | 0.645 | 104.75% |
| q15 | 25.25 | 25.67 | 0.418 | 101.66% |
| q16 | 11.50 | 12.12 | 0.618 | 105.37% |
| q17 | 82.68 | 83.82 | 1.143 | 101.38% |
| q18 | 123.77 | 121.85 | -1.918 | 98.45% |
| q19 | 14.86 | 14.27 | -0.588 | 96.04% |
| q20 | 31.78 | 30.28 | -1.499 | 95.28% |
| q21 | 207.93 | 210.82 | 2.897 | 101.39% |
| q22 | 12.75 | 10.19 | -2.568 | 79.86% |
| total | 1059.82 | 1075.79 | 15.964 | 101.51% |
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2242470120
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1687312260
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala:
##########
@@ -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 &&
Review Comment:
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.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1687481779
##########
cpp/core/shuffle/HashPartitioner.cc:
##########
@@ -43,7 +43,6 @@ arrow::Status gluten::HashPartitioner::compute(
std::vector<uint32_t>& row2partition,
std::vector<uint32_t>& partition2RowCount) {
row2partition.resize(numRows);
- std::fill(std::begin(partition2RowCount), std::end(partition2RowCount), 0);
Review Comment:
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.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2242119402
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer merged PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1687485028
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
+ partition2RowCount_.resize(numPartitions_, 0);
+ array_.resize(initialSize_);
+ return arrow::Status::OK();
+}
+
+void VeloxSortShuffleWriter::initRowType(const facebook::velox::RowVectorPtr& rv) {
+ if (UNLIKELY(!rowType_)) {
+ rowType_ = facebook::velox::asRowType(rv->type());
+ fixedRowSize_ = facebook::velox::row::CompactRow::fixedRowSize(rowType_);
+ }
+}
+
+arrow::Result<facebook::velox::RowVectorPtr> VeloxSortShuffleWriter::getPeeledRowVector(
+ const std::shared_ptr<ColumnarBatch>& cb) {
+ if (options_.partitioning == Partitioning::kRange) {
+ auto compositeBatch = std::dynamic_pointer_cast<CompositeColumnarBatch>(cb);
+ VELOX_CHECK_NOT_NULL(compositeBatch);
+ auto batches = compositeBatch->getBatches();
+ VELOX_CHECK_EQ(batches.size(), 2);
+
+ auto pidBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[0]);
+ auto pidArr = getFirstColumn(*(pidBatch->getRowVector()));
+ RETURN_NOT_OK(partitioner_->compute(pidArr, pidBatch->numRows(), row2Partition_, partition2RowCount_));
+
+ auto rvBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[1]);
+ return rvBatch->getFlattenedRowVector();
+ } else {
+ auto veloxColumnBatch = VeloxColumnarBatch::from(veloxPool_.get(), cb);
+ VELOX_CHECK_NOT_NULL(veloxColumnBatch);
+ auto rv = veloxColumnBatch->getFlattenedRowVector();
+ if (partitioner_->hasPid()) {
+ auto pidArr = getFirstColumn(*rv);
+ RETURN_NOT_OK(partitioner_->compute(pidArr, rv->size(), row2Partition_, partition2RowCount_));
+ return getStrippedRowVector(*rv);
+ } else {
+ RETURN_NOT_OK(partitioner_->compute(nullptr, rv->size(), row2Partition_, partition2RowCount_));
+ return rv;
+ }
+ }
+}
+
+arrow::Status VeloxSortShuffleWriter::insert(const facebook::velox::RowVectorPtr& vector, int64_t memLimit) {
+ ScopedTimer timer(&c2rTime_);
+ auto inputRows = vector->size();
+ VELOX_DCHECK_GT(inputRows, 0);
+
+ facebook::velox::row::CompactRow row(vector);
+
+ if (!fixedRowSize_) {
+ rowSizes_.resize(inputRows + 1);
+ rowSizes_[0] = 0;
+ for (auto i = 0; i < inputRows; ++i) {
+ rowSizes_[i + 1] = rowSizes_[i] + row.rowSize(i);
+ }
+ }
+
+ uint32_t rowOffset = 0;
+ while (rowOffset < inputRows) {
+ auto remainingRows = inputRows - rowOffset;
+ auto rows = maxRowsToInsert(rowOffset, remainingRows);
+ if (rows == 0) {
+ auto minSizeRequired = fixedRowSize_ ? fixedRowSize_.value() : rowSizes_[rowOffset + 1] - rowSizes_[rowOffset];
+ acquireNewBuffer(memLimit, minSizeRequired);
+ rows = maxRowsToInsert(rowOffset, remainingRows);
+ ARROW_RETURN_IF(
+ rows == 0, arrow::Status::Invalid("Failed to insert rows. Remaining rows: " + std::to_string(remainingRows)));
+ }
+ RETURN_NOT_OK(maybeSpill(rows));
+ insertRows(row, rowOffset, rows);
+ rowOffset += rows;
+ }
+ return arrow::Status::OK();
+}
+
+void VeloxSortShuffleWriter::insertRows(facebook::velox::row::CompactRow& row, uint32_t offset, uint32_t rows) {
+ // Allocate newArray can trigger spill.
+ growArrayIfNecessary(rows);
+ for (auto i = offset; i < offset + rows; ++i) {
+ auto size = row.serialize(i, currentPage_ + pageCursor_);
+ array_[offset_++] = {toCompactRowId(row2Partition_[i], pageNumber_, pageCursor_), size};
+ pageCursor_ += size;
+ }
+}
+
+arrow::Status VeloxSortShuffleWriter::maybeSpill(int32_t nextRows) {
+ if ((uint64_t)offset_ + nextRows > std::numeric_limits<uint32_t>::max()) {
+ EvictGuard evictGuard{evictState_};
+ RETURN_NOT_OK(evictAllPartitions());
+ }
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::evictAllPartitions() {
+ {
+ ScopedTimer timer(&sortTime_);
+ // TODO: Add radix sort to align with Spark.
+ std::sort(array_.begin(), array_.begin() + offset_);
Review Comment:
Seems like we cannot benefit from prefix sort here. By following Spark's design, `array_` stores a compacted id (uint64_t) to the row https://github.com/apache/incubator-gluten/pull/6475/files#diff-ff82d0ed4685583538881cabfe93ed92f202dd182492a66937fba0ccd5f7a22fR33-R36 Only the `array_` is sorted here, not the entire row.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implementation [incubator-gluten]
Posted by "boneanxs (via GitHub)" <gi...@apache.org>.
boneanxs commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1976826680
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
+ array_.resize(initialSize_);
+ return arrow::Status::OK();
+}
+
+void VeloxSortShuffleWriter::initRowType(const facebook::velox::RowVectorPtr& rv) {
+ if (UNLIKELY(!rowType_)) {
+ rowType_ = facebook::velox::asRowType(rv->type());
+ fixedRowSize_ = facebook::velox::row::CompactRow::fixedRowSize(rowType_);
+ }
+}
+
+arrow::Result<facebook::velox::RowVectorPtr> VeloxSortShuffleWriter::getPeeledRowVector(
+ const std::shared_ptr<ColumnarBatch>& cb) {
+ if (options_.partitioning == Partitioning::kRange) {
+ auto compositeBatch = std::dynamic_pointer_cast<CompositeColumnarBatch>(cb);
+ VELOX_CHECK_NOT_NULL(compositeBatch);
+ auto batches = compositeBatch->getBatches();
+ VELOX_CHECK_EQ(batches.size(), 2);
+
+ auto pidBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[0]);
+ auto pidArr = getFirstColumn(*(pidBatch->getRowVector()));
+ RETURN_NOT_OK(partitioner_->compute(pidArr, pidBatch->numRows(), row2Partition_));
+
+ auto rvBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[1]);
+ return rvBatch->getFlattenedRowVector();
+ } else {
+ auto veloxColumnBatch = VeloxColumnarBatch::from(veloxPool_.get(), cb);
+ VELOX_CHECK_NOT_NULL(veloxColumnBatch);
+ auto rv = veloxColumnBatch->getFlattenedRowVector();
+ if (partitioner_->hasPid()) {
+ auto pidArr = getFirstColumn(*rv);
+ RETURN_NOT_OK(partitioner_->compute(pidArr, rv->size(), row2Partition_));
+ return getStrippedRowVector(*rv);
+ } else {
+ RETURN_NOT_OK(partitioner_->compute(nullptr, rv->size(), row2Partition_));
+ return rv;
+ }
+ }
+}
+
+arrow::Status VeloxSortShuffleWriter::insert(const facebook::velox::RowVectorPtr& vector, int64_t memLimit) {
+ ScopedTimer timer(&c2rTime_);
+ auto inputRows = vector->size();
+ VELOX_DCHECK_GT(inputRows, 0);
+
+ facebook::velox::row::CompactRow row(vector);
Review Comment:
Hey, @marin-ma , might be a stupid question, why we don't use compact row for rss shuffle, I see VeloxCelebornSortShuffleWriter still use PrestoPage to do serialization, is there any reason why we don't change them as well. Does PrestoPage have better performance than Compact row when doing serialization?
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implementation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r2228842835
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
Review Comment:
The sort shuffle writer convert the inputs from column to row, sort the rows on the writer side ,and convert the rows back to column on the reader side. For single partition, there's no sort operation. We directly write the input without any other operations.
> In principle, it is supported, but the performance is not as good as Hash. Is this correct?
Yes... "sorting" single partition only introduce an extra column-to-row conversion, which has no good to the performance.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2235345340
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2244738660
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1689192048
##########
cpp/velox/shuffle/VeloxShuffleReader.cc:
##########
@@ -367,47 +366,133 @@ std::shared_ptr<ColumnarBatch> VeloxColumnarBatchDeserializer::next() {
return columnarBatch;
}
-VeloxColumnarBatchDeserializerFactory::VeloxColumnarBatchDeserializerFactory(
+VeloxSortShuffleReaderDeserializer::VeloxSortShuffleReaderDeserializer(
+ std::shared_ptr<arrow::io::InputStream> in,
const std::shared_ptr<arrow::Schema>& schema,
const std::shared_ptr<arrow::util::Codec>& codec,
- const facebook::velox::common::CompressionKind veloxCompressionType,
const RowTypePtr& rowType,
int32_t batchSize,
arrow::MemoryPool* memoryPool,
- std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
- ShuffleWriterType shuffleWriterType)
- : schema_(schema),
+ facebook::velox::memory::MemoryPool* veloxPool,
+ int64_t& deserializeTime,
+ int64_t& decompressTime)
+ : in_(std::move(in)),
+ schema_(schema),
codec_(codec),
- veloxCompressionType_(veloxCompressionType),
rowType_(rowType),
batchSize_(batchSize),
- memoryPool_(memoryPool),
+ arrowPool_(memoryPool),
veloxPool_(veloxPool),
- shuffleWriterType_(shuffleWriterType) {
- initFromSchema();
+ deserializeTime_(deserializeTime),
+ decompressTime_(decompressTime) {}
+
+std::shared_ptr<ColumnarBatch> VeloxSortShuffleReaderDeserializer::next() {
+ if (reachEos_) {
+ if (cachedRows_ > 0) {
+ return deserializeToBatch();
+ }
+ return nullptr;
+ }
+
+ if (cachedRows_ >= batchSize_) {
+ return deserializeToBatch();
+ }
+
+ while (cachedRows_ < batchSize_) {
+ uint32_t numRows;
+ GLUTEN_ASSIGN_OR_THROW(
+ auto arrowBuffers, BlockPayload::deserialize(in_.get(), schema_, codec_, arrowPool_, numRows, decompressTime_));
+
+ if (numRows == 0) {
+ reachEos_ = true;
+ if (cachedRows_ > 0) {
+ return deserializeToBatch();
+ }
+ return nullptr;
+ }
+ auto buffer = std::move(arrowBuffers[0]);
+ cachedInputs_.emplace_back(numRows, wrapInBufferViewAsOwner(buffer->data(), buffer->size(), buffer));
+ cachedRows_ += numRows;
+ }
+ return deserializeToBatch();
}
-std::unique_ptr<ColumnarBatchIterator> VeloxColumnarBatchDeserializerFactory::createDeserializer(
- std::shared_ptr<arrow::io::InputStream> in) {
- if (shuffleWriterType_ == kHashShuffle) {
- return std::make_unique<VeloxColumnarBatchDeserializer>(
- std::move(in),
- schema_,
- codec_,
- rowType_,
- batchSize_,
- memoryPool_,
- veloxPool_.get(),
- &isValidityBuffer_,
- hasComplexType_,
- deserializeTime_,
- decompressTime_);
+std::shared_ptr<ColumnarBatch> VeloxSortShuffleReaderDeserializer::deserializeToBatch() {
+ ScopedTimer timer(&deserializeTime_);
+ std::vector<std::string_view> data;
+ data.reserve(std::min(cachedRows_, batchSize_));
+
+ uint32_t readRows = 0;
+ auto cur = cachedInputs_.begin();
+ while (readRows < batchSize_ && cur != cachedInputs_.end()) {
+ auto buffer = cur->second;
+ const auto* rawBuffer = buffer->as<char>();
+ while (rowOffset_ < cur->first && readRows < batchSize_) {
+ auto rowSize = *(uint32_t*)(rawBuffer + byteOffset_);
+ byteOffset_ += sizeof(uint32_t);
+ data.push_back(std::string_view(rawBuffer + byteOffset_, rowSize));
+ byteOffset_ += rowSize;
+ ++rowOffset_;
+ ++readRows;
+ }
+ if (rowOffset_ == cur->first) {
+ rowOffset_ = 0;
+ byteOffset_ = 0;
+ ++cur;
+ }
+ }
+ cachedRows_ -= readRows;
+ auto rowVector = facebook::velox::row::CompactRow::deserialize(data, rowType_, veloxPool_);
+ // Free memory.
+ auto iter = cachedInputs_.begin();
+ while (iter++ != cur) {
+ cachedInputs_.pop_front();
}
- return std::make_unique<VeloxShuffleReaderOutStreamWrapper>(
- veloxPool_, rowType_, batchSize_, veloxCompressionType_, deserializeTime_, std::move(in));
+ return std::make_shared<VeloxColumnarBatch>(std::move(rowVector));
+}
+
+class VeloxRssSortShuffleReaderDeserializer::VeloxInputStream : public facebook::velox::ByteInputStream {
+ public:
+ VeloxInputStream(std::shared_ptr<arrow::io::InputStream> input, facebook::velox::BufferPtr buffer);
+
+ bool hasNext();
+
+ void next(bool throwIfPastEnd) override;
+
+ std::shared_ptr<arrow::io::InputStream> in_;
+ const facebook::velox::BufferPtr buffer_;
+ uint64_t offset_ = -1;
+};
+
+VeloxRssSortShuffleReaderDeserializer::VeloxInputStream::VeloxInputStream(
+ std::shared_ptr<arrow::io::InputStream> input,
+ facebook::velox::BufferPtr buffer)
+ : in_(std::move(input)), buffer_(std::move(buffer)) {
+ next(true);
}
Review Comment:
nit: Better to move the member function definitions to class definition?
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
+ array_.resize(initialSize_);
+ return arrow::Status::OK();
+}
+
+void VeloxSortShuffleWriter::initRowType(const facebook::velox::RowVectorPtr& rv) {
+ if (UNLIKELY(!rowType_)) {
+ rowType_ = facebook::velox::asRowType(rv->type());
+ fixedRowSize_ = facebook::velox::row::CompactRow::fixedRowSize(rowType_);
+ }
+}
+
+arrow::Result<facebook::velox::RowVectorPtr> VeloxSortShuffleWriter::getPeeledRowVector(
+ const std::shared_ptr<ColumnarBatch>& cb) {
+ if (options_.partitioning == Partitioning::kRange) {
+ auto compositeBatch = std::dynamic_pointer_cast<CompositeColumnarBatch>(cb);
Review Comment:
It seems that the casting highly depends on the preceding code calling `ColumnarBatch#compose`. Is it possible to directly call `VeloxColumnarBatch::from` which is more general, then operate on the output Velox columnar batch? Which shouldn't bring visible overhead based on my perspective.
##########
shims/common/src/main/scala/org/apache/gluten/GlutenConfig.scala:
##########
@@ -133,10 +133,11 @@ class GlutenConfig(conf: SQLConf) extends Logging {
.getConfString("spark.shuffle.manager", "sort")
.contains("UniffleShuffleManager")
- def isSortBasedCelebornShuffle: Boolean =
+ def celebornShuffleWriterType: String =
conf
- .getConfString("spark.celeborn.client.spark.shuffle.writer", "hash")
- .equals("sort")
+ .getConfString("spark.celeborn.client.spark.shuffle.writer", GLUTEN_HASH_SHUFFLE_WRITER)
+ .toLowerCase(Locale.ROOT)
+ .replace(GLUTEN_SORT_SHUFFLE_WRITER, GLUTEN_RSS_SORT_SHUFFLE_WRITER)
Review Comment:
nit: Could move the text replacement into `VeloxCelebornColumnarBatchSerializer.scala`, so one knows it's native code that distinguishes between `sort` and `rss_sort`
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2244791739
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implemenation [incubator-gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#issuecomment-2243295933
Run Gluten Clickhouse CI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Row based sort shuffle implementation [incubator-gluten]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #6475:
URL: https://github.com/apache/incubator-gluten/pull/6475#discussion_r1977218516
##########
cpp/velox/shuffle/VeloxSortShuffleWriter.cc:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.
+ */
+
+#include "shuffle/VeloxSortShuffleWriter.h"
+
+#include <arrow/io/memory.h>
+
+#include "memory/ArrowMemory.h"
+#include "memory/VeloxColumnarBatch.h"
+#include "utils/Common.h"
+#include "utils/Timer.h"
+
+namespace gluten {
+
+namespace {
+constexpr uint32_t kMaskLower27Bits = (1 << 27) - 1;
+constexpr uint64_t kMaskLower40Bits = (1UL << 40) - 1;
+
+uint64_t toCompactRowId(uint32_t partitionId, uint32_t pageNumber, uint32_t offsetInPage) {
+ // |63 partitionId(24) |39 inputIndex(13) |26 rowIndex(27) |
+ return (uint64_t)partitionId << 40 | (uint64_t)pageNumber << 27 | offsetInPage;
+}
+
+uint32_t extractPartitionId(uint64_t compactRowId) {
+ return (uint32_t)(compactRowId >> 40);
+}
+
+std::pair<uint32_t, uint32_t> extractPageNumberAndOffset(uint64_t compactRowId) {
+ return {(compactRowId & kMaskLower40Bits) >> 27, compactRowId & kMaskLower27Bits};
+}
+} // namespace
+
+arrow::Result<std::shared_ptr<VeloxShuffleWriter>> VeloxSortShuffleWriter::create(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* arrowPool) {
+ std::shared_ptr<VeloxSortShuffleWriter> writer(new VeloxSortShuffleWriter(
+ numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), arrowPool));
+ RETURN_NOT_OK(writer->init());
+ return writer;
+}
+
+VeloxSortShuffleWriter::VeloxSortShuffleWriter(
+ uint32_t numPartitions,
+ std::unique_ptr<PartitionWriter> partitionWriter,
+ ShuffleWriterOptions options,
+ std::shared_ptr<facebook::velox::memory::MemoryPool> veloxPool,
+ arrow::MemoryPool* pool)
+ : VeloxShuffleWriter(numPartitions, std::move(partitionWriter), std::move(options), std::move(veloxPool), pool),
+ allocator_{std::make_unique<facebook::velox::HashStringAllocator>(veloxPool_.get())},
+ array_{SortArray{Allocator(allocator_.get())}} {}
+
+arrow::Status VeloxSortShuffleWriter::write(std::shared_ptr<ColumnarBatch> cb, int64_t memLimit) {
+ ARROW_ASSIGN_OR_RAISE(auto rv, getPeeledRowVector(cb));
+ initRowType(rv);
+ RETURN_NOT_OK(insert(rv, memLimit));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::stop() {
+ ARROW_RETURN_IF(evictState_ == EvictState::kUnevictable, arrow::Status::Invalid("Unevictable state in stop."));
+
+ EvictGuard evictGuard{evictState_};
+
+ stopped_ = true;
+ RETURN_NOT_OK(evictAllPartitions());
+ RETURN_NOT_OK(partitionWriter_->stop(&metrics_));
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::reclaimFixedSize(int64_t size, int64_t* actual) {
+ if (evictState_ == EvictState::kUnevictable || offset_ == 0) {
+ *actual = 0;
+ return arrow::Status::OK();
+ }
+ EvictGuard evictGuard{evictState_};
+ auto beforeReclaim = veloxPool_->usedBytes();
+ RETURN_NOT_OK(evictAllPartitions());
+ *actual = beforeReclaim - veloxPool_->usedBytes();
+ return arrow::Status::OK();
+}
+
+arrow::Status VeloxSortShuffleWriter::init() {
+ ARROW_RETURN_IF(
+ options_.partitioning == Partitioning::kSingle,
+ arrow::Status::Invalid("VeloxSortShuffleWriter doesn't support single partition."));
+ array_.resize(initialSize_);
+ return arrow::Status::OK();
+}
+
+void VeloxSortShuffleWriter::initRowType(const facebook::velox::RowVectorPtr& rv) {
+ if (UNLIKELY(!rowType_)) {
+ rowType_ = facebook::velox::asRowType(rv->type());
+ fixedRowSize_ = facebook::velox::row::CompactRow::fixedRowSize(rowType_);
+ }
+}
+
+arrow::Result<facebook::velox::RowVectorPtr> VeloxSortShuffleWriter::getPeeledRowVector(
+ const std::shared_ptr<ColumnarBatch>& cb) {
+ if (options_.partitioning == Partitioning::kRange) {
+ auto compositeBatch = std::dynamic_pointer_cast<CompositeColumnarBatch>(cb);
+ VELOX_CHECK_NOT_NULL(compositeBatch);
+ auto batches = compositeBatch->getBatches();
+ VELOX_CHECK_EQ(batches.size(), 2);
+
+ auto pidBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[0]);
+ auto pidArr = getFirstColumn(*(pidBatch->getRowVector()));
+ RETURN_NOT_OK(partitioner_->compute(pidArr, pidBatch->numRows(), row2Partition_));
+
+ auto rvBatch = VeloxColumnarBatch::from(veloxPool_.get(), batches[1]);
+ return rvBatch->getFlattenedRowVector();
+ } else {
+ auto veloxColumnBatch = VeloxColumnarBatch::from(veloxPool_.get(), cb);
+ VELOX_CHECK_NOT_NULL(veloxColumnBatch);
+ auto rv = veloxColumnBatch->getFlattenedRowVector();
+ if (partitioner_->hasPid()) {
+ auto pidArr = getFirstColumn(*rv);
+ RETURN_NOT_OK(partitioner_->compute(pidArr, rv->size(), row2Partition_));
+ return getStrippedRowVector(*rv);
+ } else {
+ RETURN_NOT_OK(partitioner_->compute(nullptr, rv->size(), row2Partition_));
+ return rv;
+ }
+ }
+}
+
+arrow::Status VeloxSortShuffleWriter::insert(const facebook::velox::RowVectorPtr& vector, int64_t memLimit) {
+ ScopedTimer timer(&c2rTime_);
+ auto inputRows = vector->size();
+ VELOX_DCHECK_GT(inputRows, 0);
+
+ facebook::velox::row::CompactRow row(vector);
Review Comment:
PrestoSerializer still serialises the data by column. We use compact row for sort based shuffle is that we found the performance drops significantly with large shuffle partitions and wide tables for both using hash-based shuffle or presto-serialiser.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org