You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/09/08 17:29:10 UTC
[GitHub] [flink] dawidwys opened a new pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
dawidwys opened a new pull request #13357:
URL: https://github.com/apache/flink/pull/13357
## What is the purpose of the change
This is a preparation for introducing a sorted inputs in BATCH execution
mode. It refactors the UnilateralSortMerger into more composable pieces.
Additionally it adds the option to use a push-based approach instead of
spawning a Reader thread that consumes from an input iterator.
The PR still has a couple of shortcomings:
* the combine function is not being opened (however it did not work well even before that, the function was being closed before the merge stage)
* it misses tests for the push-based approach (however the reading thread simply encapsulates the push-based producer)
* it misses a test for testing closing all threads on an exception (the exception handler was reworked)
## Verifying this change
This change is already covered by existing tests.
Some tests are still to be added.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (**yes** / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (not applicable / docs / **JavaDocs** / not documented)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys closed pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys closed pull request #13357:
URL: https://github.com/apache/flink/pull/13357
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492522939
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CircularQueues.java
##########
@@ -72,16 +76,23 @@ public void sendResult(MutableObjectIterator<E> result) {
}
@Override
- public CircularElement<E> take(StageRunner.SortStage stage) {
- try {
- return getQueue(stage).take();
- } catch (InterruptedException e) {
- throw new WrappingRuntimeException(e);
+ public CircularElement<E> take(StageRunner.SortStage stage) throws InterruptedException {
+ while (!isFinished) {
+ CircularElement<E> value = getQueue(stage).poll(1, TimeUnit.SECONDS);
+ if (value != null) {
+ return value;
+ }
}
+ throw new FlinkRuntimeException("The sorter is closed already");
}
@Override
public CircularElement<E> poll(StageRunner.SortStage stage) {
return getQueue(stage).poll();
}
+
+ @Override
+ public void close() throws Exception {
Review comment:
I am so sorry about that. I completely forgot about calling this method in the `ExternalSorter#close`. I added it there.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-690119651
Is this class only used in the DataSet API, or is it also used in the SQL runtime (batch case)?
If the latter is true, we should verify that there is no performance impact through this refactoring by running a benchmark before/after.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 097e75ef7eaa295e31e8acb6fca83a2c4a254b41 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716)
* 0293e931b17bd770b7a982f0982ab273400cdb70 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaoyunhaii commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492529818
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CircularQueues.java
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Collection of queues that are used for the communication between the threads.
+ */
+final class CircularQueues<E> implements StageRunner.StageMessageDispatcher<E> {
+ private final BlockingQueue<CircularElement<E>> empty;
+ private final BlockingQueue<CircularElement<E>> sort;
+ private final BlockingQueue<CircularElement<E>> spill;
+
+ private boolean isFinished = false;
Review comment:
Very sorry that there is one issue missing, but it just come to me that we might need to make the `isFinished` to be `volatile`: the `CircularQueues#close()` might be called from `SpilledThread` or `SortThread`, and we might need to read this variable in the task main thread. Since `ExternalSorter#close()` guarantees that there should be no concurrent writer, therefore where we need only `volatile` instead of atomic variables here.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r493267562
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SortingThread.java
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that sorts filled buffers.
+ */
+class SortingThread<E> extends ThreadBase<E> {
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SortingThread.class);
+
+ private final IndexedSorter sorter;
+
+ /**
+ * Creates a new sorting thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ public SortingThread(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher) {
+ super(exceptionHandler, "SortMerger sorting thread", dispatcher);
+
+ // members
+ this.sorter = new QuickSort();
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException, InterruptedException {
+ boolean alive = true;
+
+ // loop as long as the thread is marked alive
+ while (isRunning() && alive) {
+ CircularElement<E> element = this.dispatcher.take(SortStage.SORT);
Review comment:
This could become `final`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SortingThread.java
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that sorts filled buffers.
+ */
+class SortingThread<E> extends ThreadBase<E> {
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SortingThread.class);
+
+ private final IndexedSorter sorter;
+
+ /**
+ * Creates a new sorting thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ public SortingThread(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher) {
+ super(exceptionHandler, "SortMerger sorting thread", dispatcher);
+
+ // members
+ this.sorter = new QuickSort();
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException, InterruptedException {
Review comment:
It seems you can remove the `IOException` now.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,575 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * A method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * A method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
Review comment:
I think this should become `volatile` because it is called in `close()` and in `go()` which run on/ are called by different threads.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ThreadBase.java
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Base class for all working threads in this sort-merger. The specific threads for reading, sorting, spilling,
+ * merging, etc... extend this subclass.
+ * <p>
+ * The threads are designed to terminate themselves when the task they are set up to do is completed. Further more,
+ * they terminate immediately when the <code>shutdown()</code> method is called.
+ */
+abstract class ThreadBase<E> extends Thread implements Thread.UncaughtExceptionHandler, StageRunner {
+
+ /**
+ * The queue of empty buffer that can be used for reading;
+ */
+ protected final StageMessageDispatcher<E> dispatcher;
+
+ /**
+ * The exception handler for any problems.
+ */
+ private final ExceptionHandler<IOException> exceptionHandler;
+
+ /**
+ * The flag marking this thread as alive.
+ */
+ private volatile boolean alive;
+
+ /**
+ * Creates a new thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param name The name of the thread.
+ * @param queues The queues used to pass buffers between the threads.
+ */
+ protected ThreadBase(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ String name,
+ StageMessageDispatcher<E> queues) {
+ // thread setup
+ super(checkNotNull(name));
+ this.setDaemon(true);
+
+ // exception handling
+ this.exceptionHandler = exceptionHandler;
+ this.setUncaughtExceptionHandler(this);
+
+ this.dispatcher = checkNotNull(queues);
+ this.alive = true;
+ }
+
+ /**
+ * Implements exception handling and delegates to go().
+ */
+ public void run() {
+ try {
+ go();
+ }
+ catch (Throwable t) {
+ internalHandleException(new IOException("Thread '" + getName() + "' terminated due to an exception: "
+ + t.getMessage(), t));
+ }
+ }
+
+ /**
+ * Equivalent to the run() method.
+ *
+ * @throws IOException Exceptions that prohibit correct completion of the work may be thrown by the thread.
+ */
+ protected abstract void go() throws IOException, InterruptedException;
+
+ /**
+ * Checks whether this thread is still alive.
+ *
+ * @return true, if the thread is alive, false otherwise.
+ */
+ public boolean isRunning() {
Review comment:
This can become `protected`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,575 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * A method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * A method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ SpillingThread(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher,
+ MemoryManager memManager,
+ IOManager ioManager,
+ TypeSerializer<E> serializer,
+ TypeComparator<E> comparator,
+ List<MemorySegment> sortReadMemory,
+ List<MemorySegment> writeMemory,
+ int maxNumFileHandles,
+ SpillChannelManager spillingChannelManager,
+ @Nullable LargeRecordHandler<E> largeRecordHandler,
+ SpillingBehaviour<E> spillingBehaviour,
+ int minNumWriteBuffers,
+ int maxNumWriteBuffers) {
+ super(exceptionHandler, "SortMerger spilling thread", dispatcher);
+ this.memManager = checkNotNull(memManager);
+ this.ioManager = checkNotNull(ioManager);
+ this.serializer = checkNotNull(serializer);
+ this.comparator = checkNotNull(comparator);
+ this.mergeReadMemory = checkNotNull(sortReadMemory);
+ this.writeMemory = checkNotNull(writeMemory);
+ this.maxFanIn = maxNumFileHandles;
+ this.spillChannelManager = checkNotNull(spillingChannelManager);
+ this.largeRecordHandler = largeRecordHandler;
+ this.spillingBehaviour = checkNotNull(spillingBehaviour);
+ this.minNumWriteBuffers = minNumWriteBuffers;
+ this.maxNumWriteBuffers = maxNumWriteBuffers;
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException, InterruptedException {
+
+ final Queue<CircularElement<E>> cache = new ArrayDeque<>();
+ boolean cacheOnly = readCache(cache);
+
+ // check whether the thread was canceled
+ if (!isRunning()) {
+ return;
+ }
+
+ MutableObjectIterator<E> largeRecords = null;
+
+ // check if we can stay in memory with the large record handler
+ if (cacheOnly && largeRecordHandler != null && largeRecordHandler.hasData()) {
+ List<MemorySegment> memoryForLargeRecordSorting = new ArrayList<>();
+
+ CircularElement<E> circElement;
+ while ((circElement = this.dispatcher.poll(SortStage.READ)) != null) {
+ circElement.getBuffer().dispose();
+ memoryForLargeRecordSorting.addAll(circElement.getMemory());
+ }
+
+ if (memoryForLargeRecordSorting.isEmpty()) {
+ cacheOnly = false;
+ LOG.debug("Going to disk-based merge because of large records.");
+ } else {
+ LOG.debug("Sorting large records, to add them to in-memory merge.");
+ largeRecords = largeRecordHandler.finishWriteAndSortKeys(memoryForLargeRecordSorting);
+ }
+ }
+
+ // ------------------- In-Memory Merge ------------------------
+ if (cacheOnly) {
+ mergeInMemory(cache, largeRecords);
+ return;
+ }
+
+ // ------------------- Spilling Phase ------------------------
+ List<ChannelWithBlockCount> channelIDs = startSpilling(cache);
+
+ // ------------------- Merging Phase ------------------------
+
+ mergeOnDisk(channelIDs);
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ super.close();
+ if (spillingBehaviourOpened) {
+ this.spillingBehaviour.close();
+ this.spillingBehaviourOpened = false;
+ }
+ }
+
+ private boolean readCache(Queue<CircularElement<E>> cache) throws InterruptedException {
+ CircularElement<E> element;
Review comment:
This can go in the `while()` loop now and become `final`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,575 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * A method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * A method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ SpillingThread(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher,
+ MemoryManager memManager,
+ IOManager ioManager,
+ TypeSerializer<E> serializer,
+ TypeComparator<E> comparator,
+ List<MemorySegment> sortReadMemory,
+ List<MemorySegment> writeMemory,
+ int maxNumFileHandles,
+ SpillChannelManager spillingChannelManager,
+ @Nullable LargeRecordHandler<E> largeRecordHandler,
+ SpillingBehaviour<E> spillingBehaviour,
+ int minNumWriteBuffers,
+ int maxNumWriteBuffers) {
+ super(exceptionHandler, "SortMerger spilling thread", dispatcher);
+ this.memManager = checkNotNull(memManager);
+ this.ioManager = checkNotNull(ioManager);
+ this.serializer = checkNotNull(serializer);
+ this.comparator = checkNotNull(comparator);
+ this.mergeReadMemory = checkNotNull(sortReadMemory);
+ this.writeMemory = checkNotNull(writeMemory);
+ this.maxFanIn = maxNumFileHandles;
+ this.spillChannelManager = checkNotNull(spillingChannelManager);
+ this.largeRecordHandler = largeRecordHandler;
+ this.spillingBehaviour = checkNotNull(spillingBehaviour);
+ this.minNumWriteBuffers = minNumWriteBuffers;
+ this.maxNumWriteBuffers = maxNumWriteBuffers;
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException, InterruptedException {
+
+ final Queue<CircularElement<E>> cache = new ArrayDeque<>();
+ boolean cacheOnly = readCache(cache);
+
+ // check whether the thread was canceled
+ if (!isRunning()) {
+ return;
+ }
+
+ MutableObjectIterator<E> largeRecords = null;
+
+ // check if we can stay in memory with the large record handler
+ if (cacheOnly && largeRecordHandler != null && largeRecordHandler.hasData()) {
+ List<MemorySegment> memoryForLargeRecordSorting = new ArrayList<>();
+
+ CircularElement<E> circElement;
+ while ((circElement = this.dispatcher.poll(SortStage.READ)) != null) {
+ circElement.getBuffer().dispose();
+ memoryForLargeRecordSorting.addAll(circElement.getMemory());
+ }
+
+ if (memoryForLargeRecordSorting.isEmpty()) {
+ cacheOnly = false;
+ LOG.debug("Going to disk-based merge because of large records.");
+ } else {
+ LOG.debug("Sorting large records, to add them to in-memory merge.");
+ largeRecords = largeRecordHandler.finishWriteAndSortKeys(memoryForLargeRecordSorting);
+ }
+ }
+
+ // ------------------- In-Memory Merge ------------------------
+ if (cacheOnly) {
+ mergeInMemory(cache, largeRecords);
+ return;
+ }
+
+ // ------------------- Spilling Phase ------------------------
+ List<ChannelWithBlockCount> channelIDs = startSpilling(cache);
+
+ // ------------------- Merging Phase ------------------------
+
+ mergeOnDisk(channelIDs);
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ super.close();
+ if (spillingBehaviourOpened) {
+ this.spillingBehaviour.close();
+ this.spillingBehaviourOpened = false;
+ }
+ }
+
+ private boolean readCache(Queue<CircularElement<E>> cache) throws InterruptedException {
+ CircularElement<E> element;
+
+ // ------------------- In-Memory Cache ------------------------
+ // fill cache
+ while (isRunning()) {
+ // take next element from queue
+ element = this.dispatcher.take(SortStage.SPILL);
+
+ if (element == SPILLING_MARKER) {
+ return false;
+ } else if (element == EOF_MARKER) {
+ return true;
+ }
+ cache.add(element);
+ }
+ return false;
+ }
+
+ private void mergeOnDisk(List<ChannelWithBlockCount> channelIDs) throws IOException {
+ // make sure we have enough memory to merge and for large record handling
+ List<MemorySegment> mergeReadMemory;
+ MutableObjectIterator<E> largeRecords = null;
+
+ if (largeRecordHandler != null && largeRecordHandler.hasData()) {
+
+ List<MemorySegment> longRecMem;
+ if (channelIDs.isEmpty()) {
+ // only long records
+ longRecMem = this.mergeReadMemory;
+ mergeReadMemory = Collections.emptyList();
+ }
+ else {
+ int maxMergedStreams = Math.min(this.maxFanIn, channelIDs.size());
+
+ int pagesPerStream = Math.max(
+ minNumWriteBuffers,
+ Math.min(maxNumWriteBuffers, this.mergeReadMemory.size() / 2 / maxMergedStreams));
+
+ int totalMergeReadMemory = maxMergedStreams * pagesPerStream;
+
+ // grab the merge memory
+ mergeReadMemory = new ArrayList<>(totalMergeReadMemory);
+ for (int i = 0; i < totalMergeReadMemory; i++) {
+ mergeReadMemory.add(this.mergeReadMemory.get(i));
+ }
+
+ // the remainder of the memory goes to the long record sorter
+ longRecMem = new ArrayList<>();
+ for (int i = totalMergeReadMemory; i < this.mergeReadMemory.size(); i++) {
+ longRecMem.add(this.mergeReadMemory.get(i));
+ }
+ }
+
+ LOG.debug("Sorting keys for large records.");
+ largeRecords = largeRecordHandler.finishWriteAndSortKeys(longRecMem);
+ }
+ else {
+ mergeReadMemory = this.mergeReadMemory;
+ }
+
+ // merge channels until sufficient file handles are available
+ while (isRunning() && channelIDs.size() > this.maxFanIn) {
+ channelIDs = mergeChannelList(channelIDs, mergeReadMemory, this.writeMemory);
+ }
+
+ // from here on, we won't write again
+ this.memManager.release(this.writeMemory);
+ this.writeMemory.clear();
+
+ // check if we have spilled some data at all
+ if (channelIDs.isEmpty()) {
+ if (largeRecords == null) {
+ this.dispatcher.sendResult(EmptyMutableObjectIterator.get());
+ } else {
+ this.dispatcher.sendResult(largeRecords);
+ }
+ }
+ else {
+ LOG.debug("Beginning final merge.");
+
+ // allocate the memory for the final merging step
+ List<List<MemorySegment>> readBuffers = new ArrayList<>(channelIDs.size());
+
+ // allocate the read memory and register it to be released
+ getSegmentsForReaders(readBuffers, mergeReadMemory, channelIDs.size());
+
+ // get the readers and register them to be released
+ this.dispatcher.sendResult(getMergingIterator(channelIDs, readBuffers,
+ new ArrayList<>(channelIDs.size()), largeRecords));
+ }
+
+ // done
+ LOG.debug("Spilling and merging thread done.");
+ }
+
+ private void mergeInMemory(
+ Queue<CircularElement<E>> cache,
+ MutableObjectIterator<E> largeRecords) throws IOException {
+ // operates on in-memory buffers only
+ LOG.debug("Initiating in memory merge.");
+
+ List<MutableObjectIterator<E>> iterators = new ArrayList<>(cache.size() + 1);
+
+ // iterate buffers and collect a set of iterators
+ for (CircularElement<E> cached : cache) {
+ // note: the yielded iterator only operates on the buffer heap (and disregards the stack)
+ iterators.add(cached.getBuffer().getIterator());
+ }
+
+ if (largeRecords != null) {
+ iterators.add(largeRecords);
+ }
+
+ // release the remaining sort-buffers
+ LOG.debug("Releasing unused sort-buffer memory.");
+ disposeSortBuffers(true);
+
+ // set lazy iterator
+ this.dispatcher.sendResult(iterators.isEmpty() ? EmptyMutableObjectIterator.get() :
+ iterators.size() == 1 ? iterators.get(0) :
+ new MergeIterator<>(iterators, this.comparator));
+ }
+
+ private List<ChannelWithBlockCount> startSpilling(Queue<CircularElement<E>> cache) throws IOException, InterruptedException {
+ final FileIOChannel.Enumerator enumerator = this.ioManager.createChannelEnumerator();
+ List<ChannelWithBlockCount> channelIDs = new ArrayList<>();
+
+ // loop as long as the thread is marked alive and we do not see the final element
+ CircularElement<E> element;
Review comment:
This can go in the `while()` loop now and become `final`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,575 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * A method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * A method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ SpillingThread(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher,
+ MemoryManager memManager,
+ IOManager ioManager,
+ TypeSerializer<E> serializer,
+ TypeComparator<E> comparator,
+ List<MemorySegment> sortReadMemory,
+ List<MemorySegment> writeMemory,
+ int maxNumFileHandles,
+ SpillChannelManager spillingChannelManager,
+ @Nullable LargeRecordHandler<E> largeRecordHandler,
+ SpillingBehaviour<E> spillingBehaviour,
+ int minNumWriteBuffers,
+ int maxNumWriteBuffers) {
+ super(exceptionHandler, "SortMerger spilling thread", dispatcher);
+ this.memManager = checkNotNull(memManager);
+ this.ioManager = checkNotNull(ioManager);
+ this.serializer = checkNotNull(serializer);
+ this.comparator = checkNotNull(comparator);
+ this.mergeReadMemory = checkNotNull(sortReadMemory);
+ this.writeMemory = checkNotNull(writeMemory);
+ this.maxFanIn = maxNumFileHandles;
+ this.spillChannelManager = checkNotNull(spillingChannelManager);
+ this.largeRecordHandler = largeRecordHandler;
+ this.spillingBehaviour = checkNotNull(spillingBehaviour);
+ this.minNumWriteBuffers = minNumWriteBuffers;
+ this.maxNumWriteBuffers = maxNumWriteBuffers;
+ }
+
+ /**
+ * Entry point of the thread.
+ */
Review comment:
Probably add the `@Override`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SortingThread.java
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that sorts filled buffers.
+ */
+class SortingThread<E> extends ThreadBase<E> {
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SortingThread.class);
+
+ private final IndexedSorter sorter;
+
+ /**
+ * Creates a new sorting thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ public SortingThread(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher) {
+ super(exceptionHandler, "SortMerger sorting thread", dispatcher);
+
+ // members
+ this.sorter = new QuickSort();
+ }
+
+ /**
+ * Entry point of the thread.
+ */
Review comment:
I would also add the `@Override` annotation.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ReadingThread.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The thread that consumes the input data and puts it into a buffer that will be sorted.
+ */
+final class ReadingThread<E> extends ThreadBase<E> {
+
+ /** The input channels to read from. */
+ private final MutableObjectIterator<E> reader;
+
+ /** The object into which the thread reads the data from the input. */
+ private final E readTarget;
+
+ private final SorterInputGateway<E> sorterGateway;
+
+ /**
+ * Creates a new reading thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param reader The reader to pull the data from.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ ReadingThread(
+ @Nullable ExceptionHandler<IOException> exceptionHandler,
+ MutableObjectIterator<E> reader,
+ StageMessageDispatcher<E> dispatcher,
+ @Nullable LargeRecordHandler<E> largeRecordsHandler,
+ @Nullable E readTarget,
+ long startSpillingBytes) {
+ super(exceptionHandler, "SortMerger Reading Thread", dispatcher);
+
+ // members
+ this.sorterGateway = new SorterInputGateway<>(dispatcher, largeRecordsHandler, startSpillingBytes);
+ this.reader = checkNotNull(reader);
+ this.readTarget = readTarget;
+ }
+
Review comment:
Probably add `@Override`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689028555
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 54ee6987c7054a42fa52d04ec61fae74a816792c (Tue Sep 08 17:31:51 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684)
* 097e75ef7eaa295e31e8acb6fca83a2c4a254b41 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716)
* 0293e931b17bd770b7a982f0982ab273400cdb70 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492174885
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * An method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ /**
+ * Creates the spilling thread.
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ * @param memManager The memory manager used to allocate buffers for the readers and writers.
+ * @param ioManager The I/I manager used to instantiate readers and writers from.
+ * @param serializer
+ * @param comparator
+ * @param sortReadMemory
+ * @param writeMemory
+ * @param maxNumFileHandles
+ * @param spillingChannelManager
+ * @param largeRecordHandler
+ * @param minNumWriteBuffers
+ * @param maxNumWriteBuffers
+ */
+ SpillingThread(
+ ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher,
+ MemoryManager memManager,
+ IOManager ioManager,
+ TypeSerializer<E> serializer,
+ TypeComparator<E> comparator,
+ List<MemorySegment> sortReadMemory,
+ List<MemorySegment> writeMemory,
+ int maxNumFileHandles,
+ SpillChannelManager spillingChannelManager,
+ LargeRecordHandler<E> largeRecordHandler,
+ SpillingBehaviour<E> spillingBehaviour,
+ int minNumWriteBuffers,
+ int maxNumWriteBuffers) {
+ super(exceptionHandler, "SortMerger spilling thread", dispatcher);
+ this.memManager = memManager;
+ this.ioManager = ioManager;
+ this.serializer = serializer;
+ this.comparator = comparator;
+ this.mergeReadMemory = sortReadMemory;
+ this.writeMemory = writeMemory;
+ this.maxFanIn = maxNumFileHandles;
+ this.spillChannelManager = spillingChannelManager;
+ this.largeRecordHandler = largeRecordHandler;
+ this.spillingBehaviour = spillingBehaviour;
+ this.minNumWriteBuffers = minNumWriteBuffers;
+ this.maxNumWriteBuffers = maxNumWriteBuffers;
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException {
+
+ final Queue<CircularElement<E>> cache = new ArrayDeque<>();
+ boolean cacheOnly = readCache(cache);
+
+ // check whether the thread was canceled
+ if (!isRunning()) {
+ return;
+ }
+
+ MutableObjectIterator<E> largeRecords = null;
+
+ // check if we can stay in memory with the large record handler
+ if (cacheOnly && largeRecordHandler != null && largeRecordHandler.hasData()) {
+ List<MemorySegment> memoryForLargeRecordSorting = new ArrayList<>();
+
+ CircularElement<E> circElement;
+ while ((circElement = this.dispatcher.poll(SortStage.READ)) != null) {
+ circElement.getBuffer().dispose();
+ memoryForLargeRecordSorting.addAll(circElement.getMemory());
+ }
+
+ if (memoryForLargeRecordSorting.isEmpty()) {
+ cacheOnly = false;
+ LOG.debug("Going to disk-based merge because of large records.");
+ } else {
+ LOG.debug("Sorting large records, to add them to in-memory merge.");
+ largeRecords = largeRecordHandler.finishWriteAndSortKeys(memoryForLargeRecordSorting);
+ }
+ }
+
+ // ------------------- In-Memory Merge ------------------------
+ if (cacheOnly) {
+ mergeInMemory(cache, largeRecords);
+ return;
+ }
+
+ // ------------------- Spilling Phase ------------------------
+ List<ChannelWithBlockCount> channelIDs = startSpilling(cache);
+
+ // ------------------- Merging Phase ------------------------
+
+ mergeOnDisk(channelIDs);
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ super.close();
+ if (spillingBehaviourOpened) {
+ this.spillingBehaviour.close();
+ this.spillingBehaviourOpened = false;
+ }
+ }
+
+ private boolean readCache(Queue<CircularElement<E>> cache) throws IOException {
+ CircularElement<E> element;
+
+ // ------------------- In-Memory Cache ------------------------
+ // fill cache
+ while (isRunning()) {
+ // take next element from queue
+ try {
+ element = this.dispatcher.take(SortStage.SPILL);
Review comment:
I though about it for a while after your comment. Even though we are slightly changing the behaviour I think it is actually fine. There are two situation where the thread can be interrupted.
1. When it is being closed via `ThreadBase#shutdown`. In this scenario both in the old and in the new code we are exiting the thread. In the new code we are throwing an exception, but it will be discarded in the `ThreadBase#internalExceptionHandler`.
2. If the thread is interrupted from outside. First of all I cannot think of exact scenario when this can happen. Nevertheless, I am pretty sure it should not happen during a successful processing and actually I doubt it is safe to continue on processing.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaoyunhaii commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492529818
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CircularQueues.java
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Collection of queues that are used for the communication between the threads.
+ */
+final class CircularQueues<E> implements StageRunner.StageMessageDispatcher<E> {
+ private final BlockingQueue<CircularElement<E>> empty;
+ private final BlockingQueue<CircularElement<E>> sort;
+ private final BlockingQueue<CircularElement<E>> spill;
+
+ private boolean isFinished = false;
Review comment:
Very sorry that there is one issue missing, but it just come to me that we might need to make the `isFinished` to be `volatile`: the `CircularQueues#close()` might be called from `SpilledThread` or `SortThread`, and we might need to read this variable in the task main thread. Since `ExternalSorter#close()` guarantees that there should be no concurrent writer, therefore where we need only `volatile` instead of atomic variables here.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f48774d7b5e0a14f1e7270591481cc2f83316854 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569)
* 756d84832df46992d406a579bd154dfc92c83e37 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348)
* 05cef2abaec734dddea631576862d7c374e320c7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561)
* 87c7f63785dda33d0a9bd210ff4d49962abbe638 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-692680928
Looks good to me, +1 to merge
Given that the previous sorter implementation had pretty extensive tests (also heavy tests to be manually executed, IIRC) it should be fine as long as we retain these tests.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689028555
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit f02440e449ea6ebefbd3eae151bf30ae0cebed72 (Fri Feb 19 07:24:24 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492172027
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/StageRunner.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * An interface for different stages of the sorting process. Different stages can communicate via
+ * the {@link StageMessageDispatcher}.
+ */
+public interface StageRunner extends AutoCloseable {
+ /**
+ * Starts the stage.
+ */
+ void start();
+
+ /**
+ * A marker interface for sending messages to different stages.
+ */
+ enum SortStage {
+ READ,
+ SORT,
+ SPILL
+ }
+
+ /**
+ * A dispatcher for inter-stage communication. It allows for returning a result to a {@link Sorter} via
+ * {@link StageMessageDispatcher#sendResult(MutableObjectIterator)}
+ */
+ interface StageMessageDispatcher<E> extends AutoCloseable {
+ /**
+ * Sends a message to the given stage.
+ */
+ void send(SortStage stage, CircularElement<E> element);
+
+ /**
+ * Retrieves and removes the head of the given queue, waiting if necessary
+ * until an element becomes available.
+ *
+ * @return the head of the queue
+ */
+ CircularElement<E> take(SortStage stage) throws InterruptedException;
Review comment:
I thought the about it. The reason why I did not do it, is because the `CircularQueues` interface is very much related to the `Queue` interface. This interface throws `InterruptedException` in the `take` method. Moreover there is no interaction with any IO. Therefore I found it a bit strange to throw an `IOException` from this method.
I am open to be convinced otherwise, as I was not unsure myself which aproach shall I take.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
}, {
"hash" : "01d8212fce667111920954a5a40e740486b80ab9",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759",
"triggerID" : "01d8212fce667111920954a5a40e740486b80ab9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0293e931b17bd770b7a982f0982ab273400cdb70 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721)
* 01d8212fce667111920954a5a40e740486b80ab9 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348)
* 05cef2abaec734dddea631576862d7c374e320c7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492107272
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SortingThread.java
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that sorts filled buffers.
+ */
+class SortingThread<E> extends ThreadBase<E> {
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SortingThread.class);
+
+ private final IndexedSorter sorter;
+
+ /**
+ * Creates a new sorting thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ public SortingThread(
+ ExceptionHandler<IOException> exceptionHandler, StageMessageDispatcher<E> dispatcher) {
Review comment:
I think the convention is that each argument on a separate line.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/StageRunner.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * An interface for different stages of the sorting process. Different stages can communicate via
+ * the {@link StageMessageDispatcher}.
+ */
+public interface StageRunner extends AutoCloseable {
+ /**
+ * Starts the stage.
+ */
+ void start();
+
+ /**
+ * A marker interface for sending messages to different stages.
+ */
+ enum SortStage {
+ READ,
+ SORT,
+ SPILL
+ }
+
+ /**
+ * A dispatcher for inter-stage communication. It allows for returning a result to a {@link Sorter} via
+ * {@link StageMessageDispatcher#sendResult(MutableObjectIterator)}
+ */
+ interface StageMessageDispatcher<E> extends AutoCloseable {
+ /**
+ * Sends a message to the given stage.
+ */
+ void send(SortStage stage, CircularElement<E> element);
+
+ /**
+ * Retrieves and removes the head of the given queue, waiting if necessary
+ * until an element becomes available.
+ *
+ * @return the head of the queue
+ */
+ CircularElement<E> take(SortStage stage) throws InterruptedException;
Review comment:
We always wrap the `InterruptedException` in an `IOException` one. Why not changing the signature to throw directly an `IOException`? This will simplify all subsequent code calls and we will have consistent behaviour across all callers.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
Review comment:
Good to update this comment about `<E>`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * An method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ /**
+ * Creates the spilling thread.
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ * @param memManager The memory manager used to allocate buffers for the readers and writers.
+ * @param ioManager The I/I manager used to instantiate readers and writers from.
+ * @param serializer
+ * @param comparator
+ * @param sortReadMemory
+ * @param writeMemory
+ * @param maxNumFileHandles
+ * @param spillingChannelManager
+ * @param largeRecordHandler
+ * @param minNumWriteBuffers
+ * @param maxNumWriteBuffers
+ */
+ SpillingThread(
+ ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher,
+ MemoryManager memManager,
+ IOManager ioManager,
+ TypeSerializer<E> serializer,
+ TypeComparator<E> comparator,
+ List<MemorySegment> sortReadMemory,
+ List<MemorySegment> writeMemory,
+ int maxNumFileHandles,
+ SpillChannelManager spillingChannelManager,
+ LargeRecordHandler<E> largeRecordHandler,
+ SpillingBehaviour<E> spillingBehaviour,
+ int minNumWriteBuffers,
+ int maxNumWriteBuffers) {
+ super(exceptionHandler, "SortMerger spilling thread", dispatcher);
+ this.memManager = memManager;
+ this.ioManager = ioManager;
+ this.serializer = serializer;
+ this.comparator = comparator;
+ this.mergeReadMemory = sortReadMemory;
+ this.writeMemory = writeMemory;
+ this.maxFanIn = maxNumFileHandles;
+ this.spillChannelManager = spillingChannelManager;
+ this.largeRecordHandler = largeRecordHandler;
+ this.spillingBehaviour = spillingBehaviour;
+ this.minNumWriteBuffers = minNumWriteBuffers;
+ this.maxNumWriteBuffers = maxNumWriteBuffers;
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException {
+
+ final Queue<CircularElement<E>> cache = new ArrayDeque<>();
+ boolean cacheOnly = readCache(cache);
+
+ // check whether the thread was canceled
+ if (!isRunning()) {
+ return;
+ }
+
+ MutableObjectIterator<E> largeRecords = null;
+
+ // check if we can stay in memory with the large record handler
+ if (cacheOnly && largeRecordHandler != null && largeRecordHandler.hasData()) {
+ List<MemorySegment> memoryForLargeRecordSorting = new ArrayList<>();
+
+ CircularElement<E> circElement;
+ while ((circElement = this.dispatcher.poll(SortStage.READ)) != null) {
+ circElement.getBuffer().dispose();
+ memoryForLargeRecordSorting.addAll(circElement.getMemory());
+ }
+
+ if (memoryForLargeRecordSorting.isEmpty()) {
+ cacheOnly = false;
+ LOG.debug("Going to disk-based merge because of large records.");
+ } else {
+ LOG.debug("Sorting large records, to add them to in-memory merge.");
+ largeRecords = largeRecordHandler.finishWriteAndSortKeys(memoryForLargeRecordSorting);
+ }
+ }
+
+ // ------------------- In-Memory Merge ------------------------
+ if (cacheOnly) {
+ mergeInMemory(cache, largeRecords);
+ return;
+ }
+
+ // ------------------- Spilling Phase ------------------------
+ List<ChannelWithBlockCount> channelIDs = startSpilling(cache);
+
+ // ------------------- Merging Phase ------------------------
+
+ mergeOnDisk(channelIDs);
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ super.close();
+ if (spillingBehaviourOpened) {
+ this.spillingBehaviour.close();
+ this.spillingBehaviourOpened = false;
+ }
+ }
+
+ private boolean readCache(Queue<CircularElement<E>> cache) throws IOException {
+ CircularElement<E> element;
+
+ // ------------------- In-Memory Cache ------------------------
+ // fill cache
+ while (isRunning()) {
+ // take next element from queue
+ try {
+ element = this.dispatcher.take(SortStage.SPILL);
Review comment:
Here we are changing a bit the behaviour of the exception handling compared to what is currently in the master. The change seems reasonable, but I want to point it out for future reference (if needed) and also to verify that this was a conscious change.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CombineValueIterator.java
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.TraversableOnceException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * This class implements an iterator over values from a sort buffer. The iterator returns the values of a given
+ * interval.
+ */
+final class CombineValueIterator<E> implements Iterator<E>, Iterable<E> {
+ private static final Logger LOG = LoggerFactory.getLogger(CombineValueIterator.class);
Review comment:
Nit: I would leave a line empty after each arguments but feel free to ignore this.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
Review comment:
"An" -> "A"
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ThreadBase.java
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import java.io.IOException;
+
+/**
+ * Base class for all working threads in this sort-merger. The specific threads for reading, sorting, spilling,
+ * merging, etc... extend this subclass.
+ * <p>
+ * The threads are designed to terminate themselves when the task they are set up to do is completed. Further more,
+ * they terminate immediately when the <code>shutdown()</code> method is called.
+ */
+abstract class ThreadBase<E> extends Thread implements Thread.UncaughtExceptionHandler, StageRunner {
+
+ /**
+ * The queue of empty buffer that can be used for reading;
+ */
+ protected final StageMessageDispatcher<E> dispatcher;
+
+ /**
+ * The exception handler for any problems.
+ */
+ private final ExceptionHandler<IOException> exceptionHandler;
+
+ /**
+ * The flag marking this thread as alive.
+ */
+ private volatile boolean alive;
+
+ /**
+ * Creates a new thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param name The name of the thread.
+ * @param queues The queues used to pass buffers between the threads.
+ */
+ protected ThreadBase(
+ ExceptionHandler<IOException> exceptionHandler,
+ String name,
+ StageMessageDispatcher<E> queues) {
+ // thread setup
+ super(name);
Review comment:
Why not adding `checkNotNull()` or `@Nullable` wherever needed? This will make clearer the contract for each arg.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * An method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ /**
+ * Creates the spilling thread.
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ * @param memManager The memory manager used to allocate buffers for the readers and writers.
+ * @param ioManager The I/I manager used to instantiate readers and writers from.
+ * @param serializer
+ * @param comparator
+ * @param sortReadMemory
+ * @param writeMemory
+ * @param maxNumFileHandles
+ * @param spillingChannelManager
+ * @param largeRecordHandler
+ * @param minNumWriteBuffers
+ * @param maxNumWriteBuffers
+ */
+ SpillingThread(
Review comment:
Adding some `checkNotNull()` checks?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillChannelManager.java
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+
+import java.io.File;
+import java.util.HashSet;
+import java.util.Iterator;
+
+/**
+ * Channel manager to manage the life cycle of spill channels.
+ */
+final class SpillChannelManager implements AutoCloseable {
+
+ /**
+ * Collection of all currently open channels, to be closed and deleted during cleanup.
+ */
+ private final HashSet<FileIOChannel> openChannels;
+
+ /**
+ * Collection of all temporary files created and to be removed when closing the sorter.
+ */
+ private final HashSet<FileIOChannel.ID> channelsToDeleteAtShutdown;
+
+ private volatile boolean closed;
+
+ public SpillChannelManager() {
+ this.channelsToDeleteAtShutdown = new HashSet<>(64);
+ this.openChannels = new HashSet<>(64);
+ }
+
+ /**
+ * Adds a channel to the list of channels that are to be removed at shutdown.
+ *
+ * @param channel The channel id.
+ */
+ synchronized void registerChannelToBeRemovedAtShudown(FileIOChannel.ID channel) {
Review comment:
Typo: All the method names miss a `t` in the `Shu**t**down`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-690123429
It is used exclusively in the DataSet API so far. SQL runtime (blink) has its own version that originates from this classes but diverged substantially. We can think of unifying it in the future but so far I did not want to change the sql runtime.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684)
* 097e75ef7eaa295e31e8acb6fca83a2c4a254b41 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ec668b201cb4860c9f57a8c0db312591b57ab1c4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644)
* 31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ec668b201cb4860c9f57a8c0db312591b57ab1c4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644)
* 31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0293e931b17bd770b7a982f0982ab273400cdb70 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 87c7f63785dda33d0a9bd210ff4d49962abbe638 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564)
* f48774d7b5e0a14f1e7270591481cc2f83316854 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
}, {
"hash" : "01d8212fce667111920954a5a40e740486b80ab9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759",
"triggerID" : "01d8212fce667111920954a5a40e740486b80ab9",
"triggerType" : "PUSH"
}, {
"hash" : "f02440e449ea6ebefbd3eae151bf30ae0cebed72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f02440e449ea6ebefbd3eae151bf30ae0cebed72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 01d8212fce667111920954a5a40e740486b80ab9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759)
* f02440e449ea6ebefbd3eae151bf30ae0cebed72 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r493254830
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/StageRunner.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * An interface for different stages of the sorting process. Different stages can communicate via
+ * the {@link StageMessageDispatcher}.
+ */
+public interface StageRunner extends AutoCloseable {
+ /**
+ * Starts the stage.
+ */
+ void start();
+
+ /**
+ * A marker interface for sending messages to different stages.
+ */
+ enum SortStage {
+ READ,
+ SORT,
+ SPILL
+ }
+
+ /**
+ * A dispatcher for inter-stage communication. It allows for returning a result to a {@link Sorter} via
+ * {@link StageMessageDispatcher#sendResult(MutableObjectIterator)}
+ */
+ interface StageMessageDispatcher<E> extends AutoCloseable {
+ /**
+ * Sends a message to the given stage.
+ */
+ void send(SortStage stage, CircularElement<E> element);
+
+ /**
+ * Retrieves and removes the head of the given queue, waiting if necessary
+ * until an element becomes available.
+ *
+ * @return the head of the queue
+ */
+ CircularElement<E> take(SortStage stage) throws InterruptedException;
Review comment:
Ok I saw that you changed the `go()` to throw the `InterruptedException` instead of wrapping it in an IO one. I am having another look now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-697407842
Thank you @StephanEwen @gaoyunhaii @kl0u for reviewing. I will merge it now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-696625987
Could you have another look @kl0u @gaoyunhaii ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r490907492
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SorterInputGateway.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.runtime.operators.sort.StageRunner.SortStage;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * A gateway for writing records into the sort/merge process.
+ */
+final class SorterInputGateway<E> {
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SorterInputGateway.class);
+ private final LargeRecordHandler<E> largeRecords;
+ /** The object into which the thread reads the data from the input. */
+ private final StageRunner.StageMessageDispatcher<E> dispatcher;
+ private long bytesUntilSpilling;
+ private CircularElement<E> currentBuffer;
+
+ /**
+ * Creates a new gateway for pushing records into the sorter.
+ *
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ SorterInputGateway(
+ StageRunner.StageMessageDispatcher<E> dispatcher,
+ LargeRecordHandler<E> largeRecordsHandler,
+ long startSpillingBytes) {
+
+ // members
+ this.bytesUntilSpilling = startSpillingBytes;
+ this.largeRecords = largeRecordsHandler;
+ this.dispatcher = dispatcher;
+
+ if (bytesUntilSpilling < 1) {
+ this.dispatcher.send(SortStage.SORT, CircularElement.spillingMarker());
+ }
+ }
+
+ /**
+ * Writes the given record for sorting.
+ */
+ public void writeRecord(E record) throws IOException {
+
+ if (currentBuffer == null) {
+ this.currentBuffer = this.dispatcher.take(SortStage.READ);
Review comment:
Good catch! Now that it is not run in a separate thread it has to be done accounted for.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaoyunhaii commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-697219809
Very thanks @dawidwys for the PR! I re-checked the PR and now there should be no issues from my side.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
}, {
"hash" : "01d8212fce667111920954a5a40e740486b80ab9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759",
"triggerID" : "01d8212fce667111920954a5a40e740486b80ab9",
"triggerType" : "PUSH"
}, {
"hash" : "f02440e449ea6ebefbd3eae151bf30ae0cebed72",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6836",
"triggerID" : "f02440e449ea6ebefbd3eae151bf30ae0cebed72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 01d8212fce667111920954a5a40e740486b80ab9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759)
* f02440e449ea6ebefbd3eae151bf30ae0cebed72 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6836)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684)
* 097e75ef7eaa295e31e8acb6fca83a2c4a254b41 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348)
* 05cef2abaec734dddea631576862d7c374e320c7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561)
* 87c7f63785dda33d0a9bd210ff4d49962abbe638 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564)
* f48774d7b5e0a14f1e7270591481cc2f83316854 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
}, {
"hash" : "01d8212fce667111920954a5a40e740486b80ab9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759",
"triggerID" : "01d8212fce667111920954a5a40e740486b80ab9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 01d8212fce667111920954a5a40e740486b80ab9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaoyunhaii commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492242431
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CircularQueues.java
##########
@@ -72,16 +76,23 @@ public void sendResult(MutableObjectIterator<E> result) {
}
@Override
- public CircularElement<E> take(StageRunner.SortStage stage) {
- try {
- return getQueue(stage).take();
- } catch (InterruptedException e) {
- throw new WrappingRuntimeException(e);
+ public CircularElement<E> take(StageRunner.SortStage stage) throws InterruptedException {
+ while (!isFinished) {
+ CircularElement<E> value = getQueue(stage).poll(1, TimeUnit.SECONDS);
+ if (value != null) {
+ return value;
+ }
}
+ throw new FlinkRuntimeException("The sorter is closed already");
}
@Override
public CircularElement<E> poll(StageRunner.SortStage stage) {
return getQueue(stage).poll();
}
+
+ @Override
+ public void close() throws Exception {
Review comment:
Sorry but I did not find where we call the `close` method?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -586,4 +594,8 @@ private void getSegmentsForReaders(
}
}
}
+
+ private IOException wrapWithIOException(InterruptedException e) throws IOException {
+ return new IOException("The sortes has been interrupted", e);
Review comment:
Typo `Sortes` -> `Sorter` ?
I'm a little wondering that is it also possible to direct throws `InterruptedException` (We might add it to some methods' declaration) ?
If there are blockers for throwing `InterruptedException` directly I think we might promote the method to the base class since the other threads also faced with the situation to wrapper.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 05cef2abaec734dddea631576862d7c374e320c7 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561)
* 87c7f63785dda33d0a9bd210ff4d49962abbe638 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564)
* f48774d7b5e0a14f1e7270591481cc2f83316854 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r490825217
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/StageRunner.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * An interface for different stages of the sorting process. Different stages can communicate via
+ * the {@link StageMessageDispatcher}.
+ */
+public interface StageRunner extends AutoCloseable {
+ /**
+ * Starts the stage.
+ */
+ void start();
+
+ /**
+ * A marker interface for sending messages to different stages.
+ */
+ enum SortStage {
+ READ,
+ SPILL,
Review comment:
good point!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
}, {
"hash" : "01d8212fce667111920954a5a40e740486b80ab9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6759",
"triggerID" : "01d8212fce667111920954a5a40e740486b80ab9",
"triggerType" : "PUSH"
}, {
"hash" : "f02440e449ea6ebefbd3eae151bf30ae0cebed72",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6836",
"triggerID" : "f02440e449ea6ebefbd3eae151bf30ae0cebed72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f02440e449ea6ebefbd3eae151bf30ae0cebed72 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6836)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348)
* 05cef2abaec734dddea631576862d7c374e320c7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561)
* 87c7f63785dda33d0a9bd210ff4d49962abbe638 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564)
* f48774d7b5e0a14f1e7270591481cc2f83316854 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348)
* 05cef2abaec734dddea631576862d7c374e320c7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561)
* 87c7f63785dda33d0a9bd210ff4d49962abbe638 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r493319802
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,575 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * A method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * A method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
Review comment:
Good call! I missed that.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492522939
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CircularQueues.java
##########
@@ -72,16 +76,23 @@ public void sendResult(MutableObjectIterator<E> result) {
}
@Override
- public CircularElement<E> take(StageRunner.SortStage stage) {
- try {
- return getQueue(stage).take();
- } catch (InterruptedException e) {
- throw new WrappingRuntimeException(e);
+ public CircularElement<E> take(StageRunner.SortStage stage) throws InterruptedException {
+ while (!isFinished) {
+ CircularElement<E> value = getQueue(stage).poll(1, TimeUnit.SECONDS);
+ if (value != null) {
+ return value;
+ }
}
+ throw new FlinkRuntimeException("The sorter is closed already");
}
@Override
public CircularElement<E> poll(StageRunner.SortStage stage) {
return getQueue(stage).poll();
}
+
+ @Override
+ public void close() throws Exception {
Review comment:
I am so sorry about that. I completely forgot about calling this method in the `ExternalSorter#close`. I added it there.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaoyunhaii commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r490734842
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/StageRunner.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * An interface for different stages of the sorting process. Different stages can communicate via
+ * the {@link StageMessageDispatcher}.
+ */
+public interface StageRunner extends AutoCloseable {
+ /**
+ * Starts the stage.
+ */
+ void start();
+
+ /**
+ * A marker interface for sending messages to different stages.
+ */
+ enum SortStage {
+ READ,
+ SPILL,
Review comment:
I think it might be better to exchange the order of `SORT` and `SPILL` since sorting happens before spilling.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CircularQueues.java
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.WrappingRuntimeException;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Collection of queues that are used for the communication between the threads.
+ */
+final class CircularQueues<E> implements StageRunner.StageMessageDispatcher<E> {
+ final BlockingQueue<CircularElement<E>> empty;
+ final BlockingQueue<CircularElement<E>> sort;
+ final BlockingQueue<CircularElement<E>> spill;
+ /**
+ * The iterator to be returned by the sort-merger. This variable is null, while receiving and merging is still in
+ * progress and it will be set once we have < merge factor sorted sub-streams that will then be streamed sorted.
+ */
+ private final CompletableFuture<MutableObjectIterator<E>> iteratorFuture = new CompletableFuture<>();
+
+ public CircularQueues() {
+ this.empty = new LinkedBlockingQueue<>();
+ this.sort = new LinkedBlockingQueue<>();
+ this.spill = new LinkedBlockingQueue<>();
+ }
+
+ private BlockingQueue<CircularElement<E>> getQueue(StageRunner.SortStage stage) {
+ switch (stage) {
+ case READ:
+ return empty;
+ case SPILL:
Review comment:
Similar to the one comment in `SortStage` I would prefer to move the `SORT` case before the `SPILL` case.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SorterInputGateway.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.runtime.operators.sort.StageRunner.SortStage;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * A gateway for writing records into the sort/merge process.
+ */
+final class SorterInputGateway<E> {
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SorterInputGateway.class);
+ private final LargeRecordHandler<E> largeRecords;
+ /** The object into which the thread reads the data from the input. */
+ private final StageRunner.StageMessageDispatcher<E> dispatcher;
+ private long bytesUntilSpilling;
+ private CircularElement<E> currentBuffer;
+
+ /**
+ * Creates a new gateway for pushing records into the sorter.
+ *
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ SorterInputGateway(
+ StageRunner.StageMessageDispatcher<E> dispatcher,
+ LargeRecordHandler<E> largeRecordsHandler,
+ long startSpillingBytes) {
+
+ // members
+ this.bytesUntilSpilling = startSpillingBytes;
+ this.largeRecords = largeRecordsHandler;
+ this.dispatcher = dispatcher;
+
+ if (bytesUntilSpilling < 1) {
+ this.dispatcher.send(SortStage.SORT, CircularElement.spillingMarker());
+ }
+ }
+
+ /**
+ * Writes the given record for sorting.
+ */
+ public void writeRecord(E record) throws IOException {
+
+ if (currentBuffer == null) {
+ this.currentBuffer = this.dispatcher.take(SortStage.READ);
Review comment:
One concern about infinite waiting for next buffer would be that if the spilling thread throws exception, the current thread (I think it should be the task main thread) would not get notified and would be keep waiting here. In this case the task would not be able to failover.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ExternalSorterBuilder.java
##########
@@ -0,0 +1,419 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.functions.GroupCombineFunction;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A builder for an {@link ExternalSorter}. It can construct either a pull-based sorter if provided with an input
+ * iterator via {@link #build(MutableObjectIterator)} or a push-based one via {@link #build()}.
+ */
+public final class ExternalSorterBuilder<T> {
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(ExternalSorter.class);
+
+ /** Fix length records with a length below this threshold will be in-place sorted, if possible. */
+ private static final int THRESHOLD_FOR_IN_PLACE_SORTING = 32;
+
+ /** The minimal number of buffers to use by the writers. */
+ private static final int MIN_NUM_WRITE_BUFFERS = 2;
+
+ /** The maximal number of buffers to use by the writers. */
+ private static final int MAX_NUM_WRITE_BUFFERS = 4;
+
+ /** The minimum number of segments that are required for the sort to operate. */
+ private static final int MIN_NUM_SORT_MEM_SEGMENTS = 10;
+
+ private final MemoryManager memoryManager;
+ private final AbstractInvokable parentTask;
+ private final TypeSerializer<T> serializer;
+ private final TypeComparator<T> comparator;
+ private InMemorySorterFactory<T> inMemorySorterFactory;
+ private int maxNumFileHandles = AlgorithmOptions.SPILLING_MAX_FAN.defaultValue();
+ private boolean objectReuseEnabled = false;
+ private boolean handleLargeRecords = ConfigConstants.DEFAULT_USE_LARGE_RECORD_HANDLER;
+ private double memoryFraction = 1.0;
+ private int numSortBuffers = -1;
+ private double startSpillingFraction = AlgorithmOptions.SORT_SPILLING_THRESHOLD.defaultValue();
+ private IOManager ioManager;
+ private boolean noSpillingMemory = true;
+ private GroupCombineFunction<T, T> combineFunction;
+ private Configuration udfConfig;
+ private List<MemorySegment> memorySegments = null;
+
+ ExternalSorterBuilder(
+ MemoryManager memoryManager,
+ AbstractInvokable parentTask,
+ TypeSerializer<T> serializer,
+ TypeComparator<T> comparator) {
+ this.memoryManager = memoryManager;
+ this.parentTask = parentTask;
+ this.serializer = serializer;
+ this.comparator = comparator;
+ this.inMemorySorterFactory = new DefaultInMemorySorterFactory<>(
+ serializer,
+ comparator,
+ THRESHOLD_FOR_IN_PLACE_SORTING);
+ }
+
+ public ExternalSorterBuilder<T> maxNumFileHandles(int maxNumFileHandles) {
+ if (maxNumFileHandles < 2) {
+ throw new IllegalArgumentException("Merger cannot work with less than two file handles.");
+ }
+ this.maxNumFileHandles = maxNumFileHandles;
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> objectReuse(boolean enabled) {
+ this.objectReuseEnabled = enabled;
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> largeRecords(boolean enabled) {
+ this.handleLargeRecords = enabled;
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> enableSpilling(IOManager ioManager) {
+ this.noSpillingMemory = false;
+ this.ioManager = checkNotNull(ioManager);
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> enableSpilling(IOManager ioManager, double startSpillingFraction) {
+ this.startSpillingFraction = startSpillingFraction;
+ return enableSpilling(ioManager);
+ }
+
+ public ExternalSorterBuilder<T> memoryFraction(double fraction) {
+ this.memoryFraction = fraction;
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> memory(List<MemorySegment> memorySegments) {
+ this.memorySegments = checkNotNull(memorySegments);
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> sortBuffers(int numSortBuffers) {
+ this.numSortBuffers = numSortBuffers;
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> withCombiner(GroupCombineFunction<T, T> combineFunction, Configuration udfConfig) {
+ this.combineFunction = checkNotNull(combineFunction);
+ this.udfConfig = checkNotNull(udfConfig);
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> withCombiner(GroupCombineFunction<T, T> combineFunction) {
+ this.combineFunction = checkNotNull(combineFunction);
+ this.udfConfig = new Configuration();
+ return this;
+ }
+
+ public ExternalSorterBuilder<T> sorterFactory(InMemorySorterFactory<T> sorterFactory) {
+ this.inMemorySorterFactory = checkNotNull(sorterFactory);
+ return this;
+ }
+
+ /**
+ * Creates a pull-based {@link Sorter}. The {@link Sorter#getIterator()} will return when all the records from
+ * the given input are consumed. Will spawn three threads: read, sort, spill.
+ */
+ public ExternalSorter<T> build(MutableObjectIterator<T> input) throws MemoryAllocationException {
+ return doBuild((exceptionHandler, dispatcher, largeRecordHandler, startSpillingBytes) ->
+ new ReadingThread<>(
+ exceptionHandler,
+ input,
+ dispatcher,
+ largeRecordHandler,
+ serializer.createInstance(),
+ startSpillingBytes
+ )
+ );
+ }
+
+ /**
+ * Creates a push-based {@link PushSorter}. The {@link PushSorter#getIterator()} will return when the
+ * {@link PushSorter#finishReading()} is called. Will spawn two threads: sort, spill.
+ */
+ public PushSorter<T> build() throws MemoryAllocationException {
+ PushFactory<T> pushFactory = new PushFactory<>();
+ ExternalSorter<T> tExternalSorter = doBuild(pushFactory);
+
+ return new PushSorter<T>() {
+ private final SorterInputGateway<T> recordProducer = pushFactory.sorterInputGateway;
+
+ @Override
+ public void writeRecord(T record) throws IOException {
+ recordProducer.writeRecord(record);
+ }
+
+ @Override
+ public void finishReading() {
+ recordProducer.finishReading();
+ }
+
+ @Override
+ public MutableObjectIterator<T> getIterator() throws InterruptedException {
+ return tExternalSorter.getIterator();
+ }
+
+ @Override
+ public void close() throws IOException {
+ tExternalSorter.close();
+ }
+ };
+ }
+
+ @FunctionalInterface
+ private interface ReadingStageFactory<E> {
+ @Nullable StageRunner createReadingStage(
+ ExceptionHandler<IOException> exceptionHandler,
+ StageRunner.StageMessageDispatcher<E> dispatcher,
+ LargeRecordHandler<E> largeRecordHandler,
+ long startSpillingBytes);
+ }
+
+ private static final class PushFactory<E> implements ReadingStageFactory<E> {
+ private SorterInputGateway<E> sorterInputGateway;
+
+ @Override
+ public StageRunner createReadingStage(
+ ExceptionHandler<IOException> exceptionHandler,
+ StageRunner.StageMessageDispatcher<E> dispatcher,
+ LargeRecordHandler<E> largeRecordHandler,
+ long startSpillingBytes) {
+ sorterInputGateway = new SorterInputGateway<>(
+ dispatcher,
+ largeRecordHandler,
+ startSpillingBytes
+ );
+ return null;
+ }
+ }
+
+ private ExternalSorter<T> doBuild(ReadingStageFactory<T> readingStageFactory) throws MemoryAllocationException {
+
+ final List<MemorySegment> memory;
+ if (this.memorySegments != null) {
+ memory = this.memorySegments;
+ } else {
+ memory = memoryManager.allocatePages(
+ parentTask,
+ memoryManager.computeNumberOfPages(memoryFraction));
+ }
+
+ // adjust the memory quotas to the page size
+ final int numPagesTotal = memory.size();
+
+ if (numPagesTotal < MIN_NUM_WRITE_BUFFERS + MIN_NUM_SORT_MEM_SEGMENTS) {
+ throw new IllegalArgumentException("Too little memory provided to sorter to perform task. " +
+ "Required are at least " + (MIN_NUM_WRITE_BUFFERS + MIN_NUM_SORT_MEM_SEGMENTS) +
+ " pages. Current page size is " + memoryManager.getPageSize() + " bytes.");
+ }
+
+ // determine how many buffers to use for writing
+ final int numWriteBuffers;
+ final int numLargeRecordBuffers;
+
+ if (noSpillingMemory && !handleLargeRecords) {
+ numWriteBuffers = 0;
+ numLargeRecordBuffers = 0;
+ } else {
+ int numConsumers = (noSpillingMemory ? 0 : 1) + (handleLargeRecords ? 2 : 0);
+
+ // determine how many buffers we have when we do a full mere with maximal fan-in
Review comment:
Typo `mere` -> `merge`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348)
* 05cef2abaec734dddea631576862d7c374e320c7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ec668b201cb4860c9f57a8c0db312591b57ab1c4 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684)
* 097e75ef7eaa295e31e8acb6fca83a2c4a254b41 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716)
* 0293e931b17bd770b7a982f0982ab273400cdb70 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-695957281
Thanks for the review @gaoyunhaii, I addressed your comments. Could you have another look?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6644",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
}, {
"hash" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6684",
"triggerID" : "31493cdd0e23d32a3cd1a5c1dd065a5cab87fa86",
"triggerType" : "PUSH"
}, {
"hash" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6716",
"triggerID" : "097e75ef7eaa295e31e8acb6fca83a2c4a254b41",
"triggerType" : "PUSH"
}, {
"hash" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721",
"triggerID" : "0293e931b17bd770b7a982f0982ab273400cdb70",
"triggerType" : "PUSH"
}, {
"hash" : "01d8212fce667111920954a5a40e740486b80ab9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "01d8212fce667111920954a5a40e740486b80ab9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0293e931b17bd770b7a982f0982ab273400cdb70 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6721)
* 01d8212fce667111920954a5a40e740486b80ab9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-696625987
Could you have another look @kl0u @gaoyunhaii ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 54ee6987c7054a42fa52d04ec61fae74a816792c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
}, {
"hash" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ec668b201cb4860c9f57a8c0db312591b57ab1c4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 756d84832df46992d406a579bd154dfc92c83e37 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542)
* ec668b201cb4860c9f57a8c0db312591b57ab1c4 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] gaoyunhaii commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
gaoyunhaii commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492242431
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CircularQueues.java
##########
@@ -72,16 +76,23 @@ public void sendResult(MutableObjectIterator<E> result) {
}
@Override
- public CircularElement<E> take(StageRunner.SortStage stage) {
- try {
- return getQueue(stage).take();
- } catch (InterruptedException e) {
- throw new WrappingRuntimeException(e);
+ public CircularElement<E> take(StageRunner.SortStage stage) throws InterruptedException {
+ while (!isFinished) {
+ CircularElement<E> value = getQueue(stage).poll(1, TimeUnit.SECONDS);
+ if (value != null) {
+ return value;
+ }
}
+ throw new FlinkRuntimeException("The sorter is closed already");
}
@Override
public CircularElement<E> poll(StageRunner.SortStage stage) {
return getQueue(stage).poll();
}
+
+ @Override
+ public void close() throws Exception {
Review comment:
Sorry but I did not find where we call the `close` method?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -586,4 +594,8 @@ private void getSegmentsForReaders(
}
}
}
+
+ private IOException wrapWithIOException(InterruptedException e) throws IOException {
+ return new IOException("The sortes has been interrupted", e);
Review comment:
Typo `Sortes` -> `Sorter` ?
I'm a little wondering that is it also possible to direct throws `InterruptedException` (We might add it to some methods' declaration) ?
If there are blockers for throwing `InterruptedException` directly I think we might promote the method to the base class since the other threads also faced with the situation to wrapper.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492107272
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SortingThread.java
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that sorts filled buffers.
+ */
+class SortingThread<E> extends ThreadBase<E> {
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SortingThread.class);
+
+ private final IndexedSorter sorter;
+
+ /**
+ * Creates a new sorting thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ */
+ public SortingThread(
+ ExceptionHandler<IOException> exceptionHandler, StageMessageDispatcher<E> dispatcher) {
Review comment:
I think the convention is that each argument on a separate line.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/StageRunner.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * An interface for different stages of the sorting process. Different stages can communicate via
+ * the {@link StageMessageDispatcher}.
+ */
+public interface StageRunner extends AutoCloseable {
+ /**
+ * Starts the stage.
+ */
+ void start();
+
+ /**
+ * A marker interface for sending messages to different stages.
+ */
+ enum SortStage {
+ READ,
+ SORT,
+ SPILL
+ }
+
+ /**
+ * A dispatcher for inter-stage communication. It allows for returning a result to a {@link Sorter} via
+ * {@link StageMessageDispatcher#sendResult(MutableObjectIterator)}
+ */
+ interface StageMessageDispatcher<E> extends AutoCloseable {
+ /**
+ * Sends a message to the given stage.
+ */
+ void send(SortStage stage, CircularElement<E> element);
+
+ /**
+ * Retrieves and removes the head of the given queue, waiting if necessary
+ * until an element becomes available.
+ *
+ * @return the head of the queue
+ */
+ CircularElement<E> take(SortStage stage) throws InterruptedException;
Review comment:
We always wrap the `InterruptedException` in an `IOException` one. Why not changing the signature to throw directly an `IOException`? This will simplify all subsequent code calls and we will have consistent behaviour across all callers.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
Review comment:
Good to update this comment about `<E>`.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * An method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ /**
+ * Creates the spilling thread.
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ * @param memManager The memory manager used to allocate buffers for the readers and writers.
+ * @param ioManager The I/I manager used to instantiate readers and writers from.
+ * @param serializer
+ * @param comparator
+ * @param sortReadMemory
+ * @param writeMemory
+ * @param maxNumFileHandles
+ * @param spillingChannelManager
+ * @param largeRecordHandler
+ * @param minNumWriteBuffers
+ * @param maxNumWriteBuffers
+ */
+ SpillingThread(
+ ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher,
+ MemoryManager memManager,
+ IOManager ioManager,
+ TypeSerializer<E> serializer,
+ TypeComparator<E> comparator,
+ List<MemorySegment> sortReadMemory,
+ List<MemorySegment> writeMemory,
+ int maxNumFileHandles,
+ SpillChannelManager spillingChannelManager,
+ LargeRecordHandler<E> largeRecordHandler,
+ SpillingBehaviour<E> spillingBehaviour,
+ int minNumWriteBuffers,
+ int maxNumWriteBuffers) {
+ super(exceptionHandler, "SortMerger spilling thread", dispatcher);
+ this.memManager = memManager;
+ this.ioManager = ioManager;
+ this.serializer = serializer;
+ this.comparator = comparator;
+ this.mergeReadMemory = sortReadMemory;
+ this.writeMemory = writeMemory;
+ this.maxFanIn = maxNumFileHandles;
+ this.spillChannelManager = spillingChannelManager;
+ this.largeRecordHandler = largeRecordHandler;
+ this.spillingBehaviour = spillingBehaviour;
+ this.minNumWriteBuffers = minNumWriteBuffers;
+ this.maxNumWriteBuffers = maxNumWriteBuffers;
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException {
+
+ final Queue<CircularElement<E>> cache = new ArrayDeque<>();
+ boolean cacheOnly = readCache(cache);
+
+ // check whether the thread was canceled
+ if (!isRunning()) {
+ return;
+ }
+
+ MutableObjectIterator<E> largeRecords = null;
+
+ // check if we can stay in memory with the large record handler
+ if (cacheOnly && largeRecordHandler != null && largeRecordHandler.hasData()) {
+ List<MemorySegment> memoryForLargeRecordSorting = new ArrayList<>();
+
+ CircularElement<E> circElement;
+ while ((circElement = this.dispatcher.poll(SortStage.READ)) != null) {
+ circElement.getBuffer().dispose();
+ memoryForLargeRecordSorting.addAll(circElement.getMemory());
+ }
+
+ if (memoryForLargeRecordSorting.isEmpty()) {
+ cacheOnly = false;
+ LOG.debug("Going to disk-based merge because of large records.");
+ } else {
+ LOG.debug("Sorting large records, to add them to in-memory merge.");
+ largeRecords = largeRecordHandler.finishWriteAndSortKeys(memoryForLargeRecordSorting);
+ }
+ }
+
+ // ------------------- In-Memory Merge ------------------------
+ if (cacheOnly) {
+ mergeInMemory(cache, largeRecords);
+ return;
+ }
+
+ // ------------------- Spilling Phase ------------------------
+ List<ChannelWithBlockCount> channelIDs = startSpilling(cache);
+
+ // ------------------- Merging Phase ------------------------
+
+ mergeOnDisk(channelIDs);
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ super.close();
+ if (spillingBehaviourOpened) {
+ this.spillingBehaviour.close();
+ this.spillingBehaviourOpened = false;
+ }
+ }
+
+ private boolean readCache(Queue<CircularElement<E>> cache) throws IOException {
+ CircularElement<E> element;
+
+ // ------------------- In-Memory Cache ------------------------
+ // fill cache
+ while (isRunning()) {
+ // take next element from queue
+ try {
+ element = this.dispatcher.take(SortStage.SPILL);
Review comment:
Here we are changing a bit the behaviour of the exception handling compared to what is currently in the master. The change seems reasonable, but I want to point it out for future reference (if needed) and also to verify that this was a conscious change.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CombineValueIterator.java
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.TraversableOnceException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * This class implements an iterator over values from a sort buffer. The iterator returns the values of a given
+ * interval.
+ */
+final class CombineValueIterator<E> implements Iterator<E>, Iterable<E> {
+ private static final Logger LOG = LoggerFactory.getLogger(CombineValueIterator.class);
Review comment:
Nit: I would leave a line empty after each arguments but feel free to ignore this.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
Review comment:
"An" -> "A"
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ThreadBase.java
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import java.io.IOException;
+
+/**
+ * Base class for all working threads in this sort-merger. The specific threads for reading, sorting, spilling,
+ * merging, etc... extend this subclass.
+ * <p>
+ * The threads are designed to terminate themselves when the task they are set up to do is completed. Further more,
+ * they terminate immediately when the <code>shutdown()</code> method is called.
+ */
+abstract class ThreadBase<E> extends Thread implements Thread.UncaughtExceptionHandler, StageRunner {
+
+ /**
+ * The queue of empty buffer that can be used for reading;
+ */
+ protected final StageMessageDispatcher<E> dispatcher;
+
+ /**
+ * The exception handler for any problems.
+ */
+ private final ExceptionHandler<IOException> exceptionHandler;
+
+ /**
+ * The flag marking this thread as alive.
+ */
+ private volatile boolean alive;
+
+ /**
+ * Creates a new thread.
+ *
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param name The name of the thread.
+ * @param queues The queues used to pass buffers between the threads.
+ */
+ protected ThreadBase(
+ ExceptionHandler<IOException> exceptionHandler,
+ String name,
+ StageMessageDispatcher<E> queues) {
+ // thread setup
+ super(name);
Review comment:
Why not adding `checkNotNull()` or `@Nullable` wherever needed? This will make clearer the contract for each arg.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * An method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ /**
+ * Creates the spilling thread.
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ * @param memManager The memory manager used to allocate buffers for the readers and writers.
+ * @param ioManager The I/I manager used to instantiate readers and writers from.
+ * @param serializer
+ * @param comparator
+ * @param sortReadMemory
+ * @param writeMemory
+ * @param maxNumFileHandles
+ * @param spillingChannelManager
+ * @param largeRecordHandler
+ * @param minNumWriteBuffers
+ * @param maxNumWriteBuffers
+ */
+ SpillingThread(
Review comment:
Adding some `checkNotNull()` checks?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillChannelManager.java
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+
+import java.io.File;
+import java.util.HashSet;
+import java.util.Iterator;
+
+/**
+ * Channel manager to manage the life cycle of spill channels.
+ */
+final class SpillChannelManager implements AutoCloseable {
+
+ /**
+ * Collection of all currently open channels, to be closed and deleted during cleanup.
+ */
+ private final HashSet<FileIOChannel> openChannels;
+
+ /**
+ * Collection of all temporary files created and to be removed when closing the sorter.
+ */
+ private final HashSet<FileIOChannel.ID> channelsToDeleteAtShutdown;
+
+ private volatile boolean closed;
+
+ public SpillChannelManager() {
+ this.channelsToDeleteAtShutdown = new HashSet<>(64);
+ this.openChannels = new HashSet<>(64);
+ }
+
+ /**
+ * Adds a channel to the list of channels that are to be removed at shutdown.
+ *
+ * @param channel The channel id.
+ */
+ synchronized void registerChannelToBeRemovedAtShudown(FileIOChannel.ID channel) {
Review comment:
Typo: All the method names miss a `t` in the `Shu**t**down`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 87c7f63785dda33d0a9bd210ff4d49962abbe638 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564)
* f48774d7b5e0a14f1e7270591481cc2f83316854 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569)
* 756d84832df46992d406a579bd154dfc92c83e37 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-689034696
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6348",
"triggerID" : "54ee6987c7054a42fa52d04ec61fae74a816792c",
"triggerType" : "PUSH"
}, {
"hash" : "05cef2abaec734dddea631576862d7c374e320c7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6561",
"triggerID" : "05cef2abaec734dddea631576862d7c374e320c7",
"triggerType" : "PUSH"
}, {
"hash" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6564",
"triggerID" : "87c7f63785dda33d0a9bd210ff4d49962abbe638",
"triggerType" : "PUSH"
}, {
"hash" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6569",
"triggerID" : "f48774d7b5e0a14f1e7270591481cc2f83316854",
"triggerType" : "PUSH"
}, {
"hash" : "756d84832df46992d406a579bd154dfc92c83e37",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542",
"triggerID" : "756d84832df46992d406a579bd154dfc92c83e37",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 756d84832df46992d406a579bd154dfc92c83e37 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6542)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13357:
URL: https://github.com/apache/flink/pull/13357#discussion_r492172027
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/StageRunner.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.util.MutableObjectIterator;
+
+/**
+ * An interface for different stages of the sorting process. Different stages can communicate via
+ * the {@link StageMessageDispatcher}.
+ */
+public interface StageRunner extends AutoCloseable {
+ /**
+ * Starts the stage.
+ */
+ void start();
+
+ /**
+ * A marker interface for sending messages to different stages.
+ */
+ enum SortStage {
+ READ,
+ SORT,
+ SPILL
+ }
+
+ /**
+ * A dispatcher for inter-stage communication. It allows for returning a result to a {@link Sorter} via
+ * {@link StageMessageDispatcher#sendResult(MutableObjectIterator)}
+ */
+ interface StageMessageDispatcher<E> extends AutoCloseable {
+ /**
+ * Sends a message to the given stage.
+ */
+ void send(SortStage stage, CircularElement<E> element);
+
+ /**
+ * Retrieves and removes the head of the given queue, waiting if necessary
+ * until an element becomes available.
+ *
+ * @return the head of the queue
+ */
+ CircularElement<E> take(SortStage stage) throws InterruptedException;
Review comment:
I thought the about it. The reason why I did not do it, is because the `CircularQueues` interface is very much related to the `Queue` interface. This interface throws `InterruptedException` in the `take` method. Moreover there is no interaction with any IO. Therefore I found it a bit strange to throw an `IOException` from this method.
I am open to be convinced otherwise, as I was not unsure myself which aproach shall I take.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
##########
@@ -0,0 +1,601 @@
+/*
+ * 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.flink.runtime.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelReader;
+import org.apache.flink.runtime.io.disk.iomanager.BlockChannelWriter;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelReaderInputView;
+import org.apache.flink.runtime.io.disk.iomanager.ChannelWriterOutputView;
+import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Queue;
+
+import static org.apache.flink.runtime.operators.sort.CircularElement.EOF_MARKER;
+import static org.apache.flink.runtime.operators.sort.CircularElement.SPILLING_MARKER;
+
+/**
+ * The thread that handles the spilling of intermediate results and sets up the merging. It also merges the
+ * channels until sufficiently few channels remain to perform the final streamed merge.
+ */
+final class SpillingThread<E> extends ThreadBase<E> {
+ /**
+ * An interface for injecting custom behaviour for spilling and merging phases.
+ * @param <E>
+ */
+ interface SpillingBehaviour<E> {
+ default void open() {}
+
+ default void close() {}
+
+ /**
+ * An method that allows adjusting the spilling phase. We can inject e.g. combining the elements while spilling.
+ */
+ void spillBuffer(
+ CircularElement<E> element,
+ ChannelWriterOutputView output,
+ LargeRecordHandler<E> largeRecordHandler) throws IOException;
+
+ /**
+ * An method that allows adjusting the merging phase. We can inject e.g. combining the spilled elements.
+ */
+ void mergeRecords(MergeIterator<E> mergeIterator, ChannelWriterOutputView output) throws IOException;
+ }
+
+ /** Logging. */
+ private static final Logger LOG = LoggerFactory.getLogger(SpillingThread.class);
+
+ private final MemoryManager memManager; // memory manager to release memory
+
+ private final IOManager ioManager; // I/O manager to create channels
+
+ private final TypeSerializer<E> serializer; // The serializer for the data type
+
+ private final TypeComparator<E> comparator; // The comparator that establishes the order relation.
+
+ private final List<MemorySegment> writeMemory; // memory segments for writing
+
+ private final List<MemorySegment> mergeReadMemory; // memory segments for sorting/reading
+
+ private final int maxFanIn;
+
+ private final SpillChannelManager spillChannelManager;
+
+ private final LargeRecordHandler<E> largeRecordHandler;
+
+ private final SpillingBehaviour<E> spillingBehaviour;
+
+ private boolean spillingBehaviourOpened = false;
+
+ private final int minNumWriteBuffers;
+
+ private final int maxNumWriteBuffers;
+
+ /**
+ * Creates the spilling thread.
+ * @param exceptionHandler The exception handler to call for all exceptions.
+ * @param dispatcher The queues used to pass buffers between the threads.
+ * @param memManager The memory manager used to allocate buffers for the readers and writers.
+ * @param ioManager The I/I manager used to instantiate readers and writers from.
+ * @param serializer
+ * @param comparator
+ * @param sortReadMemory
+ * @param writeMemory
+ * @param maxNumFileHandles
+ * @param spillingChannelManager
+ * @param largeRecordHandler
+ * @param minNumWriteBuffers
+ * @param maxNumWriteBuffers
+ */
+ SpillingThread(
+ ExceptionHandler<IOException> exceptionHandler,
+ StageMessageDispatcher<E> dispatcher,
+ MemoryManager memManager,
+ IOManager ioManager,
+ TypeSerializer<E> serializer,
+ TypeComparator<E> comparator,
+ List<MemorySegment> sortReadMemory,
+ List<MemorySegment> writeMemory,
+ int maxNumFileHandles,
+ SpillChannelManager spillingChannelManager,
+ LargeRecordHandler<E> largeRecordHandler,
+ SpillingBehaviour<E> spillingBehaviour,
+ int minNumWriteBuffers,
+ int maxNumWriteBuffers) {
+ super(exceptionHandler, "SortMerger spilling thread", dispatcher);
+ this.memManager = memManager;
+ this.ioManager = ioManager;
+ this.serializer = serializer;
+ this.comparator = comparator;
+ this.mergeReadMemory = sortReadMemory;
+ this.writeMemory = writeMemory;
+ this.maxFanIn = maxNumFileHandles;
+ this.spillChannelManager = spillingChannelManager;
+ this.largeRecordHandler = largeRecordHandler;
+ this.spillingBehaviour = spillingBehaviour;
+ this.minNumWriteBuffers = minNumWriteBuffers;
+ this.maxNumWriteBuffers = maxNumWriteBuffers;
+ }
+
+ /**
+ * Entry point of the thread.
+ */
+ public void go() throws IOException {
+
+ final Queue<CircularElement<E>> cache = new ArrayDeque<>();
+ boolean cacheOnly = readCache(cache);
+
+ // check whether the thread was canceled
+ if (!isRunning()) {
+ return;
+ }
+
+ MutableObjectIterator<E> largeRecords = null;
+
+ // check if we can stay in memory with the large record handler
+ if (cacheOnly && largeRecordHandler != null && largeRecordHandler.hasData()) {
+ List<MemorySegment> memoryForLargeRecordSorting = new ArrayList<>();
+
+ CircularElement<E> circElement;
+ while ((circElement = this.dispatcher.poll(SortStage.READ)) != null) {
+ circElement.getBuffer().dispose();
+ memoryForLargeRecordSorting.addAll(circElement.getMemory());
+ }
+
+ if (memoryForLargeRecordSorting.isEmpty()) {
+ cacheOnly = false;
+ LOG.debug("Going to disk-based merge because of large records.");
+ } else {
+ LOG.debug("Sorting large records, to add them to in-memory merge.");
+ largeRecords = largeRecordHandler.finishWriteAndSortKeys(memoryForLargeRecordSorting);
+ }
+ }
+
+ // ------------------- In-Memory Merge ------------------------
+ if (cacheOnly) {
+ mergeInMemory(cache, largeRecords);
+ return;
+ }
+
+ // ------------------- Spilling Phase ------------------------
+ List<ChannelWithBlockCount> channelIDs = startSpilling(cache);
+
+ // ------------------- Merging Phase ------------------------
+
+ mergeOnDisk(channelIDs);
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ super.close();
+ if (spillingBehaviourOpened) {
+ this.spillingBehaviour.close();
+ this.spillingBehaviourOpened = false;
+ }
+ }
+
+ private boolean readCache(Queue<CircularElement<E>> cache) throws IOException {
+ CircularElement<E> element;
+
+ // ------------------- In-Memory Cache ------------------------
+ // fill cache
+ while (isRunning()) {
+ // take next element from queue
+ try {
+ element = this.dispatcher.take(SortStage.SPILL);
Review comment:
I though about it for a while after your comment. Even though we are slightly changing the behaviour I think it is actually fine. There are two situation where the thread can be interrupted.
1. When it is being closed via `ThreadBase#shutdown`. In this scenario both in the old and in the new code we are exiting the thread. In the new code we are throwing an exception, but it will be discarded in the `ThreadBase#internalExceptionHandler`.
2. If the thread is interrupted from outside. First of all I cannot think of exact scenario when this can happen. Nevertheless, I am pretty sure it should not happen during a successful processing and actually I doubt it is safe to continue on processing.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13357: [FLINK-19165] Refactor the UnilateralSortMerger
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13357:
URL: https://github.com/apache/flink/pull/13357#issuecomment-695957281
Thanks for the review @gaoyunhaii, I addressed your comments. Could you have another look?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org