Add Sedona Flink SQL module

Author: radekaadek

flink/src/main/java/org/apache/sedona/flink/SedonaModule.java

@@ -0,0 +1,53 @@
+/*
+ * 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.sedona.flink;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.module.Module;
+
+public class SedonaModule implements Module {
+
+  private Map<String, FunctionDefinition> functions = new HashMap<>();
+
+  public SedonaModule() {
+    Stream<FunctionDefinition> funcs = Arrays.stream(Catalog.getFuncs());
+    Stream<FunctionDefinition> predicates = Arrays.stream(Catalog.getPredicates());
+
+    this.functions =
+        Stream.concat(funcs, predicates)
+            .collect(Collectors.toMap(func -> func.getClass().getSimpleName().toLowerCase(), func -> func));
+  }
+
+  @Override
+  public Set<String> listFunctions() {
+    return functions.keySet();
+  }
+
+  @Override
+  public Optional<FunctionDefinition> getFunctionDefinition(String name) {
+    return Optional.ofNullable(functions.get(name.toLowerCase()));
+  }
+}

flink/src/main/java/org/apache/sedona/flink/SedonaModuleFactory.java

@@ -0,0 +1,50 @@
+/*
+ * 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.sedona.flink;
+
+import java.util.Collections;
+import java.util.Set;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.table.factories.ModuleFactory;
+import org.apache.flink.table.module.Module;
+
+public class SedonaModuleFactory implements ModuleFactory {
+
+  public static final String IDENTIFIER = "sedona";
+
+  @Override
+  public Module createModule(Context context) {
+    return new SedonaModule();
+  }
+
+  @Override
+  public String factoryIdentifier() {
+    return IDENTIFIER;
+  }
+
+  @Override
+  public Set<ConfigOption<?>> requiredOptions() {
+    return Collections.emptySet();
+  }
+
+  @Override
+  public Set<ConfigOption<?>> optionalOptions() {
+    return Collections.emptySet();
+  }
+}

flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory

@@ -0,0 +1 @@
+org.apache.sedona.flink.SedonaModuleFactory