flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [2/3] incubator-flink git commit: [FLINK-1278] [runtime] (part 1) Remove special code paths for the Record data type in the input readers and the source task.
Date Wed, 26 Nov 2014 17:51:26 GMT
[FLINK-1278] [runtime] (part 1) Remove special code paths for the Record data type in the input
readers and the source task.


Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/48b6d01c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/48b6d01c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/48b6d01c

Branch: refs/heads/master
Commit: 48b6d01c2ec73b10bcb3d668dd0bac4dc715f524
Parents: d858930
Author: Stephan Ewen <sewen@apache.org>
Authored: Wed Nov 26 17:18:55 2014 +0100
Committer: Stephan Ewen <sewen@apache.org>
Committed: Wed Nov 26 18:14:59 2014 +0100

----------------------------------------------------------------------
 .../plantranslate/NepheleJobGraphGenerator.java |   1 -
 .../flink/runtime/operators/DataSinkTask.java   |  22 +---
 .../flink/runtime/operators/DataSourceTask.java | 109 ++++++-------------
 .../runtime/operators/RegularPactTask.java      |  20 +---
 .../operators/util/RecordReaderIterator.java    |  57 ----------
 .../operators/chaining/ChainTaskTest.java       |   1 +
 .../operators/testutils/MockEnvironment.java    |  15 ++-
 7 files changed, 55 insertions(+), 170 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48b6d01c/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
b/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
index bb537b7..9c2efb3 100644
--- a/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
+++ b/flink-compiler/src/main/java/org/apache/flink/compiler/plantranslate/NepheleJobGraphGenerator.java
@@ -851,7 +851,6 @@ public class NepheleJobGraphGenerator implements Visitor<PlanNode>
{
 		final TaskConfig config = new TaskConfig(vertex.getConfiguration());
 
 		vertex.setInvokableClass(DataSinkTask.class);
-		vertex.getConfiguration().setInteger(DataSinkTask.DEGREE_OF_PARALLELISM_KEY, node.getDegreeOfParallelism());
 		vertex.setFormatDescription(getDescriptionForUserCode(node.getPactContract().getUserCodeWrapper()));
 		
 		// set user code

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48b6d01c/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
index b1185c2..74c625f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
@@ -37,22 +37,17 @@ import org.apache.flink.runtime.operators.chaining.ExceptionInChainedStubExcepti
 import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
 import org.apache.flink.runtime.operators.util.CloseableInputProvider;
 import org.apache.flink.runtime.operators.util.ReaderIterator;
-import org.apache.flink.runtime.operators.util.RecordReaderIterator;
 import org.apache.flink.runtime.operators.util.TaskConfig;
 import org.apache.flink.runtime.plugable.DeserializationDelegate;
-import org.apache.flink.types.Record;
 import org.apache.flink.util.MutableObjectIterator;
 
 /**
- * DataSinkTask which is executed by a Flink task manager.
- * The task hands the data to an output format.
+ * DataSinkTask which is executed by a task manager. The task hands the data to an output
format.
  * 
  * @see OutputFormat
  */
 public class DataSinkTask<IT> extends AbstractInvokable {
 	
-	public static final String DEGREE_OF_PARALLELISM_KEY = "sink.dop";
-	
 	// Obtain DataSinkTask Logger
 	private static final Logger LOG = LoggerFactory.getLogger(DataSinkTask.class);
 
@@ -339,17 +334,10 @@ public class DataSinkTask<IT> extends AbstractInvokable {
 		
 		this.inputTypeSerializerFactory = this.config.getInputSerializer(0, getUserCodeClassLoader());
 		
-		if (this.inputTypeSerializerFactory.getDataType() == Record.class) {
-			// record specific deserialization
-			MutableReader<Record> reader = (MutableReader<Record>) inputReader;
-			this.reader = (MutableObjectIterator<IT>)new RecordReaderIterator(reader);
-		} else {
-			// generic data type serialization
-			MutableReader<DeserializationDelegate<?>> reader = (MutableReader<DeserializationDelegate<?>>)
inputReader;
-			@SuppressWarnings({ "rawtypes" })
-			final MutableObjectIterator<?> iter = new ReaderIterator(reader, this.inputTypeSerializerFactory.getSerializer());
-			this.reader = (MutableObjectIterator<IT>)iter;
-		}
+		MutableReader<DeserializationDelegate<?>> reader = (MutableReader<DeserializationDelegate<?>>)
inputReader;
+		@SuppressWarnings({ "rawtypes" })
+		final MutableObjectIterator<?> iter = new ReaderIterator(reader, this.inputTypeSerializerFactory.getSerializer());
+		this.reader = (MutableObjectIterator<IT>)iter;
 		
 		// final sanity check
 		if (numGates != this.config.getNumInputs()) {

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48b6d01c/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
index bfd2507..2db652f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
@@ -16,7 +16,6 @@
  * limitations under the License.
  */
 
-
 package org.apache.flink.runtime.operators;
 
 import java.util.ArrayList;
@@ -41,13 +40,11 @@ import org.apache.flink.runtime.operators.chaining.ChainedCollectorMapDriver;
 import org.apache.flink.runtime.operators.chaining.ChainedDriver;
 import org.apache.flink.runtime.operators.chaining.ExceptionInChainedStubException;
 import org.apache.flink.runtime.operators.shipping.OutputCollector;
-import org.apache.flink.runtime.operators.shipping.RecordOutputCollector;
 import org.apache.flink.runtime.operators.util.TaskConfig;
-import org.apache.flink.types.Record;
 import org.apache.flink.util.Collector;
 
 /**
- * DataSourceTask which is executed by a Nephele task manager. The task reads data and uses
an 
+ * DataSourceTask which is executed by a task manager. The task reads data and uses an 
  * {@link InputFormat} to create records from the input.
  * 
  * @see org.apache.flink.api.common.io.InputFormat
@@ -140,80 +137,44 @@ public class DataSourceTask<OT> extends AbstractInvokable {
 				}
 				
 				try {
-					// ======= special-case the Record, to help the JIT and avoid some casts ======
-					if (record.getClass() == Record.class) {
-						Record typedRecord = (Record) record;
-						@SuppressWarnings("unchecked")
-						final InputFormat<Record, InputSplit> inFormat = (InputFormat<Record, InputSplit>)
format;
+					// special case to make the loops tighter
+					if (this.output instanceof OutputCollector) {
+						final OutputCollector<OT> output = (OutputCollector<OT>) this.output;
 						
-						if (this.output instanceof RecordOutputCollector) {
-							// Record going directly into network channels
-							final RecordOutputCollector output = (RecordOutputCollector) this.output;
-							while (!this.taskCanceled && !inFormat.reachedEnd()) {
-								// build next pair and ship pair if it is valid
-								typedRecord.clear();
-								Record returnedRecord = null;
-								if ((returnedRecord = inFormat.nextRecord(typedRecord)) != null) {
-									output.collect(returnedRecord);
-								}
-							}
-						} else if (this.output instanceof ChainedCollectorMapDriver) {
-							// Record going to a chained map task
-							@SuppressWarnings("unchecked")
-							final ChainedCollectorMapDriver<Record, ?> output = (ChainedCollectorMapDriver<Record,
?>) this.output;
+						// as long as there is data to read
+						while (!this.taskCanceled && !format.reachedEnd()) {
 							
-							// as long as there is data to read
-							while (!this.taskCanceled && !inFormat.reachedEnd()) {
-								// build next pair and ship pair if it is valid
-								typedRecord.clear();
-								if ((typedRecord = inFormat.nextRecord(typedRecord)) != null) {
-									// This is where map of UDF gets called
-									output.collect(typedRecord);
-								}
-							}
-						} else {
-							// Record going to some other chained task
-							@SuppressWarnings("unchecked")
-							final Collector<Record> output = (Collector<Record>) this.output;
-							// as long as there is data to read
-							while (!this.taskCanceled && !inFormat.reachedEnd()) {
-								// build next pair and ship pair if it is valid
-								typedRecord.clear();
-								if ((typedRecord = inFormat.nextRecord(typedRecord)) != null){
-									output.collect(typedRecord);
-								}
+							OT returned;
+							if ((returned = format.nextRecord(record)) != null) {
+								output.collect(returned);
+								record = returned;
 							}
 						}
-					} else {
-						// general types. we make a case distinction here for the common cases, in order to
help
-						// JIT method inlining
-						if (this.output instanceof OutputCollector) {
-							final OutputCollector<OT> output = (OutputCollector<OT>) this.output;
-							// as long as there is data to read
-							while (!this.taskCanceled && !format.reachedEnd()) {
-								// build next pair and ship pair if it is valid
-								if ((record = format.nextRecord(record)) != null) {
-									output.collect(record);
-								}
-							}
-						} else if (this.output instanceof ChainedCollectorMapDriver) {
-							@SuppressWarnings("unchecked")
-							final ChainedCollectorMapDriver<OT, ?> output = (ChainedCollectorMapDriver<OT,
?>) this.output;
-							// as long as there is data to read
-							while (!this.taskCanceled && !format.reachedEnd()) {
-								// build next pair and ship pair if it is valid
-								if ((record = format.nextRecord(record)) != null) {
-									output.collect(record);
-								}
+					}
+					else if (this.output instanceof ChainedCollectorMapDriver) {
+						@SuppressWarnings("unchecked")
+						final ChainedCollectorMapDriver<OT, ?> output = (ChainedCollectorMapDriver<OT,
?>) this.output;
+						
+						// as long as there is data to read
+						while (!this.taskCanceled && !format.reachedEnd()) {
+							
+							OT returned;
+							if ((returned = format.nextRecord(record)) != null) {
+								output.collect(returned);
+								record = returned;
 							}
-						} else {
-							final Collector<OT> output = this.output;
-							// as long as there is data to read
-							while (!this.taskCanceled && !format.reachedEnd()) {
-								// build next pair and ship pair if it is valid
-								if ((record = format.nextRecord(record)) != null) {
-									output.collect(record);
-								}
+						}
+					}
+					else {
+						final Collector<OT> output = this.output;
+						
+						// as long as there is data to read
+						while (!this.taskCanceled && !format.reachedEnd()) {
+							
+							OT returned;
+							if ((returned = format.nextRecord(record)) != null) {
+								output.collect(returned);
+								record = returned;
 							}
 						}
 					}
@@ -279,7 +240,7 @@ public class DataSourceTask<OT> extends AbstractInvokable {
 	
 	/**
 	 * Initializes the InputFormat implementation and configuration.
-l	 * 
+	 * 
 	 * @throws RuntimeException
 	 *         Throws if instance of InputFormat implementation can not be
 	 *         obtained.

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48b6d01c/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
index c1037b5..2a598ca 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
@@ -59,7 +59,6 @@ import org.apache.flink.runtime.operators.util.CloseableInputProvider;
 import org.apache.flink.runtime.operators.util.DistributedRuntimeUDFContext;
 import org.apache.flink.runtime.operators.util.LocalStrategy;
 import org.apache.flink.runtime.operators.util.ReaderIterator;
-import org.apache.flink.runtime.operators.util.RecordReaderIterator;
 import org.apache.flink.runtime.operators.util.TaskConfig;
 import org.apache.flink.runtime.plugable.DeserializationDelegate;
 import org.apache.flink.runtime.plugable.SerializationDelegate;
@@ -1031,20 +1030,11 @@ public class RegularPactTask<S extends Function, OT> extends
AbstractInvokable i
 	}
 	
 	protected MutableObjectIterator<?> createInputIterator(MutableReader<?> inputReader,
TypeSerializerFactory<?> serializerFactory) {
-
-		if (serializerFactory.getDataType().equals(Record.class)) {
-			// record specific deserialization
-			@SuppressWarnings("unchecked")
-			MutableReader<Record> reader = (MutableReader<Record>) inputReader;
-			return new RecordReaderIterator(reader);
-		} else {
-			// generic data type serialization
-			@SuppressWarnings("unchecked")
-			MutableReader<DeserializationDelegate<?>> reader = (MutableReader<DeserializationDelegate<?>>)
inputReader;
-			@SuppressWarnings({ "unchecked", "rawtypes" })
-			final MutableObjectIterator<?> iter = new ReaderIterator(reader, serializerFactory.getSerializer());
-			return iter;
-		}
+		@SuppressWarnings("unchecked")
+		MutableReader<DeserializationDelegate<?>> reader = (MutableReader<DeserializationDelegate<?>>)
inputReader;
+		@SuppressWarnings({ "unchecked", "rawtypes" })
+		final MutableObjectIterator<?> iter = new ReaderIterator(reader, serializerFactory.getSerializer());
+		return iter;
 	}
 
 	protected int getNumTaskInputs() {

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48b6d01c/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/RecordReaderIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/RecordReaderIterator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/RecordReaderIterator.java
deleted file mode 100644
index d95a42c..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/RecordReaderIterator.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.util;
-
-import java.io.IOException;
-
-import org.apache.flink.runtime.io.network.api.MutableReader;
-import org.apache.flink.types.Record;
-import org.apache.flink.util.MutableObjectIterator;
-
-/**
-* A {@link MutableObjectIterator} that wraps a Nephele Reader producing {@link Record}s.
-*/
-public final class RecordReaderIterator implements MutableObjectIterator<Record> {
-	
-	private final MutableReader<Record> reader;		// the source
-
-	/**
-	 * Creates a new iterator, wrapping the given reader.
-	 * 
-	 * @param reader The reader to wrap.
-	 */
-	public RecordReaderIterator(MutableReader<Record> reader) {
-		this.reader = reader;
-	}
-
-	@Override
-	public Record next(Record reuse) throws IOException {
-		try {
-			if (this.reader.next(reuse)) {
-				return reuse;
-			} else {
-				return null;
-			}
-		}
-		catch (InterruptedException e) {
-			throw new IOException("Reader interrupted.", e);
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48b6d01c/flink-runtime/src/test/java/org/apache/flink/runtime/operators/chaining/ChainTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/chaining/ChainTaskTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/chaining/ChainTaskTest.java
index 4c08ebd..0e029a8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/chaining/ChainTaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/chaining/ChainTaskTest.java
@@ -146,6 +146,7 @@ public class ChainTaskTest extends TaskTestBase {
 				// driver
 				combineConfig.setDriverStrategy(DriverStrategy.SORTED_GROUP_COMBINE);
 				combineConfig.setDriverComparator(compFact, 0);
+				combineConfig.setDriverComparator(compFact, 1);
 				combineConfig.setRelativeMemoryDriver(memoryFraction);
 				
 				// udf

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48b6d01c/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
index 60a81c1..9916b61 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
@@ -53,6 +53,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
 import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
 import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.plugable.DeserializationDelegate;
 import org.apache.flink.runtime.protocols.AccumulatorProtocol;
 import org.apache.flink.types.Record;
 import org.apache.flink.util.MutableObjectIterator;
@@ -69,7 +70,7 @@ public class MockEnvironment implements Environment, BufferProvider, LocalBuffer
 
 	private final Configuration taskConfiguration;
 
-	private final List<InputGate<Record>> inputs;
+	private final List<InputGate<DeserializationDelegate<Record>>> inputs;
 
 	private final List<OutputGate> outputs;
 
@@ -83,7 +84,7 @@ public class MockEnvironment implements Environment, BufferProvider, LocalBuffer
 	public MockEnvironment(long memorySize, MockInputSplitProvider inputSplitProvider, int bufferSize)
{
 		this.jobConfiguration = new Configuration();
 		this.taskConfiguration = new Configuration();
-		this.inputs = new LinkedList<InputGate<Record>>();
+		this.inputs = new LinkedList<InputGate<DeserializationDelegate<Record>>>();
 		this.outputs = new LinkedList<OutputGate>();
 
 		this.memManager = new DefaultMemoryManager(memorySize, 1);
@@ -172,7 +173,7 @@ public class MockEnvironment implements Environment, BufferProvider, LocalBuffer
 
 	}
 
-	private static class MockInputGate extends InputGate<Record> {
+	private static class MockInputGate extends InputGate<DeserializationDelegate<Record>>
{
 		
 		private MutableObjectIterator<Record> it;
 
@@ -182,15 +183,17 @@ public class MockEnvironment implements Environment, BufferProvider,
LocalBuffer
 		}
 
 		@Override
-		public void registerRecordAvailabilityListener(final RecordAvailabilityListener<Record>
listener) {
+		public void registerRecordAvailabilityListener(final RecordAvailabilityListener<DeserializationDelegate<Record>>
listener) {
 			super.registerRecordAvailabilityListener(listener);
 			this.notifyRecordIsAvailable(0);
 		}
 		
 		@Override
-		public InputChannelResult readRecord(Record target) throws IOException, InterruptedException
{
+		public InputChannelResult readRecord(DeserializationDelegate<Record> target) throws
IOException, InterruptedException {
 
-			if ((target = it.next(target)) != null) {
+			Record reuse = target != null ? target.getInstance() : null;
+			
+			if ((reuse = it.next(reuse)) != null) {
 				// everything comes from the same source channel and buffer in this mock
 				notifyRecordIsAvailable(0);
 				return InputChannelResult.INTERMEDIATE_RECORD_FROM_BUFFER;


Mime
View raw message