[FLINK-19473] Implement multi inputs sorting DataInput #13529

dawidwys · 2020-10-01T14:33:04Z

What is the purpose of the change

This PR implements sorting inputs for multi input operators/tasks. It is based on the work in #13521
First I sort all inputs individually and after that I do a sorting merge of all the inputs to make sure that operator sees all incoming records with the same key from all inputs.

Brief change log

For a more thorough explanation see the commits messages. Some highlights:

Some of the commits refactor the TwoInputStreamProcessor/MultiInputStreamProcessor etc. to reduce code duplication
Some of the commits extract the instantiation logic out of the processor and moves it to the Task. It makes it a) easier to instantiate in tests b) we need to pass less arguments to the ctor from the Task
One of the commits implements proper AvailabilityProvider handling logic, which did not work before.

Verifying this change

Added tests in:

MultiInputSortingDataInputsTest
SortedDataInputITCase
SortingBoundedInputITCase.java

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)

flinkbot · 2020-10-01T14:35:15Z

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 6270bf0 (Thu Oct 01 14:35:15 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

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 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

Bot commands

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

flinkbot · 2020-10-01T14:56:28Z

CI report:

f72f675 UNKNOWN
993afb5 Azure: SUCCESS

Bot commands

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

dawidwys · 2020-10-07T07:29:21Z

@flinkbot run azure

pnowojski

I really like most of the commits here. Well done @dawidwys with the refactoring/cleanups/deduplicating the code.

I think my biggest concerns are about the bug fix commit hidden in this PR.

Note, I didn't review the MultiInputSortingDataInputs code itself, I hope @aljoscha will be able to do the most work here. Please let me know if you would like me to take a look at something there after all.

...g-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java

...src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java

...k-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/EndOfInputAware.java

...ming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortedInputITCase.java

...aming-java/src/main/java/org/apache/flink/streaming/runtime/io/TwoInputSelectionHandler.java

...eaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java

pnowojski · 2020-10-07T14:22:26Z

...java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessorFactory.java

+			@SuppressWarnings("unchecked")
+			MultiInputSortingDataInputs<Object> multiInputs = new MultiInputSortingDataInputs<Object>(


(I remember someone complaining about unchecked warnings 🙈 and Object? 😈 )

Yes, I must admit I was a bit harsh on this one 😅

aljoscha · 2020-10-08T10:29:01Z

Just one comment regarding commits tags. I mostly use [minor] when it's minor refactoring changes because [hotfix] to me implies a fix for sth that was broken. Lately I have also seen Stephan do commits with just [refactor] ..., that's also an option.

aljoscha

Overall the changes look very good! I had one comment before about commit tags and I had some inline comments about exception messages and some possible code changes.

aljoscha · 2020-10-08T10:34:48Z

...src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInputs.java

+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * An entry class for creating coupled, sorting inputs. The inputs are sorted independently and afterwards


I forget if we mentioned this in the one-input implementations but we should mention that this does sorting by key and timestamp (in that order). It's not some generic sorting utility.

aljoscha · 2020-10-08T10:37:11Z

...src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInputs.java

+		public CompletableFuture<Void> prepareSnapshot(
+				ChannelStateWriter channelStateWriter,
+				long checkpointId) {
+			throw new UnsupportedOperationException();


Again, not sure I caught this on the one-input version but we should give some information about why it's not supported.

aljoscha · 2020-10-08T10:42:19Z

...src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInputs.java

+	 */
+	private class SortingPhaseDataOutput implements PushingAsyncDataInput.DataOutput<Object> {
+
+		int currentIdx;


Would it maybe be better to have one SortingPhaseDataOutput per indexed input. It could also have "hard links" to the things it needs like key selector, sorter, etc.

And then we also wouldn't need this mutable field.

aljoscha · 2020-10-08T10:49:28Z

...src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInputs.java

+		public void close() throws IOException {
+			IOException ex = null;
+			try {
+				inputs[idx].close();


Instead of using array accesses for inputs and sorters, could we just get them once in the constructor and store in fields? This is related to the question below about using one SortingPhaseDataOutput per input.

aljoscha

This looks very good! +1 to merge once CI is green.

I had two nitpicks about indentation and Javadocs.

aljoscha · 2020-10-08T14:42:26Z

.../src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInput.java

+ * serialized key is constant, or {@link VariableLengthByteKeyComparator} otherwise.
+ *
+ * <p>Watermarks, stream statuses, nor latency markers are propagated downstream as they do not make
+ * sense with buffered records. The input emits a MAX_WATERMARK after all records.


It emits the maximum watermark it has seen, MAX_WATERMARK could be interpreted as Long.MAX_VALUE.

aljoscha · 2020-10-08T14:45:40Z

.../src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInput.java

+			.mapToObj(
+				idx -> {
+					try {
+					KeyAndValueSerializer<Object> keyAndValueSerializer = new KeyAndValueSerializer<>(


Indentation seems off here. (Another point for having an automatic formatting tool... 😅)

dawidwys · 2020-10-08T14:50:21Z

@pnowojski I'd like to double check with you, are you fine with the current state of the PR?

pnowojski

LGTM % a couple of nits

...eaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamOneInputProcessor.java

.../src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInput.java

…cessors I replaced the OperatorChain in StreamOneInputProcessor and other InputProcessors with a BoundedMultiInput. Moroever I made the OperatorChain implement BoundedMultiInput interface. It makes it easier to instantiate StreamOneInputProcessor in tests without the need to instantiate a whole OperatorChain.

I implement a MultiInputSortingDataInputs which is kind of a factory for multiple, related inputs. In case of sorting inputs of Two/Multiple input operators not only the independent inputs should be sorted, but also records across different inputs. Therefore the produced inputs share a common context to synchronize which input should emit records next. The coordination is done via inputs Availability. Only one of the inputs, with the current smallest element, is available at a time.

…taInput

pnowojski

one optional nit besides, LGTM (assuming azure will be green).

Thanks for the changes @dawidwys.

pnowojski · 2020-10-12T13:44:48Z

...eaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamTwoInputProcessor.java

+			inputSelectionHandler.nextSelection();
 		} else {
 			secondInputStatus = processor2.processInput();
-			checkFinished(secondInputStatus);
+			inputSelectionHandler.nextSelection();


nitty nit: move inputSelectionHandler.nextSelection(); outside of the if/else?

rmetzger added the review=description? label Oct 1, 2020

dawidwys requested review from aljoscha and pnowojski October 1, 2020 14:35

rmetzger added the component=Runtime/Task label Oct 1, 2020

dawidwys assigned aljoscha Oct 5, 2020

dawidwys mentioned this pull request Oct 5, 2020

[FLINK-19472] Implement a one input sorting DataInput #13521

Closed

dawidwys force-pushed the FLINK-19473 branch 3 times, most recently from d91430e to e1aaede Compare October 6, 2020 13:33

dawidwys marked this pull request as ready for review October 6, 2020 13:40

dawidwys force-pushed the FLINK-19473 branch from e1aaede to 055e55a Compare October 6, 2020 13:43

dawidwys force-pushed the FLINK-19473 branch 2 times, most recently from 0adec04 to 0678c76 Compare October 7, 2020 14:04

pnowojski requested changes Oct 7, 2020

View reviewed changes

dawidwys force-pushed the FLINK-19473 branch from 0678c76 to 43a60de Compare October 8, 2020 08:34

aljoscha reviewed Oct 8, 2020

View reviewed changes

dawidwys force-pushed the FLINK-19473 branch from 43a60de to 9d86c10 Compare October 8, 2020 13:29

aljoscha approved these changes Oct 8, 2020

View reviewed changes

pnowojski reviewed Oct 8, 2020

View reviewed changes

...eaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamOneInputProcessor.java Outdated Show resolved Hide resolved

...eaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamOneInputProcessor.java Show resolved Hide resolved

dawidwys force-pushed the FLINK-19473 branch from 9d86c10 to 5eaf633 Compare October 9, 2020 06:56

pnowojski reviewed Oct 10, 2020

View reviewed changes

.../src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInput.java Outdated Show resolved Hide resolved

dawidwys added 4 commits October 12, 2020 12:59

[refactor] Reuse StreamOneInputProcessor in StreamMultiProcessor

ebf6b86

[refactor] Reuse StreamOneInputProcessor in StreamTwoInputProcessor

7a2c8c5

[refactor] Remove SourceInputProcessor from StreamMultipleInputProcessor

2baa151

[refactor] Extract StreamMultipleInputProcessor factory method

c75b3e1

dawidwys added 3 commits October 12, 2020 12:59

[refactor] Call nextSelection after each emitNext

6cb4b3c

dawidwys force-pushed the FLINK-19473 branch from 5eaf633 to f72f675 Compare October 12, 2020 11:34

dawidwys added 7 commits October 12, 2020 13:42

fixup! Use selection handler

c2eb4ce

[minor] Minor fixes in javadocs and an exception message in SortingDa…

e1e8b67

…taInput

[refactor] Extract StreamTwoInputProcessor factory method

aa90566

[FLINK-19473] Enable two input sorting in TwoInputStreamTask

7f0f5a6

fixup! Use sorting selection handler in Two input task

cebacc8

[FLINK-19473] Enable sorting inputs in MultiInputTasks

0876bf8

fixup! Use selection handler for multi input task

993afb5

dawidwys force-pushed the FLINK-19473 branch from f72f675 to 993afb5 Compare October 12, 2020 11:50

pnowojski approved these changes Oct 12, 2020

View reviewed changes

dawidwys closed this in 77e4e3b Oct 12, 2020

		@SuppressWarnings("unchecked")
		MultiInputSortingDataInputs<Object> multiInputs = new MultiInputSortingDataInputs<Object>(

[FLINK-19473] Implement multi inputs sorting DataInput #13529

[FLINK-19473] Implement multi inputs sorting DataInput #13529

Uh oh!

Conversation

dawidwys commented Oct 1, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Oct 1, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Oct 1, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

dawidwys commented Oct 7, 2020

Uh oh!

pnowojski left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aljoscha commented Oct 8, 2020

Uh oh!

aljoscha left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aljoscha left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dawidwys commented Oct 8, 2020

Uh oh!

pnowojski left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pnowojski left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

dawidwys commented Oct 1, 2020 •

edited

Loading

flinkbot commented Oct 1, 2020 •

edited

Loading

pnowojski left a comment •

edited

Loading