You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2024/01/03 07:32:15 UTC
(flink) branch master updated: [FLINK-26694][table] Support lookup join via a multi-level inheritance of TableFunction
This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6e78eb18524 [FLINK-26694][table] Support lookup join via a multi-level inheritance of TableFunction
6e78eb18524 is described below
commit 6e78eb18524ead3abd60da0ca41751b45e0e2482
Author: Xianxun Ye <ye...@gmail.com>
AuthorDate: Wed Jan 3 15:32:06 2024 +0800
[FLINK-26694][table] Support lookup join via a multi-level inheritance of TableFunction
This closes #23684.
---
.../table/types/extraction/ExtractionUtils.java | 2 +-
.../planner/runtime/stream/sql/FunctionITCase.java | 62 +++++++++++++++++++++-
2 files changed, 62 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
index 27d689e5425..1b27c469667 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/ExtractionUtils.java
@@ -336,7 +336,7 @@ public final class ExtractionUtils {
public static Optional<Class<?>> extractSimpleGeneric(
Class<?> baseClass, Class<?> clazz, int pos) {
try {
- if (clazz.getSuperclass() != baseClass) {
+ if (!baseClass.isAssignableFrom(clazz)) {
return Optional.empty();
}
final Type t =
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
index 628c2f10e73..8f562b4f189 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
@@ -36,6 +36,8 @@ import org.apache.flink.table.catalog.CatalogFunction;
import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
@@ -1142,6 +1144,28 @@ public class FunctionITCase extends StreamingTestBase {
@Test
void testLookupTableFunction() throws ExecutionException, InterruptedException {
+ testLookupTableFunctionBase(LookupTableFunction.class.getName());
+ }
+
+ @Test
+ void testLookupTableFunctionWithHintLevel1() throws ExecutionException, InterruptedException {
+ testLookupTableFunctionBase(LookupTableWithHintLevel1Function.class.getName());
+ }
+
+ @Test
+ void testLookupTableFunctionWithoutHintLevel0()
+ throws ExecutionException, InterruptedException {
+ testLookupTableFunctionBase(LookupTableWithoutHintLevel0Function.class.getName());
+ }
+
+ @Test
+ void testLookupTableFunctionWithoutHintLevel1()
+ throws ExecutionException, InterruptedException {
+ testLookupTableFunctionBase(LookupTableWithoutHintLevel1Function.class.getName());
+ }
+
+ private void testLookupTableFunctionBase(String lookupTableFunctionClassName)
+ throws ExecutionException, InterruptedException {
final List<Row> sourceData = Arrays.asList(Row.of("Bob"), Row.of("Alice"));
final List<Row> sinkData =
@@ -1171,7 +1195,7 @@ public class FunctionITCase extends StreamingTestBase {
+ "WITH ("
+ " 'connector' = 'values',"
+ " 'lookup-function-class' = '"
- + LookupTableFunction.class.getName()
+ + lookupTableFunctionClassName
+ "'"
+ ")");
@@ -1656,6 +1680,42 @@ public class FunctionITCase extends StreamingTestBase {
}
}
+ /**
+ * Synchronous table function for {@link LookupTableSource} and inherits {@link TableFunction}
+ * at multiple levels.
+ */
+ public static class LookupTableWithHintLevel1Function extends LookupTableFunction {
+ public void eval(@DataTypeHint("STRING") StringData s) {
+ super.eval(s);
+ }
+ }
+
+ /** This is an empty synchronous table function. */
+ private static class LookupTableLevel0Function<T> extends TableFunction<T> {}
+
+ /**
+ * Synchronous table function that uses {@link RowData} type for {@link LookupTableSource} and
+ * inherits {@link TableFunction} at multiple levels.
+ */
+ public static class LookupTableWithoutHintLevel1Function
+ extends LookupTableLevel0Function<RowData> {
+ public void eval(@DataTypeHint("STRING") StringData s) {
+ collect(GenericRowData.of(StringData.fromString(s.toString()), new byte[0]));
+ collect(GenericRowData.of(StringData.fromString(s.toString()), s.toBytes()));
+ }
+ }
+
+ /**
+ * Synchronous table function that uses {@link RowData} type for {@link LookupTableSource} and
+ * inherits {@link TableFunction} at one level.
+ */
+ public static class LookupTableWithoutHintLevel0Function extends TableFunction<RowData> {
+ public void eval(@DataTypeHint("STRING") StringData s) {
+ collect(GenericRowData.of(StringData.fromString(s.toString()), new byte[0]));
+ collect(GenericRowData.of(StringData.fromString(s.toString()), s.toBytes()));
+ }
+ }
+
/** A specialized "compile time" function for returning the argument's data type. */
public static class TypeOfScalarFunction extends ScalarFunction implements SpecializedFunction {