This is an automated email from the ASF dual-hosted git repository.
twalthr 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 2897ab7db10 [FLINK-36705][table-common] Add initial
ProcessTableFunction class and annotations
2897ab7db10 is described below
commit 2897ab7db1081c9bf1f477dc221b84e7c6686a6d
Author: Timo Walther <[email protected]>
AuthorDate: Fri Nov 29 10:15:29 2024 +0100
[FLINK-36705][table-common] Add initial ProcessTableFunction class and
annotations
---
.../flink/table/annotation/ArgumentHint.java | 7 +-
.../flink/table/annotation/ArgumentTrait.java | 32 ++-
.../flink/table/annotation/DataTypeHint.java | 3 +-
.../flink/table/annotation/FunctionHint.java | 32 ++-
.../{ArgumentHint.java => StateHint.java} | 50 ++---
.../apache/flink/table/functions/FunctionKind.java | 2 +
.../table/functions/ProcessTableFunction.java | 246 +++++++++++++++++++++
.../flink/table/functions/TableFunction.java | 16 +-
.../flink/table/functions/TableSemantics.java | 128 +++++++++++
.../types/extraction/BaseMappingExtractor.java | 3 +-
.../table/types/extraction/FunctionTemplate.java | 5 +-
.../extraction/TypeInferenceExtractorTest.java | 14 +-
12 files changed, 482 insertions(+), 56 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentHint.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentHint.java
index 87a438c6eb5..70f51f2a096 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentHint.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentHint.java
@@ -31,8 +31,9 @@ import java.lang.annotation.Target;
* <p>An {@code ArgumentHint} can be used to provide hints about the name,
optionality, and data
* type of argument.
*
- * <p>{@code @ArgumentHint(name = "in1", type = @DataTypeHint("STRING"),
isOptional = false)} is a
- * scalar argument with the data type STRING, named "in1", and cannot be
omitted when calling.
+ * <p>For example, {@code @ArgumentHint(name = "in1", type =
@DataTypeHint("STRING"), isOptional =
+ * false)} is a scalar argument with the data type STRING, named "in1", and
cannot be omitted when
+ * calling.
*
* @see FunctionHint
*/
@@ -49,7 +50,7 @@ public @interface ArgumentHint {
ArgumentTrait[] value() default {ArgumentTrait.SCALAR};
/**
- * The name of the argument.
+ * The name of the argument. It must be unique among other arguments.
*
* <p>This can be used to provide a descriptive name for the argument.
*/
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
index 5f6f7bc6ea7..df44b5a64f7 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentTrait.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.annotation;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.functions.ProcessTableFunction;
import org.apache.flink.table.types.inference.StaticArgumentTrait;
import java.util.Arrays;
@@ -43,31 +44,38 @@ public enum ArgumentTrait {
/**
* An argument that accepts a table "as row" (i.e. with row semantics).
This trait only applies
- * to {@code ProcessTableFunction} (PTF).
+ * to {@link ProcessTableFunction} (PTF).
*
- * <p>For scalability, input tables are distributed into virtual
processors. Each virtual
- * processor executes a PTF instance and has access only to a share of the
entire table. The
- * argument declaration decides about the size of the share and
co-location of data.
+ * <p>For scalability, input tables are distributed across so-called
"virtual processors". A
+ * virtual processor, as defined by the SQL standard, executes a PTF
instance and has access
+ * only to a portion of the entire table. The argument declaration decides
about the size of the
+ * portion and co-location of data. Conceptually, tables can be processed
either "as row" (i.e.
+ * with row semantics) or "as set" (i.e. with set semantics).
*
* <p>A table with row semantics assumes that there is no correlation
between rows and each row
- * can be processed independently. The framework is free in how to
distribute rows among virtual
- * processors and each virtual processor has access only to the currently
processed row.
+ * can be processed independently. The framework is free in how to
distribute rows across
+ * virtual processors and each virtual processor has access only to the
currently processed row.
*/
TABLE_AS_ROW(StaticArgumentTrait.TABLE_AS_ROW),
/**
* An argument that accepts a table "as set" (i.e. with set semantics).
This trait only applies
- * to {@code ProcessTableFunction} (PTF).
+ * to {@link ProcessTableFunction} (PTF).
*
- * <p>For scalability, input tables are distributed into virtual
processors. Each virtual
- * processor executes a PTF instance and has access only to a share of the
entire table. The
- * argument declaration decides about the size of the share and
co-location of data.
+ * <p>For scalability, input tables are distributed across so-called
"virtual processors". A
+ * virtual processor, as defined by the SQL standard, executes a PTF
instance and has access
+ * only to a portion of the entire table. The argument declaration decides
about the size of the
+ * portion and co-location of data. Conceptually, tables can be processed
either "as row" (i.e.
+ * with row semantics) or "as set" (i.e. with set semantics).
*
* <p>A table with set semantics assumes that there is a correlation
between rows. When calling
* the function, the PARTITION BY clause defines the columns for
correlation. The framework
* ensures that all rows belonging to same set are co-located. A PTF
instance is able to access
- * all rows belonging to the same set. In other words: The virtual
processor is scoped under a
- * key context.
+ * all rows belonging to the same set. In other words: The virtual
processor is scoped by a key
+ * context.
+ *
+ * <p>It is also possible not to provide a key ({@link
#OPTIONAL_PARTITION_BY}), in which case
+ * only one virtual processor handles the entire table, thereby losing
scalability benefits.
*/
TABLE_AS_SET(StaticArgumentTrait.TABLE_AS_SET),
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/DataTypeHint.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/DataTypeHint.java
index 382012aa107..df97fd8d25c 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/DataTypeHint.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/DataTypeHint.java
@@ -81,8 +81,7 @@ public @interface DataTypeHint {
// Note to implementers:
// Because "null" is not supported as an annotation value. Every
annotation parameter *must*
- // have
- // some representation for unknown values in order to merge multi-level
annotations.
+ // have some representation for unknown values in order to merge
multi-level annotations.
//
--------------------------------------------------------------------------------------------
// Explicit data type specification
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java
index 2610882807f..ef551d6e2b8 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java
@@ -19,6 +19,9 @@
package org.apache.flink.table.annotation;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.functions.UserDefinedFunction;
import org.apache.flink.table.types.inference.TypeInference;
@@ -175,13 +178,40 @@ public @interface FunctionHint {
ArgumentHint[] arguments() default {};
/**
- * Explicitly defines the intermediate result type that a function uses as
accumulator.
+ * Explicitly defines the intermediate result type (i.e. state entry) that
an aggregating
+ * function uses as its accumulator. The entry is managed by the framework
(usually via Flink's
+ * managed state).
*
* <p>By default, an explicit accumulator type is undefined and the
reflection-based extraction
* is used.
+ *
+ * <p>This parameter is primarily intended for aggregating functions (i.e.
{@link
+ * AggregateFunction} and {@link TableAggregateFunction}). It is
recommended to use {@link
+ * #state()} for {@link ProcessTableFunction}.
*/
DataTypeHint accumulator() default @DataTypeHint();
+ /**
+ * Explicitly lists the intermediate results (i.e. state entries) of a
function that is managed
+ * by the framework (i.e. Flink managed state). Including their names and
data types.
+ *
+ * <p>State hints are primarily intended for {@link ProcessTableFunction}.
A PTF supports
+ * multiple state entries at the beginning of an eval()/onTimer() method
(after an optional
+ * context parameter).
+ *
+ * <p>Aggregating functions (i.e. {@link AggregateFunction} and {@link
TableAggregateFunction})
+ * support a single state entry at the beginning of an
accumulate()/retract() method (i.e. the
+ * accumulator).
+ *
+ * <p>By default, explicit state is undefined and the reflection-based
extraction is used where
+ * {@link StateHint} is present.
+ *
+ * <p>Using both {@link #accumulator()} and this parameter is not allowed.
Specifying the list
+ * of state entries manually disables the entire reflection-based
extraction around {@link
+ * StateHint} and accumulators for aggregating functions.
+ */
+ StateHint[] state() default {};
+
/**
* Explicitly defines the result type that a function uses as output.
*
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentHint.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/StateHint.java
similarity index 53%
copy from
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentHint.java
copy to
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/StateHint.java
index 87a438c6eb5..fdab11790ac 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/ArgumentHint.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/StateHint.java
@@ -19,6 +19,9 @@
package org.apache.flink.table.annotation;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.TableAggregateFunction;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
@@ -26,46 +29,41 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
- * A hint that provides additional information about an argument.
+ * A hint that declares an intermediate result (i.e. state entry) that is
managed by the framework
+ * (i.e. Flink managed state).
*
- * <p>An {@code ArgumentHint} can be used to provide hints about the name,
optionality, and data
- * type of argument.
+ * <p>State hints are primarily intended for {@link ProcessTableFunction}. A
PTF supports multiple
+ * state entries at the beginning of an eval()/onTimer() method (after an
optional context
+ * parameter).
*
- * <p>{@code @ArgumentHint(name = "in1", type = @DataTypeHint("STRING"),
isOptional = false)} is a
- * scalar argument with the data type STRING, named "in1", and cannot be
omitted when calling.
+ * <p>Aggregating functions (i.e. {@link AggregateFunction} and {@link
TableAggregateFunction})
+ * support a single state entry at the beginning of an accumulate()/retract()
method (i.e. the
+ * accumulator).
+ *
+ * <p>For example, {@code @StateHint(name = "count", type =
@DataTypeHint("BIGINT"))} is a state
+ * entry with the data type BIGINT named "count".
+ *
+ * <p>Note: Usually, a state entry is partitioned by a key and can not be
accessed globally. The
+ * partitioning (or whether it is only a single partition) is defined by the
corresponding function
+ * call.
*
* @see FunctionHint
*/
@PublicEvolving
@Retention(RetentionPolicy.RUNTIME)
-@Target({ElementType.TYPE, ElementType.METHOD, ElementType.FIELD,
ElementType.PARAMETER})
-public @interface ArgumentHint {
-
- /**
- * The kind of the argument.
- *
- * <p>Only applies to {@code ProcessTableFunction}s (PTFs). Others can
only take scalar values.
- */
- ArgumentTrait[] value() default {ArgumentTrait.SCALAR};
+@Target({ElementType.TYPE, ElementType.METHOD, ElementType.PARAMETER})
+public @interface StateHint {
/**
- * The name of the argument.
+ * The name of the state entry. It must be unique among other state
entries.
*
- * <p>This can be used to provide a descriptive name for the argument.
+ * <p>This can be used to provide a descriptive name for the state entry.
The name can be used
+ * for referencing the entry during clean up.
*/
String name() default "";
/**
- * Specifies whether the argument is optional or required.
- *
- * <p>If set to {@code true}, the argument is considered optional.And if
the user does not
- * specify this parameter when calling, 'null' will be passed in. By
default, an argument is
- * considered required.
- */
- boolean isOptional() default false;
-
- /**
- * The data type hint for the argument.
+ * The data type hint for the state entry.
*
* <p>This can be used to provide additional information about the
expected data type of the
* argument. The {@link DataTypeHint} annotation can be used to specify
the data type explicitly
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
index 025a4fee026..1ba64b2ca5c 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/FunctionKind.java
@@ -35,5 +35,7 @@ public enum FunctionKind {
TABLE_AGGREGATE,
+ PROCESS_TABLE,
+
OTHER
}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
new file mode 100644
index 00000000000..1e2e215dca8
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java
@@ -0,0 +1,246 @@
+/*
+ * 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.table.functions;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.ArgumentTrait;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.FunctionHint;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.flink.util.Collector;
+
+/**
+ * Base class for a user-defined process table function. A process table
function (PTF) maps zero,
+ * one, or multiple tables to zero, one, or multiple rows (or structured
types). Scalar arguments
+ * are also supported. If the output record consists of only one field, the
wrapper can be omitted,
+ * and a scalar value can be emitted that will be implicitly wrapped into a
row by the runtime.
+ *
+ * <p>PTFs are the most powerful function kind for Flink SQL and Table API.
They enable implementing
+ * user-defined operators that can be as feature-rich as built-in operations.
PTFs have access to
+ * Flink's managed state, event-time and timer services, underlying table
changelogs, and can take
+ * multiple ordered and/or partitioned tables to produce a new table.
+ *
+ * <h1>Table Semantics and Virtual Processors</h1>
+ *
+ * <p>PTFs can produce a new table by consuming tables as arguments. For
scalability, input tables
+ * are distributed across so-called "virtual processors". A virtual processor,
as defined by the SQL
+ * standard, executes a PTF instance and has access only to a portion of the
entire table. The
+ * argument declaration decides about the size of the portion and co-location
of data. Conceptually,
+ * tables can be processed either "as row" (i.e. with row semantics) or "as
set" (i.e. with set
+ * semantics).
+ *
+ * <h2>Table Argument with Row Semantics</h2>
+ *
+ * <p>A PTF that takes a table with row semantics assumes that there is no
correlation between rows
+ * and each row can be processed independently. The framework is free in how
to distribute rows
+ * across virtual processors and each virtual processor has access only to the
currently processed
+ * row.
+ *
+ * <h2>Table Argument with Set Semantics</h2>
+ *
+ * <p>A PTF that takes a table with set semantics assumes that there is a
correlation between rows.
+ * When calling the function, the PARTITION BY clause defines the columns for
correlation. The
+ * framework ensures that all rows belonging to same set are co-located. A PTF
instance is able to
+ * access all rows belonging to the same set. In other words: The virtual
processor is scoped by a
+ * key context.
+ *
+ * <p>It is also possible not to provide a key ({@link
ArgumentTrait#OPTIONAL_PARTITION_BY}), in
+ * which case only one virtual processor handles the entire table, thereby
losing scalability
+ * benefits.
+ *
+ * <h1>Implementation</h1>
+ *
+ * <p>The behavior of a {@link ProcessTableFunction} can be defined by
implementing a custom
+ * evaluation method. The evaluation method must be declared publicly, not
static, and named <code>
+ * eval</code>. Overloading is not supported.
+ *
+ * <p>For storing a user-defined function in a catalog, the class must have a
default constructor
+ * and must be instantiable during runtime. Anonymous functions in Table API
can only be persisted
+ * if the function object is not stateful (i.e. containing only transient and
static fields).
+ *
+ * <h2>Data Types</h2>
+ *
+ * <p>By default, input and output data types are automatically extracted
using reflection. This
+ * includes the generic argument {@code T} of the class for determining an
output data type. Input
+ * arguments are derived from the {@code eval()} method. If the reflective
information is not
+ * sufficient, it can be supported and enriched with {@link FunctionHint},
{@link ArgumentHint}, and
+ * {@link DataTypeHint} annotations.
+ *
+ * <p>The following examples show how to specify data types:
+ *
+ * <pre>{@code
+ * // Function that accepts two scalar INT arguments and emits them as an
implicit ROW < INT >
+ * class AdditionFunction extends ProcessTableFunction<Integer> {
+ * public void eval(Integer a, Integer b) {
+ * collect(a + b);
+ * }
+ * }
+ *
+ * // Function that produces an explicit ROW < i INT, s STRING > from
arguments, the function hint helps in
+ * // declaring the row's fields
+ * @FunctionHint(output = @DataTypeHint("ROW< i INT, s STRING >"))
+ * class DuplicatorFunction extends ProcessTableFunction<Row> {
+ * public void eval(Integer i, String s) {
+ * collect(Row.of(i, s));
+ * collect(Row.of(i, s));
+ * }
+ * }
+ *
+ * // Function that accepts DECIMAL(10, 4) and emits it as an explicit ROW <
DECIMAL(10, 4) >
+ * @FunctionHint(output = @DataTypeHint("ROW< DECIMAL(10, 4) >"))
+ * class DuplicatorFunction extends TableFunction<Row> {
+ * public void eval(@DataTypeHint("DECIMAL(10, 4)") BigDecimal d) {
+ * collect(Row.of(d));
+ * collect(Row.of(d));
+ * }
+ * }
+ * }</pre>
+ *
+ * <h2>Arguments</h2>
+ *
+ * <p>The {@link ArgumentHint} annotation enables declaring the name, data
type, and kind of each
+ * argument (i.e. ArgumentTrait.SCALAR, ArgumentTrait.TABLE_AS_SET, or
ArgumentTrait.TABLE_AS_ROW).
+ * It allows specifying other traits for table arguments as well:
+ *
+ * <pre>{@code
+ * // Function that has two arguments:
+ * // "input_table" (a table with set semantics) and "threshold" (a scalar
value)
+ * class ThresholdFunction extends ProcessTableFunction<Integer> {
+ * public void eval(
+ * // For table arguments, a data type for Row is optional (leading to
polymorphic behavior)
+ * @ArgumentHint(value = ArgumentTrait.TABLE_AS_SET, name =
"input_table") Row t,
+ * // Scalar arguments require a data type either explicit or via
reflection
+ * @ArgumentHint(value = ArgumentTrait.SCALAR, name = "threshold")
Integer threshold) {
+ * int amount = t.getFieldAs("amount");
+ * if (amount >= threshold) {
+ * collect(amount);
+ * }
+ * }
+ * }
+ * }</pre>
+ *
+ * <p>Table arguments can declare a concrete data type (of either row or
structured type) or accept
+ * any type of row in polymorphic fashion:
+ *
+ * <pre>{@code
+ * // Function with explicit table argument type of row
+ * class MyPTF extends ProcessTableFunction<String> {
+ * public void eval(Context ctx, @ArgumentHint(value =
ArgumentTrait.TABLE_AS_SET, type = "ROW < s STRING >") Row t) {
+ * TableSemantics semantics = ctx.tableSemanticsFor("t");
+ * // Always returns "ROW < s STRING >"
+ * semantics.dataType();
+ * ...
+ * }
+ * }
+ *
+ * // Function with explicit table argument type of structured type "Customer"
+ * class MyPTF extends ProcessTableFunction<String> {
+ * public void eval(Context ctx, @ArgumentHint(value =
ArgumentTrait.TABLE_AS_SET) Customer c) {
+ * TableSemantics semantics = ctx.tableSemanticsFor("c");
+ * // Always returns structured type of "Customer"
+ * semantics.dataType();
+ * ...
+ * }
+ * }
+ *
+ * // Function with polymorphic table argument
+ * class MyPTF extends ProcessTableFunction<String> {
+ * public void eval(Context ctx, @ArgumentHint(value =
ArgumentTrait.TABLE_AS_SET) Row t) {
+ * TableSemantics semantics = ctx.tableSemanticsFor("t");
+ * // Always returns "ROW" but content depends on the table that is passed
into the call
+ * semantics.dataType();
+ * ...
+ * }
+ * }
+ * }</pre>
+ *
+ * <h2>Context</h2>
+ *
+ * <p>A {@link Context} can be added as a first argument to the eval() method
for additional
+ * information about the input tables and other services provided by the
framework:
+ *
+ * <pre>{@code
+ * // a function that accesses the Context for reading the PARTITION BY
columns and
+ * // excluding them when building a result string
+ * class ConcatNonKeysFunction extends ProcessTableFunction<String> {
+ * public void eval(Context ctx, @ArgumentHint(ArgumentTrait.TABLE_AS_SET)
Row inputTable) {
+ * TableSemantics semantics = ctx.tableSemanticsFor("inputTable");
+ * List<Integer> keys = Arrays.asList(semantics.partitionByColumns());
+ * return IntStream.range(0, inputTable.getArity())
+ * .filter(pos -> !keys.contains(pos))
+ * .mapToObj(inputTable::getField)
+ * .map(Object::toString)
+ * .collect(Collectors.joining(", "));
+ * }
+ * }
+ * }</pre>
+ *
+ * @param <T> The type of the output row. Either an explicit composite type or
an atomic type that
+ * is implicitly wrapped into a row consisting of one field.
+ */
+@PublicEvolving
+public abstract class ProcessTableFunction<T> extends UserDefinedFunction {
+
+ /** The code generated collector used to emit rows. */
+ private transient Collector<T> collector;
+
+ /** Internal use. Sets the current collector. */
+ public final void setCollector(Collector<T> collector) {
+ this.collector = collector;
+ }
+
+ /**
+ * Emits an (implicit or explicit) output row.
+ *
+ * <p>If null is emitted as an explicit row, it will be skipped by the
runtime. For implicit
+ * rows, the row's field will be null.
+ *
+ * @param row the output row
+ */
+ protected final void collect(T row) {
+ collector.collect(row);
+ }
+
+ @Override
+ public final FunctionKind getKind() {
+ return FunctionKind.PROCESS_TABLE;
+ }
+
+ @Override
+ public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+ throw new UnsupportedOperationException("Type inference is not
implemented yet.");
+ }
+
+ /**
+ * Context that can be added as a first argument to the eval() method for
additional information
+ * about the input tables and other services provided by the framework.
+ */
+ @PublicEvolving
+ public interface Context {
+
+ /**
+ * Returns additional information about the semantics of a table
argument.
+ *
+ * @param argName name of the table argument
+ */
+ TableSemantics tableSemanticsFor(String argName);
+ }
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunction.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunction.java
index d5eceafe2f5..d554734b6e8 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunction.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunction.java
@@ -32,9 +32,9 @@ import org.apache.flink.util.Collector;
/**
* Base class for a user-defined table function. A user-defined table function
maps zero, one, or
- * multiple scalar values to zero, one, or multiple rows (or structured
types). If an output record
- * consists of only one field, the structured record can be omitted, and a
scalar value can be
- * emitted that will be implicitly wrapped into a row by the runtime.
+ * multiple scalar values to zero, one, or multiple rows (or structured
types). If the output record
+ * consists of only one field, the wrapper can be omitted, and a scalar value
can be emitted that
+ * will be implicitly wrapped into a row by the runtime.
*
* <p>The behavior of a {@link TableFunction} can be defined by implementing a
custom evaluation
* method. An evaluation method must be declared publicly, not static, and
named <code>eval</code>.
@@ -50,7 +50,7 @@ import org.apache.flink.util.Collector;
* <p>The following examples show how to specify a table function:
*
* <pre>{@code
- * // a function that accepts an arbitrary number of INT arguments and emits
them as implicit ROW < INT >
+ * // Function that accepts an arbitrary number of INT arguments and emits
them as implicit ROW < INT >
* class FlattenFunction extends TableFunction<Integer> {
* public void eval(Integer... args) {
* for (Integer i : args) {
@@ -59,7 +59,7 @@ import org.apache.flink.util.Collector;
* }
* }
*
- * // a function that accepts either INT or STRING and emits them as implicit
ROW < STRING >
+ * // Function that accepts either INT or STRING and emits them as an implicit
ROW < STRING >
* class DuplicatorFunction extends TableFunction<String> {
* public void eval(Integer i) {
* eval(String.valueOf(i));
@@ -70,7 +70,7 @@ import org.apache.flink.util.Collector;
* }
* }
*
- * // a function that produces a ROW < i INT, s STRING > from arguments, the
function hint helps in
+ * // Function that produces an explicit ROW < i INT, s STRING > from
arguments, the function hint helps in
* // declaring the row's fields
* @FunctionHint(output = @DataTypeHint("ROW< i INT, s STRING >"))
* class DuplicatorFunction extends TableFunction<Row> {
@@ -80,7 +80,7 @@ import org.apache.flink.util.Collector;
* }
* }
*
- * // a function that accepts either INT or DECIMAL(10, 4) and emits them as
implicit ROW < INT > or
+ * // Function that accepts either INT or DECIMAL(10, 4) and emits them as an
implicit ROW < INT > or
* // ROW<DECIMAL(10, 4)> using function hints for declaring the output type
* class DuplicatorFunction extends TableFunction<Object> {
* @FunctionHint(output = @DataTypeHint("INT"))
@@ -98,7 +98,7 @@ import org.apache.flink.util.Collector;
*
* <p>For storing a user-defined function in a catalog, the class must have a
default constructor
* and must be instantiable during runtime. Anonymous functions in Table API
can only be persisted
- * if the function is not stateful (i.e. containing only transient and static
fields).
+ * if the function object is not stateful (i.e. containing only transient and
static fields).
*
* <p>In the API, a table function can be used as follows:
*
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableSemantics.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableSemantics.java
new file mode 100644
index 00000000000..0e83071ad3c
--- /dev/null
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableSemantics.java
@@ -0,0 +1,128 @@
+/*
+ * 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.table.functions;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.annotation.ArgumentTrait;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.types.DataType;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Provides call information about the table that has been passed to a table
argument.
+ *
+ * <p>This class is only available for table arguments (i.e. arguments of a
{@link
+ * ProcessTableFunction} that are annotated with {@code
@ArgumentHint(TABLE_AS_SET)} or
+ * {@code @ArgumentHint(TABLE_AS_ROW)}).
+ */
+@PublicEvolving
+public interface TableSemantics {
+
+ /**
+ * Data type of the passed table.
+ *
+ * <p>The returned data type might be the one that has been explicitly
defined for the argument
+ * or a {@link DataTypes#ROW} data type for polymorphic arguments that
accept any type of row.
+ *
+ * <p>For example:
+ *
+ * <pre>{@code
+ * // Function with explicit table argument type of row
+ * class MyPTF extends ProcessTableFunction<String> {
+ * public void eval(Context ctx, @ArgumentHint(value =
ArgumentTrait.TABLE_AS_SET, type = "ROW < s STRING >") Row t) {
+ * TableSemantics semantics = ctx.tableSemanticsFor("t");
+ * // Always returns "ROW < s STRING >"
+ * semantics.dataType();
+ * ...
+ * }
+ * }
+ *
+ * // Function with explicit table argument type of structured type
"Customer"
+ * class MyPTF extends ProcessTableFunction<String> {
+ * public void eval(Context ctx, @ArgumentHint(value =
ArgumentTrait.TABLE_AS_SET) Customer c) {
+ * TableSemantics semantics = ctx.tableSemanticsFor("c");
+ * // Always returns structured type of "Customer"
+ * semantics.dataType();
+ * ...
+ * }
+ * }
+ *
+ * // Function with polymorphic table argument
+ * class MyPTF extends ProcessTableFunction<String> {
+ * public void eval(Context ctx, @ArgumentHint(value =
ArgumentTrait.TABLE_AS_SET) Row t) {
+ * TableSemantics semantics = ctx.tableSemanticsFor("t");
+ * // Always returns "ROW" but content depends on the table that is
passed into the call
+ * semantics.dataType();
+ * ...
+ * }
+ * }
+ * }</pre>
+ */
+ DataType dataType();
+
+ /**
+ * Returns information about how the passed table is partitioned. Applies
only to table
+ * arguments with set semantics.
+ *
+ * @return An array of indexes (0-based) that specify the PARTITION BY
columns.
+ */
+ int[] partitionByColumns();
+
+ /**
+ * Returns information about how the passed table is ordered. Applies only
to table arguments
+ * with set semantics.
+ *
+ * @return An array of indexes (0-based) that specify the ORDER BY columns.
+ */
+ int[] orderByColumns();
+
+ /**
+ * Returns information about the time attribute of the passed table. The
time attribute column
+ * powers the concept of rowtime and timers. Applies to both table
arguments with row and set
+ * semantics.
+ *
+ * @return Position of the "ON_TIME" column. Returns -1 in case no time
attribute has been
+ * passed.
+ */
+ int timeColumn();
+
+ /**
+ * Returns information about which passed tables are co-partitioned with
the passed table.
+ * Applies only to table arguments with set semantics.
+ *
+ * @return List of table argument names (not table names!) that are
co-partitioned with the
+ * passed table.
+ */
+ List<String> coPartitionArgs();
+
+ /**
+ * Actual changelog mode for the passed table. By default, table arguments
take only {@link
+ * ChangelogMode#insertOnly()}. They are able to take tables of other
changelog modes, if
+ * specified to do so (e.g. via an {@link ArgumentTrait}). This method
returns the final
+ * changelog mode determined by the planner.
+ *
+ * @return The definitive changelog mode expected for the passed table
after physical
+ * optimization. Returns an actual value when called during runtime.
Returns empty during
+ * type inference phase as the changelog mode is still unknown.
+ */
+ Optional<ChangelogMode> changelogMode();
+}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/BaseMappingExtractor.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/BaseMappingExtractor.java
index 80d9546c203..8e3aab464e9 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/BaseMappingExtractor.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/BaseMappingExtractor.java
@@ -469,7 +469,8 @@ abstract class BaseMappingExtractor {
}
final ArgumentTrait[] traits = h.value();
if (traits.length != 1 || traits[0] !=
ArgumentTrait.SCALAR) {
- throw extractionError("Only scalar arguments
are supported yet.");
+ throw extractionError(
+ "Only scalar arguments are supported
so far.");
}
return h.isOptional();
})
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionTemplate.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionTemplate.java
index a35c83d45b7..c92830393cc 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionTemplate.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/extraction/FunctionTemplate.java
@@ -65,6 +65,9 @@ final class FunctionTemplate {
* types.
*/
static FunctionTemplate fromAnnotation(DataTypeFactory typeFactory,
FunctionHint hint) {
+ if (hint.state().length > 0) {
+ throw extractionError("State hints are not supported yet.");
+ }
return new FunctionTemplate(
createSignatureTemplate(
typeFactory,
@@ -221,7 +224,7 @@ final class FunctionTemplate {
&& traits[0] ==
ArgumentTrait.SCALAR;
});
if (!allScalar) {
- throw extractionError("Only scalar arguments are supported
yet.");
+ throw extractionError("Only scalar arguments are supported so
far.");
}
argumentHintNames = new String[argumentHints.length];
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
index 0491801265e..5fbc6dc81d3 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.annotation.StateHint;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.data.RowData;
@@ -630,9 +631,11 @@ class TypeInferenceExtractorTest {
}),
TypeStrategies.explicit(DataTypes.STRING())),
TestSpec.forScalarFunction(FunctionHintTableArgScalarFunction.class)
- .expectErrorMessage("Only scalar arguments are
supported yet."),
+ .expectErrorMessage("Only scalar arguments are
supported so far."),
TestSpec.forScalarFunction(ArgumentHintTableArgScalarFunction.class)
- .expectErrorMessage("Only scalar arguments are
supported yet."));
+ .expectErrorMessage("Only scalar arguments are
supported so far."),
+ TestSpec.forScalarFunction(StateHintScalarFunction.class)
+ .expectErrorMessage("State hints are not supported
yet."));
}
private static Stream<TestSpec> procedureSpecs() {
@@ -2148,4 +2151,11 @@ class TypeInferenceExtractorTest {
return "";
}
}
+
+ @FunctionHint(state = @StateHint(name = "state", type =
@DataTypeHint("INT")))
+ private static class StateHintScalarFunction extends ScalarFunction {
+ public String eval() {
+ return "";
+ }
+ }
}
