[Flink-5734] code generation for normalizedkey sorter

docs/dev/execution_configuration.md

@@ -55,12 +55,14 @@ With the closure cleaner disabled, it might happen that an anonymous user functi
 
 - `getExecutionMode()` / `setExecutionMode()`. The default execution mode is PIPELINED. Sets the execution mode to execute the program. The execution mode defines whether data exchanges are performed in a batch or on a pipelined manner.
 
-- `enableForceKryo()` / **`disableForceKryo`**. Kryo is not forced by default. Forces the GenericTypeInformation to use the Kryo serializer for POJOS even though we could analyze them as a POJO. In some cases this might be preferable. For example, when Flink's internal serializers fail to handle a POJO properly.
+- `enableForceKryo()` / **`disableForceKryo()`**. Kryo is not forced by default. Forces the GenericTypeInformation to use the Kryo serializer for POJOS even though we could analyze them as a POJO. In some cases this might be preferable. For example, when Flink's internal serializers fail to handle a POJO properly.
 
 - `enableForceAvro()` / **`disableForceAvro()`**. Avro is not forced by default. Forces the Flink AvroTypeInformation to use the Avro serializer instead of Kryo for serializing Avro POJOs.
 
 - `enableObjectReuse()` / **`disableObjectReuse()`** By default, objects are not reused in Flink. Enabling the object reuse mode will instruct the runtime to reuse user objects for better performance. Keep in mind that this can lead to bugs when the user-code function of an operation is not aware of this behavior.
 
+- `enableCodeGenerationForSorters()` / **`disableCodeGenerationForSorters()`**. This setting can increase the performance of batch jobs by instructing Flink to generate specialized code for every sorting situation (i.e., type and comparator).
+
 - **`enableSysoutLogging()`** / `disableSysoutLogging()` JobManager status updates are printed to `System.out` by default. This setting allows to disable this behavior.
 
 - `getGlobalJobParameters()` / `setGlobalJobParameters()` This method allows users to set custom objects as a global configuration for the job. Since the `ExecutionConfig` is accessible in all user defined functions, this is an easy method for making configuration globally available in a job.

flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java

@@ -126,6 +126,12 @@ public class ExecutionConfig implements Serializable, Archiveable<ArchivedExecut
 
 	private long autoWatermarkInterval = 0;
 
+	/**
+	 * The flag determines whether a custom NormalizedKeySorter will be dynamically created
+	 * for underlying data
+	 */
+	private boolean codeGenerationForSortersEnabled = false;
+
 	/**
 	 * Interval in milliseconds for sending latency tracking marks from the sources to the sinks.
 	 */
@@ -610,7 +616,33 @@ public ExecutionConfig disableObjectReuse() {
 	public boolean isObjectReuseEnabled() {
 		return objectReuse;
 	}
-
+
+	/**
+	 * Enables code generation for sorters. This can increase the performance of batch jobs, as
+	 * Flink generates specialized sorting code for every type and comparator.
+	 *
+	 * <p>This is disabled by default.
+	 */
+	public ExecutionConfig enableCodeGenerationForSorters() {
+		this.codeGenerationForSortersEnabled = true;
+		return this;
+	}
+
+	/**
+	 * Disables code generation for sorters. @see #enableCodeGenerationForSorters()
+	 */
+	public ExecutionConfig disableCodeGenerationForSorters() {
+		this.codeGenerationForSortersEnabled = false;
+		return this;
+	}
+
+	/**
+	 * Returns whether code generation for sorters has been enabled or disabled. @see #enableCodeGenerationForSorters()
+	 */
+	public boolean isCodeGenerationForSortersEnabled() {
+		return codeGenerationForSortersEnabled;
+	}
+
 	/**
 	 * Sets the {@link CodeAnalysisMode} of the program. Specifies to which extent user-defined
 	 * functions are analyzed in order to give the Flink optimizer an insight of UDF internals
@@ -877,7 +909,8 @@ public boolean equals(Object obj) {
 				registeredKryoTypes.equals(other.registeredKryoTypes) &&
 				registeredPojoTypes.equals(other.registeredPojoTypes) &&
 				taskCancellationIntervalMillis == other.taskCancellationIntervalMillis &&
-				useSnapshotCompression == other.useSnapshotCompression;
+				useSnapshotCompression == other.useSnapshotCompression &&
+				codeGenerationForSortersEnabled == other.codeGenerationForSortersEnabled;
 
 		} else {
 			return false;
@@ -905,7 +938,9 @@ public int hashCode() {
 			registeredKryoTypes,
 			registeredPojoTypes,
 			taskCancellationIntervalMillis,
-			useSnapshotCompression);
+			useSnapshotCompression,
+			codeGenerationForSortersEnabled
+		);
 	}
 
 	public boolean canEqual(Object obj) {
@@ -918,7 +953,6 @@ public ArchivedExecutionConfig archive() {
 		return new ArchivedExecutionConfig(this);
 	}
 
-
 	// ------------------------------ Utilities  ----------------------------------
 
 	public static class SerializableSerializer<T extends Serializer<?> & Serializable> implements Serializable {

flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeComparator.java

@@ -157,7 +157,7 @@ public boolean supportsCompareAgainstReference() {
 	 *
 	 * @param first The first record.
 	 * @param second The second record.
-	 * @return An integer defining the oder among the objects in the same way as {@link java.util.Comparator#compare(Object, Object)}.
+	 * @return An integer defining the order among the objects in the same way as {@link java.util.Comparator#compare(Object, Object)}.
 	 * 
 	 *  @see java.util.Comparator#compare(Object, Object)
 	 */

flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java

@@ -1290,4 +1290,13 @@ else if (seg2.address > seg2.addressLimit) {
 					String.format("offset1=%d, offset2=%d, len=%d, bufferSize=%d, address1=%d, address2=%d",
 							offset1, offset2, len, tempBuffer.length, this.address, seg2.address));
 	}
+
+	public final byte getByte(int index){
+		return get(index);
+	}
+
+	public final void putByte(int index, byte b){
+		put(index,b);
+	}
+
 }

flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/drivers/DriverBaseITCase.java

@@ -190,6 +190,7 @@ private String getSystemOutput(String[] args) throws Exception {
 		switch (mode) {
 			case CLUSTER:
 			case COLLECTION:
+			case CLUSTER_WITH_CODEGENERATION_ENABLED:
 				args = ArrayUtils.add(args, "--__disable_object_reuse");
 				break;
 

flink-libraries/flink-table/pom.xml

@@ -46,7 +46,7 @@ under the License.
 		<dependency>
 			<groupId>org.codehaus.janino</groupId>
 			<artifactId>janino</artifactId>
-			<version>3.0.7</version>
+			<version>${janino.version}</version>
 		</dependency>
 
 		<dependency>

flink-runtime/pom.xml

@@ -204,6 +204,18 @@ under the License.
 			<artifactId>reflections</artifactId>
 		</dependency>
 
+		<dependency>
+			<groupId>org.freemarker</groupId>
+			<artifactId>freemarker</artifactId>
+			<version>2.3.26-incubating</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.codehaus.janino</groupId>
+			<artifactId>janino</artifactId>
+			<version>${janino.version}</version>
+		</dependency>
+
 	</dependencies>
 
 	<build>

flink-runtime/src/main/java/org/apache/flink/runtime/codegeneration/SorterFactory.java

@@ -0,0 +1,197 @@
+/*
+ * 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.codegeneration;
+
+import freemarker.template.Version;
+import org.apache.flink.api.common.ExecutionConfig;
+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.operators.sort.FixedLengthRecordSorter;
+import org.apache.flink.runtime.operators.sort.InMemorySorter;
+import org.apache.flink.runtime.operators.sort.NormalizedKeySorter;
+
+import freemarker.template.Configuration;
+import freemarker.template.Template;
+import freemarker.template.TemplateException;
+import freemarker.template.TemplateExceptionHandler;
+import org.codehaus.commons.compiler.CompileException;
+import org.codehaus.janino.SimpleCompiler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.StringWriter;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * {@link SorterFactory} is a singleton class that provides functionalities to create the most suitable sorter
+ * for underlying data based on {@link TypeComparator}.
+ * Note: the generated code can be inspected by configuring Janino to write the code that is being compiled
+ * to a file, see http://janino-compiler.github.io/janino/#debugging
+ */
+public class SorterFactory {
+	// ------------------------------------------------------------------------
+	//                                   Constants
+	// ------------------------------------------------------------------------
+	private static final Logger LOG = LoggerFactory.getLogger(SorterFactory.class);
+
+	/** Fixed 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;
+
+	// ------------------------------------------------------------------------
+	//                                   Singleton Attribute
+	// ------------------------------------------------------------------------
+	private static SorterFactory sorterFactory;
+
+	// ------------------------------------------------------------------------
+	//                                   Attributes
+	// ------------------------------------------------------------------------
+	private SimpleCompiler classCompiler;
+	private HashMap<String, Constructor> constructorCache;
+	private final Template template;
+
+	/**
+	 * This is only for testing. If an error occurs, we want to fail the test, instead of falling back
+	 * to a non-generated sorter.
+	 */
+	public boolean forceCodeGeneration = false;
+
+	/**
+	 * Constructor.
+	 */
+	private SorterFactory() {
+		this.classCompiler = new SimpleCompiler();
+		this.constructorCache = new HashMap<>();
+		Configuration templateConf;
+		templateConf = new Configuration(new Version(2,3,26));
+		templateConf.setClassForTemplateLoading(SorterFactory.class, "/templates");
+		templateConf.setDefaultEncoding("UTF-8");
+		templateConf.setTemplateExceptionHandler(TemplateExceptionHandler.RETHROW_HANDLER);
+		try {
+			template = templateConf.getTemplate(SorterTemplateModel.TEMPLATE_NAME);
+		} catch (IOException e) {
+			throw new RuntimeException("Couldn't read sorter template.", e);
+		}
+	}
+
+	/**
+	 * A method to get a singleton instance
+	 * or create one if it hasn't been created yet.
+	 * @return
+	 */
+	public static synchronized SorterFactory getInstance() {
+		if (sorterFactory == null){
+			sorterFactory = new SorterFactory();
+		}
+
+		return sorterFactory;
+	}
+
+
+	/**
+	 * Create a sorter for the given type comparator and
+	 * assign serializer, comparator and memory to the sorter.
+	 * @param serializer
+	 * @param comparator
+	 * @param memory
+	 * @return
+	 */
+	public <T> InMemorySorter<T> createSorter(ExecutionConfig config, TypeSerializer<T> serializer,
+											TypeComparator<T> comparator, List<MemorySegment> memory) {
+		if (config.isCodeGenerationForSortersEnabled()){
+			try {
+				return createCodegenSorter(serializer, comparator, memory);
+			} catch (IOException | TemplateException | ClassNotFoundException | IllegalAccessException |
+					InstantiationException | NoSuchMethodException | InvocationTargetException | CompileException e) {
+
+				String msg = "Serializer: " + serializer +
+						"[" + serializer + "], comparator: [" + comparator + "], exception: " + e.toString();
+				if (!forceCodeGeneration) {
+					LOG.warn("An error occurred while trying to create a code-generated sorter. " +
+							"Using non-codegen sorter instead. " + msg);
+					return createNonCodegenSorter(serializer, comparator, memory);
+				} else {
+					throw new RuntimeException("An error occurred while trying to create a code-generated sorter. " +
+							"Failing the job, because forceCodeGeneration is true. " + msg);
+				}
+			}
+		} else {
+			return createNonCodegenSorter(serializer, comparator, memory);
+		}
+	}
+
+	private <T> InMemorySorter<T> createCodegenSorter(TypeSerializer<T> serializer, TypeComparator<T> comparator,
+													List<MemorySegment> memory)
+			throws IOException, TemplateException, ClassNotFoundException, IllegalAccessException,
+			InstantiationException, NoSuchMethodException, InvocationTargetException, CompileException {
+		SorterTemplateModel sorterModel = new SorterTemplateModel(comparator);
+
+		Constructor sorterConstructor;
+
+		synchronized (this){
+			if (constructorCache.getOrDefault(sorterModel.getSorterName(), null) != null) {
+				sorterConstructor = constructorCache.get(sorterModel.getSorterName());
+			} else {
+				String sorterName = sorterModel.getSorterName();
+
+				StringWriter generatedCodeWriter = new StringWriter();
+				template.process(sorterModel.getTemplateVariables(), generatedCodeWriter);
+
+				this.classCompiler.cook(generatedCodeWriter.toString());
+
+				sorterConstructor = this.classCompiler.getClassLoader().loadClass(sorterName).getConstructor(
+						TypeSerializer.class, TypeComparator.class, List.class
+				);
+
+				constructorCache.put(sorterName, sorterConstructor);
+			}
+		}
+
+		@SuppressWarnings("unchecked")
+		InMemorySorter<T> sorter = (InMemorySorter<T>) sorterConstructor.newInstance(serializer, comparator, memory);
+
+		if (LOG.isInfoEnabled()){
+			LOG.info("Using a custom sorter : " + sorter.toString());
+		}
+
+		return sorter;
+	}
+
+	private <T> InMemorySorter<T> createNonCodegenSorter(TypeSerializer<T> serializer, TypeComparator<T> comparator,
+														List<MemorySegment> memory) {
+		InMemorySorter<T> sorter;
+		// instantiate a fix-length in-place sorter, if possible, otherwise the out-of-place sorter
+		if (comparator.supportsSerializationWithKeyNormalization() &&
+				serializer.getLength() > 0 && serializer.getLength() <= THRESHOLD_FOR_IN_PLACE_SORTING &&
+				comparator.isNormalizedKeyPrefixOnly(comparator.getNormalizeKeyLen())) {
+			// Note about the last part of the condition:
+			// FixedLengthRecordSorter doesn't do an additional check after the bytewise comparison, so
+			// we cannot choose that if the normalized key doesn't always determine the order.
+			// (cf. the part of NormalizedKeySorter.compare after the if)
+			sorter = new FixedLengthRecordSorter<>(serializer, comparator.duplicate(), memory);
+		} else {
+			sorter = new NormalizedKeySorter<>(serializer, comparator.duplicate(), memory);
+		}
+		return sorter;
+	}
+}