[KYUUBI-1352] Introduce FlinkSQLEngine infrastructure

apache · Nov 11, 2021 · cf1ad6b · cf1ad6b
1 parent 82cca0e
commit cf1ad6b
Show file tree

Hide file tree

Showing 40 changed files with 4,842 additions and 0 deletions.
diff --git a/externals/kyuubi-flink-sql-engine/pom.xml b/externals/kyuubi-flink-sql-engine/pom.xml
@@ -0,0 +1,135 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>kyuubi-parent</artifactId>
+        <groupId>org.apache.kyuubi</groupId>
+        <version>1.4.0-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>kyuubi-flink-sql-engine_${scala.binary.version}</artifactId>
+
+    <dependencies>
+        <!-- kyuubi -->
+        <dependency>
+            <groupId>org.apache.kyuubi</groupId>
+            <artifactId>kyuubi-common_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kyuubi</groupId>
+            <artifactId>kyuubi-ha_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <!-- flink -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-common</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-java</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-runtime-blink_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-sql-parser</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-shaded-jackson</artifactId>
+            <version>2.10.1-9.0</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-yarn_${scala.binary.version}</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.kyuubi</groupId>
+            <artifactId>kyuubi-common_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>
diff --git a/...uubi-flink-sql-engine/src/main/java/org/apache/kyuubi/engine/flink/config/ConfigUtil.java b/...uubi-flink-sql-engine/src/main/java/org/apache/kyuubi/engine/flink/config/ConfigUtil.java
@@ -0,0 +1,102 @@
+/*
+ * 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.kyuubi.engine.flink.config;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonToken;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.io.IOContext;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YAMLParser;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+
+/**
+ * Auxiliary functions for configuration file handling.
+ */
+public class ConfigUtil {
+
+	private ConfigUtil() {
+		// private
+	}
+
+	/**
+	 * Normalizes key-value properties from Yaml in the normalized format of the Table API.
+	 */
+	public static DescriptorProperties normalizeYaml(Map<String, Object> yamlMap) {
+		final Map<String, String> normalized = new HashMap<>();
+		yamlMap.forEach((k, v) -> normalizeYamlObject(normalized, k, v));
+		final DescriptorProperties properties = new DescriptorProperties(true);
+		properties.putProperties(normalized);
+		return properties;
+	}
+
+	private static void normalizeYamlObject(Map<String, String> normalized, String key, Object value) {
+		if (value instanceof Map) {
+			final Map<?, ?> map = (Map<?, ?>) value;
+			map.forEach((k, v) -> normalizeYamlObject(normalized, key + "." + k, v));
+		} else if (value instanceof List) {
+			final List<?> list = (List<?>) value;
+			for (int i = 0; i < list.size(); i++) {
+				normalizeYamlObject(normalized, key + "." + i, list.get(i));
+			}
+		} else {
+			normalized.put(key, value.toString());
+		}
+	}
+
+	// --------------------------------------------------------------------------------------------
+
+	/**
+	 * Modified object mapper that converts to lower-case keys.
+	 */
+	public static class LowerCaseYamlMapper extends ObjectMapper {
+		public LowerCaseYamlMapper() {
+			super(new YAMLFactory() {
+				@Override
+				protected YAMLParser _createParser(InputStream in, IOContext ctxt) throws IOException {
+					final Reader r = _createReader(in, null, ctxt);
+					// normalize all key to lower case keys
+					return new YAMLParser(ctxt, _getBufferRecycler(), _parserFeatures, _yamlParserFeatures,
+						_objectCodec, r) {
+						@Override
+						public String getCurrentName() throws IOException {
+							if (_currToken == JsonToken.FIELD_NAME) {
+								return _currentFieldName.toLowerCase();
+							}
+							return super.getCurrentName();
+						}
+
+						@Override
+						public String getText() throws IOException {
+							if (_currToken == JsonToken.FIELD_NAME) {
+								return _currentFieldName.toLowerCase();
+							}
+							return super.getText();
+						}
+					};
+				}
+			});
+		}
+	}
+}