You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by GitBox <gi...@apache.org> on 2022/11/01 11:02:20 UTC
[GitHub] [incubator-uniffle] zuston opened a new pull request, #293: [WIP][Improvement][AQE] Sort MapId before the data are flushed
zuston opened a new pull request, #293:
URL: https://github.com/apache/incubator-uniffle/pull/293
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://github.com/Tencent/Firestorm/blob/master/CONTRIBUTING.md
2. Ensure you have added or run the appropriate tests for your PR
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]XXXX Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014055605
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
This splitter is bound to the LOCAL_ORDER. I don't quite understand what you mean.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013884713
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
We have 4 blocks, block 1, block 2, block 3, block4, block 5. Block 3 is filtered. Block 1, Block 2, Block 4, Block 5 can't be ShuffleDataSegment. The blocks in one ShuffleDataSegment must be continous. You can see https://github.com/apache/incubator-uniffle/blob/c69f1735e5bf93852a82de6202e1e5b7c1bd7092/client/src/main/java/org/apache/uniffle/client/impl/ShuffleReadClientImpl.java#L182 for details.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013671653
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
So this splitter may be useless.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013799272
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -137,6 +138,8 @@ public void registerShuffle(ShuffleRegisterRequest req,
int shuffleId = req.getShuffleId();
String remoteStoragePath = req.getRemoteStorage().getPath();
String user = req.getUser();
+ ShuffleDataDistributionType shuffleDataDistributionType =
Review Comment:
Could we consider the old version client to use the new server?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013860699
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
Could you describe more in detail ?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013817095
##########
storage/src/main/java/org/apache/uniffle/storage/factory/ShuffleHandlerFactory.java:
##########
@@ -136,7 +138,11 @@ private ClientReadHandler getHdfsClientReadHandler(CreateShuffleReadHandlerReque
request.getExpectBlockIds(),
request.getProcessBlockIds(),
request.getStorageBasePath(),
- request.getHadoopConf());
+ request.getHadoopConf(),
+ request.getDistributionType(),
+ request.getStartMapIndex(),
Review Comment:
Done
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014548704
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -359,4 +359,8 @@ public long getCreateTimeStamp() {
return createTimeStamp;
}
}
+
+ public ShuffleServer getShuffleServer() {
Review Comment:
OK.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013680651
##########
common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java:
##########
@@ -35,4 +36,11 @@ public class RssClientConf {
.intType()
.defaultValue(3)
.withDescription("The zstd compression level, the default level is 3");
+
+ public static final ConfigOption<ShuffleDataDistributionType> DATA_DISTRIBUTION_TYPE = ConfigOptions
+ .key("rss.client.shuffle.data.distribution.type")
Review Comment:
Of course. It will be attached later
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013837552
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
The data of ShuffleDataSegment seems to be continuous. It will be wrong when you filter one record.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] leixm commented on pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
leixm commented on PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#issuecomment-1305529260
> cc @leixm . Maybe you have interest about this feature.
It's a pity that I was busy and didn't participate in the review.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#issuecomment-1306577253
> > cc @leixm . Maybe you have interest about this feature.
>
> It's a pity that I was busy and didn't participate in the review.
Looking forward to your ideas on this optimization.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013820251
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
In the future, maybe we could introduce the global order strategy which will be implemented in current abstraction easier.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014054198
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -359,4 +359,8 @@ public long getCreateTimeStamp() {
return createTimeStamp;
}
}
+
+ public ShuffleServer getShuffleServer() {
Review Comment:
OK. Could we expose a method of `getDistirbution()` in `ShuffleFlushManager` ?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi merged pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi merged PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013813142
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
I have removed the startMapId and endMapId.
> What's the difference between LocalOrderSegmentSplitter and FixedSizeSegementSplitter?
LocalOrderSegmentSplitter is to get the segments according to the index file local order.
FixedSizeSegementSplitter is to keep the consistent with the original segment split logic
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013811293
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -359,4 +359,8 @@ public long getCreateTimeStamp() {
return createTimeStamp;
}
}
+
+ public ShuffleServer getShuffleServer() {
Review Comment:
Done.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013837858
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
This point need to be ensured.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013681141
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private int startMapId;
+ private int endMapId;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(int startMapId, int endMapId, int readBufferSize) {
+ this.startMapId = startMapId;
+ this.endMapId = endMapId;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId]
Review Comment:
Yes mistakes
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013893526
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
It sounds we still need to use startMapId to get the data.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013647808
##########
common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java:
##########
@@ -35,4 +36,11 @@ public class RssClientConf {
.intType()
.defaultValue(3)
.withDescription("The zstd compression level, the default level is 3");
+
+ public static final ConfigOption<ShuffleDataDistributionType> DATA_DISTRIBUTION_TYPE = ConfigOptions
+ .key("rss.client.shuffle.data.distribution.type")
Review Comment:
OK, got it. Should we add some documents for this feature?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013697396
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
OK. this will avoid using the startMapId and endMapId
##########
storage/src/main/java/org/apache/uniffle/storage/factory/ShuffleHandlerFactory.java:
##########
@@ -136,7 +138,11 @@ private ClientReadHandler getHdfsClientReadHandler(CreateShuffleReadHandlerReque
request.getExpectBlockIds(),
request.getProcessBlockIds(),
request.getStorageBasePath(),
- request.getHadoopConf());
+ request.getHadoopConf(),
+ request.getDistributionType(),
+ request.getStartMapIndex(),
Review Comment:
+1.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014627040
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
I rethink the discontinuous problem you mentioned will only occur if the block's taskId is not in this reader partition and this taskId is in the `expectTaskIds` range in this index file. like this index file, its blocks are as follow:
```
Block-A (partition: 1, taskId: 1)
Block-B (partition: 2, taskId: 1)
Block-C (partition: 1, taskId: 2)
```
If the reader want to get the partition-1 data and the taskId range is in [1, 3). And so the Block-B will be filtered, which will cause discontinuous problem.
I think we need to sort the index file by two factors, first one is partitionId, second one is the taskId.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013825207
##########
common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java:
##########
@@ -35,4 +36,11 @@ public class RssClientConf {
.intType()
.defaultValue(3)
.withDescription("The zstd compression level, the default level is 3");
+
+ public static final ConfigOption<ShuffleDataDistributionType> DATA_DISTRIBUTION_TYPE = ConfigOptions
+ .key("rss.client.shuffle.data.distribution.type")
Review Comment:
Done
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013893526
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
It sounds we still need to use startMapId to get the data.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013585449
##########
common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java:
##########
@@ -35,4 +36,11 @@ public class RssClientConf {
.intType()
.defaultValue(3)
.withDescription("The zstd compression level, the default level is 3");
+
+ public static final ConfigOption<ShuffleDataDistributionType> DATA_DISTRIBUTION_TYPE = ConfigOptions
+ .key("rss.client.shuffle.data.distribution.type")
Review Comment:
What's the relation with `aqe` config option?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#issuecomment-1303014555
cc @leixm . Maybe you have interest about this feature.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013825512
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -137,6 +138,8 @@ public void registerShuffle(ShuffleRegisterRequest req,
int shuffleId = req.getShuffleId();
String remoteStoragePath = req.getRemoteStorage().getPath();
String user = req.getUser();
+ ShuffleDataDistributionType shuffleDataDistributionType =
Review Comment:
Done
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013951662
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
It seems that my worried case won't happen because our data is sorted. But we also need to check this point again and add some 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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014563402
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -131,12 +132,13 @@ private ShuffleRegisterResponse doRegisterShuffle(
int shuffleId,
List<PartitionRange> partitionRanges,
RemoteStorageInfo remoteStorageInfo,
- String user) {
+ String user, ShuffleDataDistributionType dataDistributionType) {
Review Comment:
new line?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013799272
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -137,6 +138,8 @@ public void registerShuffle(ShuffleRegisterRequest req,
int shuffleId = req.getShuffleId();
String remoteStoragePath = req.getRemoteStorage().getPath();
String user = req.getUser();
+ ShuffleDataDistributionType shuffleDataDistributionType =
Review Comment:
Could we consider that users use the old version client to connect the new server?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013590310
##########
common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java:
##########
@@ -35,4 +36,11 @@ public class RssClientConf {
.intType()
.defaultValue(3)
.withDescription("The zstd compression level, the default level is 3");
+
+ public static final ConfigOption<ShuffleDataDistributionType> DATA_DISTRIBUTION_TYPE = ConfigOptions
+ .key("rss.client.shuffle.data.distribution.type")
Review Comment:
The local order is just a type, should not be bound to any spark optimization. Although it will benefit for spark AQE.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013671653
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
So this splitter may be useless.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013697396
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
OK. this will avoid using the startMapId and endMapId
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
Let me think twice.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013815827
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -359,4 +359,8 @@ public long getCreateTimeStamp() {
return createTimeStamp;
}
}
+
+ public ShuffleServer getShuffleServer() {
Review Comment:
I think the distribution should be bound to the app, which managed in ShuffleTaskManager will be better
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014115486
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
ExpectTaskId should be a range. The range is between startMapId and endMapId. If a blockId A isn't in this range, there won't be any other blockId after blockId A in this ShuffleDataSegement. Because data is local order in the ShuffleDataSegment. So there won't discontinuous blockIds in a ShuffleDataSegement. The discontinuous blockIds in a ShuffleDataSegment isn't permitted in our system. You can see related logic for details in https://github.com/apache/incubator-uniffle/blob/c69f1735e5bf93852a82de6202e1e5b7c1bd7092/client/src/main/java/org/apache/uniffle/client/impl/ShuffleReadClientImpl.java#L182
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] codecov-commenter commented on pull request #293: [WIP][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#issuecomment-1299537713
# [Codecov](https://codecov.io/gh/apache/incubator-uniffle/pull/293?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#293](https://codecov.io/gh/apache/incubator-uniffle/pull/293?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (41723f0) into [master](https://codecov.io/gh/apache/incubator-uniffle/commit/7d4428ed22055f5f76db7b9e289945d276ddf716?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7d4428e) will **decrease** coverage by `1.07%`.
> The diff coverage is `66.66%`.
```diff
@@ Coverage Diff @@
## master #293 +/- ##
============================================
- Coverage 60.10% 59.03% -1.08%
+ Complexity 1413 1355 -58
============================================
Files 175 170 -5
Lines 9082 8697 -385
Branches 872 848 -24
============================================
- Hits 5459 5134 -325
+ Misses 3331 3276 -55
+ Partials 292 287 -5
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-uniffle/pull/293?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...apache/hadoop/mapreduce/v2/app/RssMRAppMaster.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkdWNlL3YyL2FwcC9Sc3NNUkFwcE1hc3Rlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...e/uniffle/client/factory/ShuffleClientFactory.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NsaWVudC9mYWN0b3J5L1NodWZmbGVDbGllbnRGYWN0b3J5LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...he/uniffle/client/impl/ShuffleWriteClientImpl.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NsaWVudC9pbXBsL1NodWZmbGVXcml0ZUNsaWVudEltcGwuamF2YQ==) | `21.23% <ø> (ø)` | |
| [...client/request/CreateShuffleReadClientRequest.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NsaWVudC9yZXF1ZXN0L0NyZWF0ZVNodWZmbGVSZWFkQ2xpZW50UmVxdWVzdC5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...uniffle/common/segment/SegmentSplitterFactory.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9zZWdtZW50L1NlZ21lbnRTcGxpdHRlckZhY3RvcnkuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [.../java/org/apache/uniffle/common/util/RssUtils.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi91dGlsL1Jzc1V0aWxzLmphdmE=) | `63.07% <ø> (-5.98%)` | :arrow_down: |
| [...pache/uniffle/server/ShuffleServerGrpcService.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9TaHVmZmxlU2VydmVyR3JwY1NlcnZpY2UuamF2YQ==) | `0.87% <0.00%> (-0.01%)` | :arrow_down: |
| [...uniffle/storage/factory/ShuffleHandlerFactory.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3RvcmFnZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvdW5pZmZsZS9zdG9yYWdlL2ZhY3RvcnkvU2h1ZmZsZUhhbmRsZXJGYWN0b3J5LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...handler/impl/LocalFileQuorumClientReadHandler.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3RvcmFnZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvdW5pZmZsZS9zdG9yYWdlL2hhbmRsZXIvaW1wbC9Mb2NhbEZpbGVRdW9ydW1DbGllbnRSZWFkSGFuZGxlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...orage/request/CreateShuffleReadHandlerRequest.java](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3RvcmFnZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvdW5pZmZsZS9zdG9yYWdlL3JlcXVlc3QvQ3JlYXRlU2h1ZmZsZVJlYWRIYW5kbGVyUmVxdWVzdC5qYXZh) | `0.00% <0.00%> (ø)` | |
| ... and [24 more](https://codecov.io/gh/apache/incubator-uniffle/pull/293/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013656207
##########
storage/src/main/java/org/apache/uniffle/storage/factory/ShuffleHandlerFactory.java:
##########
@@ -136,7 +138,11 @@ private ClientReadHandler getHdfsClientReadHandler(CreateShuffleReadHandlerReque
request.getExpectBlockIds(),
request.getProcessBlockIds(),
request.getStorageBasePath(),
- request.getHadoopConf());
+ request.getHadoopConf(),
+ request.getDistributionType(),
+ request.getStartMapIndex(),
Review Comment:
Do ClientReadHandler need startMapIndex and EndMapIndex?
I think it's a Spark AQE concept, I prefer not exposing them?
What do you think?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013657155
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private int startMapId;
+ private int endMapId;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(int startMapId, int endMapId, int readBufferSize) {
+ this.startMapId = startMapId;
+ this.endMapId = endMapId;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId]
Review Comment:
[startMapId, endMapId)
?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013671450
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
Our blockId have the taskAttemptId information. We usually use the blockId to filter the data. You can see
https://github.com/apache/incubator-uniffle/blob/47effb25044780013ff51cc516261464435b5829/client/src/main/java/org/apache/uniffle/client/impl/ShuffleReadClientImpl.java#L101
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013900711
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
> It sounds we still need to use startMapId to get the data.
startMapId will also filter the data. It's the same.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014115486
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
ExpectTaskId should be a range. The range is between startMapId and endMapId. If a blockId A isn't in this range, there won't be any other blockId after blockId A in this ShuffleDataSegement. Because data is local order in the ShuffleDataSegment. So there won't discontinuous blockIds in a ShuffleDataSegement.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013827816
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -359,4 +359,8 @@ public long getCreateTimeStamp() {
return createTimeStamp;
}
}
+
+ public ShuffleServer getShuffleServer() {
Review Comment:
It's better to managed by ShuffleTaskManager. But I wanna to avoid using `getShuffleServer`.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014136222
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
Got it. This won’t happen, the index data will be split to multi ordered parts. And then to get the segments in the range by expectTaskIds.
Maybe I need to add more comment on this class header.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013806071
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
What's the difference between LocalOrderSegmentSplitter?
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
Review Comment:
What's the difference between LocalOrderSegmentSplitter and FixedSizeSegementSplitter?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013805081
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -359,4 +359,8 @@ public long getCreateTimeStamp() {
return createTimeStamp;
}
}
+
+ public ShuffleServer getShuffleServer() {
Review Comment:
Em... maybe we can add a `distribution` field in ShuffleBuffer, `getShuffleServer` may expose too much information.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013884713
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.roaringbitmap.longlong.Roaring64NavigableMap;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private Roaring64NavigableMap expectTaskIds;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(Roaring64NavigableMap expectTaskIds, int readBufferSize) {
+ this.expectTaskIds = expectTaskIds;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId) taskIds bitmap
+ * 2. ShuffleDataSegment size should < readBufferSize
+ * 3. ShuffleDataSegment's blocks should be continuous
+ *
+ */
+ while (byteBuffer.hasRemaining()) {
+ try {
+ long offset = byteBuffer.getLong();
+ int length = byteBuffer.getInt();
+ int uncompressLength = byteBuffer.getInt();
+ long crc = byteBuffer.getLong();
+ long blockId = byteBuffer.getLong();
+ long taskAttemptId = byteBuffer.getLong();
+
+ if (lastTaskAttemptId == -1) {
+ lastTaskAttemptId = taskAttemptId;
+ }
+
+ // If ShuffleServer is flushing the file at this time, the length in the index file record may be greater
+ // than the length in the actual data file, and it needs to be returned at this time to avoid EOFException
+ if (dataFileLen != -1 && totalLen >= dataFileLen) {
+ break;
+ }
+
+ if ((taskAttemptId < lastTaskAttemptId && bufferSegments.size() > 0) || bufferOffset >= readBufferSize) {
+ ShuffleDataSegment sds = new ShuffleDataSegment(fileOffset, bufferOffset, bufferSegments);
+ dataFileSegments.add(sds);
+ bufferSegments = Lists.newArrayList();
+ bufferOffset = 0;
+ fileOffset = -1;
+ }
+
+ if (expectTaskIds.contains(taskAttemptId)) {
Review Comment:
We have 4 blocks, block 1, block 2, block 3, block4, block 5. Block 3 is filtered. Block 1, Block 2, Block 4, Block 4 can't be ShuffleDataSegment. The blocks in one ShuffleDataSegment must be continous. You can see https://github.com/apache/incubator-uniffle/blob/c69f1735e5bf93852a82de6202e1e5b7c1bd7092/client/src/main/java/org/apache/uniffle/client/impl/ShuffleReadClientImpl.java#L182 for details.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013816747
##########
server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java:
##########
@@ -137,6 +138,8 @@ public void registerShuffle(ShuffleRegisterRequest req,
int shuffleId = req.getShuffleId();
String remoteStoragePath = req.getRemoteStorage().getPath();
String user = req.getUser();
+ ShuffleDataDistributionType shuffleDataDistributionType =
Review Comment:
Oh. I missed this point. Let me do it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013825023
##########
common/src/main/java/org/apache/uniffle/common/segment/LocalOrderSegmentSplitter.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.common.segment;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+import org.apache.uniffle.common.BufferSegment;
+import org.apache.uniffle.common.ShuffleDataSegment;
+import org.apache.uniffle.common.ShuffleIndexResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public class LocalOrderSegmentSplitter implements SegmentSplitter {
+
+ private int startMapId;
+ private int endMapId;
+ private int readBufferSize;
+
+ public LocalOrderSegmentSplitter(int startMapId, int endMapId, int readBufferSize) {
+ this.startMapId = startMapId;
+ this.endMapId = endMapId;
+ this.readBufferSize = readBufferSize;
+ }
+
+ @Override
+ public List<ShuffleDataSegment> split(ShuffleIndexResult shuffleIndexResult) {
+ if (shuffleIndexResult == null || shuffleIndexResult.isEmpty()) {
+ return Lists.newArrayList();
+ }
+
+ byte[] indexData = shuffleIndexResult.getIndexData();
+ long dataFileLen = shuffleIndexResult.getDataFileLen();
+
+ ByteBuffer byteBuffer = ByteBuffer.wrap(indexData);
+ List<BufferSegment> bufferSegments = Lists.newArrayList();
+
+ List<ShuffleDataSegment> dataFileSegments = Lists.newArrayList();
+ int bufferOffset = 0;
+ long fileOffset = -1;
+ long totalLen = 0;
+
+ long lastTaskAttemptId = -1;
+
+ /**
+ * One ShuffleDataSegment should meet following requirements:
+ *
+ * 1. taskId in [startMapId, endMapId]
Review Comment:
Done
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014588117
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java:
##########
@@ -131,12 +132,13 @@ private ShuffleRegisterResponse doRegisterShuffle(
int shuffleId,
List<PartitionRange> partitionRanges,
RemoteStorageInfo remoteStorageInfo,
- String user) {
+ String user, ShuffleDataDistributionType dataDistributionType) {
Review Comment:
Done
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1013697507
##########
storage/src/main/java/org/apache/uniffle/storage/factory/ShuffleHandlerFactory.java:
##########
@@ -136,7 +138,11 @@ private ClientReadHandler getHdfsClientReadHandler(CreateShuffleReadHandlerReque
request.getExpectBlockIds(),
request.getProcessBlockIds(),
request.getStorageBasePath(),
- request.getHadoopConf());
+ request.getHadoopConf(),
+ request.getDistributionType(),
+ request.getStartMapIndex(),
Review Comment:
+1.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #293: [ISSUE-137][Improvement][AQE] Sort MapId before the data are flushed
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #293:
URL: https://github.com/apache/incubator-uniffle/pull/293#discussion_r1014552386
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -359,4 +359,8 @@ public long getCreateTimeStamp() {
return createTimeStamp;
}
}
+
+ public ShuffleServer getShuffleServer() {
Review Comment:
Done
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org