You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "twalthr (via GitHub)" <gi...@apache.org> on 2024/11/28 09:16:04 UTC
[PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
twalthr opened a new pull request, #25707:
URL: https://github.com/apache/flink/pull/25707
## What is the purpose of the change
Adds the initial interfaces of `ProcessTableFunction` and `StateHint` mentioned in FLIP-440 to the code base. This is a requirement to continue with the `TypeInferenceExtractor` extractor work.
## Brief change log
- Introduce `ProcessTableFunction` and `StateHint`
- Add `FunctionHint.state()`
- Add `TableSementics` for powering the context of PTFs
## Verifying this change
Please make sure both new and modified tests in this PR follow [the conventions for tests defined in our code quality guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing).
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on PR #25707:
URL: https://github.com/apache/flink/pull/25707#issuecomment-2506035053
> ideally testing the behavior documented in the comments.
This is a feature under development. The behavior in comments is not fully implemented yet. As can be seen in the umbrella ticket FLINK-36703, this is just the very beginning of the implementation. The interfaces are necessary as a first step to power the next tasks which is TypeInferenceExtractor update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1861980546
##########
flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/extraction/TypeInferenceExtractorTest.java:
##########
@@ -632,7 +633,9 @@ private static Stream<TestSpec> functionSpecs() {
TestSpec.forScalarFunction(FunctionHintTableArgScalarFunction.class)
.expectErrorMessage("Only scalar arguments are supported yet."),
TestSpec.forScalarFunction(ArgumentHintTableArgScalarFunction.class)
- .expectErrorMessage("Only scalar arguments are supported yet."));
+ .expectErrorMessage("Only scalar arguments are supported yet."),
Review Comment:
You don't change the message, but I spotted the sentence is not grammatically correct. Do you think we could fix it?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/StateHint.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.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;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * A hint that declares an intermediate result (i.e. state entry) that is managed by the framework
+ * (i.e. Flink managed state).
+ *
+ * <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>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.PARAMETER})
+public @interface StateHint {
+
+ /**
+ * The name of the state entry.
+ *
+ * <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 "";
Review Comment:
Are there any uniqueness requirements? If so can we write them explicitly?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.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 into virtual processors. Each virtual processor executes a PTF instance and has
Review Comment:
Just writing out my thoughts...
Is it a good idea to introduce a new term here? What is a "virtual processor"? How does it relate to Flink's TMs and/or Tasks? Maybe I am biased, but I find it raises more questions then provides answers.
##########
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 if only available for table arguments (i.e. arguments of a {@link
Review Comment:
```suggestion
* <p>This class is only available for table arguments (i.e. arguments of a {@link
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862076676
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.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 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. 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
+ * among 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 under
+ * a key context.
+ *
+ * <h1>Basic 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.
Review Comment:
I wonder why Overloading is not supported. Is there some relationship to the way the parameters are supplied from SQL that means they need to be one shape, so overloading would not work?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862101386
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.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 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. 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
+ * among 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 under
+ * a key context.
+ *
+ * <h1>Basic 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.
Review Comment:
Overloading does not go well with named parameters. In the end we want to support a syntax like:
```
SELECT * FROM f(name => 's', 'threshold' => 12)
```
Supporting `name == STRING || name == INT` makes the implementation highly complex. We support optional arguments, which should solve most of the use cases. E.g. `f(numbers = [123])` or `f(strings = ['1', '2', '3'])`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on PR #25707:
URL: https://github.com/apache/flink/pull/25707#issuecomment-2506316008
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1861852208
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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.
Review Comment:
Should it also be mentioned in doc? Somewhere here e.g.
https://github.com/apache/flink/blob/b9c92371dba07b6bfe4368d7b6d7f7c575b4c603/docs/content/docs/dev/table/functions/udfs.md?plain=1#L439
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862058735
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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).
Review Comment:
it would be worth saying how this is managed by the frame =work if not using managed state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr merged PR #25707:
URL: https://github.com/apache/flink/pull/25707
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862211415
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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.
Review Comment:
Let wait with this. Currently, PTFs are not fully functional. Exposing the changes via docs might cause confusion.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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.
Review Comment:
Let's wait with this. Currently, PTFs are not fully functional. Exposing the changes via docs might cause confusion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862114539
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.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 into virtual processors. Each virtual processor executes a PTF instance and has
Review Comment:
A virtual processor could be translated into the Flink concept "key context" but this is rather an internal term. Virtual processor is neither TM, nor Task, nor Subtask. And I think this is also what confuses people when they work with KeyedProcessFunction for the first time. Understanding that state is scoped under the key.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862103659
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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).
Review Comment:
This is hard to say and might be too implementation specific. It varies from operator to operator. Sometimes its MemorySegments, sometimes its just Java heap.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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).
Review Comment:
This is hard to say and might be too implementation specific. It varies from operator to operator. Sometimes its MemorySegments, sometimes it's just Java heap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25707:
URL: https://github.com/apache/flink/pull/25707#issuecomment-2505633651
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e8c2ec88bf59239f3c40a88d228905772eab9459",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e8c2ec88bf59239f3c40a88d228905772eab9459",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e8c2ec88bf59239f3c40a88d228905772eab9459 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862101386
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ProcessTableFunction.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.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 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. 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
+ * among 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 under
+ * a key context.
+ *
+ * <h1>Basic 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.
Review Comment:
Overloading does not go well with named parameters. In the end we want to support a syntax like:
```
SELECT * FROM f(name => 's', 'threshold' => 12)
```
Supporting `name == STRING || name == INT` makes the implementation highly complex. We support optional arguments, which should solve most of the use cases. E.g. `f(numbers = [1, 2, 3])` or `f(strings = ['1', '2', '3'])`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862082959
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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).
Review Comment:
if would be worth saying how the entry is managed when not via managed state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862212335
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/StateHint.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.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;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * A hint that declares an intermediate result (i.e. state entry) that is managed by the framework
+ * (i.e. Flink managed state).
+ *
+ * <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>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.PARAMETER})
+public @interface StateHint {
Review Comment:
Absolutely, but not yet. First there needs to be more logic merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1861853852
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/StateHint.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.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;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * A hint that declares an intermediate result (i.e. state entry) that is managed by the framework
+ * (i.e. Flink managed state).
+ *
+ * <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>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.PARAMETER})
+public @interface StateHint {
Review Comment:
Also a question: whether it should be covered in doc as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862058735
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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).
Review Comment:
for clarity - it would be worth saying how this is managed by the frame work if not using managed state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36705] Add initial ProcessTableFunction class and annotations [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25707:
URL: https://github.com/apache/flink/pull/25707#discussion_r1862058735
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/annotation/FunctionHint.java:
##########
@@ -175,13 +178,40 @@
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).
Review Comment:
for clarity - it would be worth saying how this is managed by the frame work if not using managed state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org