You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "gustavodemorais (via GitHub)" <gi...@apache.org> on 2026/03/16 16:17:46 UTC
[PR] [FLINK-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
gustavodemorais opened a new pull request, #27777:
URL: https://github.com/apache/flink/pull/27777
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This is the first implementation of TO_CHANGELOG. The idea is to implement the use cases one by one interactively. With this ticket we will write the first version which allows flat mapping of **upsert stream**.
Implement TO_CHANGELOG built-in PTF as specified in FLIP-564, section 4.2.2.2 (retract/upsert stream to upsert stream, flat mode).
TO_CHANGELOG converts a dynamic table (retract/upsert stream) into an append-only stream with an explicit operation code column. Each input row is emitted as an INSERT-only row with a string op column indicating the original RowKind. This is stateless - each event maps directly to one output record.
## Brief change log
- Built-in function definition and type strategies
- Runtime implementation
- Table API interface
- MAP literal support in OperatorBindingCallContext and CallBindingCallContext
- Plan tests, semantic tests (SQL, Table API, convenience API)
- Use-case test: LAG on upsert stream (previously impossible without TO_CHANGELOG, like other aggregation functions)
- Documentation
## Verifying this change
- Plan tests
- Semantic tests
## 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)`: (no)
- 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? (docs / 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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #27777:
URL: https://github.com/apache/flink/pull/27777#issuecomment-4097131602
Thanks for the high quality but quick reviews, @twalthr!
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2941901097
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java:
##########
@@ -901,6 +901,11 @@ public Table process(Class<? extends UserDefinedFunction> function, Object... ar
createPartitionQueryOperation(), table.tableEnvironment, arguments));
}
+ @Override
+ public Table toChangelog() {
+ return process("TO_CHANGELOG");
Review Comment:
don't hard code the name here. use the built in function definition. I know this was an issue for ml predict, but we should fix it properly now.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -138,4 +139,26 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this dynamic table into an append-only stream with an explicit operation code column
+ * using the built-in {@link BuiltInFunctionDefinitions#TO_CHANGELOG TO_CHANGELOG} PTF.
Review Comment:
Remove references to BuiltInFunctionDefinitions#TO_CHANGELOG TO_CHANGELOG. BuiltInFunctionDefinitions is rather internal.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableImpl.java:
##########
@@ -901,6 +901,11 @@ public Table process(Class<? extends UserDefinedFunction> function, Object... ar
createPartitionQueryOperation(), table.tableEnvironment, arguments));
}
+ @Override
+ public Table toChangelog() {
Review Comment:
add op_mapping parameter and op, same functionality as in sql. to avoid to many overloads, we can accept Expression and force people to use `.asArgument`.
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #27777:
URL: https://github.com/apache/flink/pull/27777#issuecomment-4089478486
Hi @twalthr, thanks for the review. I've addressed your comments, take a look
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2958353702
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.planner.plan.stream.sql;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.planner.utils.TableTestUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the TO_CHANGELOG built-in process table function. */
+public class ToChangelogTest extends TableTestBase {
Review Comment:
> Testing argument validation errors feels correct here.
This we can also do with semantic tests nowadays.
> show the changelog mode coming out of each node
This is a valid argument for me. In general, my goal is to keep the code base lean while having sufficient test coverage. Testing the same via different test bases should be avoided as much as possible. Otherwise people (and in the future AI) will add tests to multiple classes, instead to semantic tests only.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java:
##########
@@ -18,66 +18,94 @@
package org.apache.flink.table.planner.plan.stream.sql;
+import org.apache.flink.table.api.ExplainDetail;
import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.flink.table.planner.utils.TableTestBase;
import org.apache.flink.table.planner.utils.TableTestUtil;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Tests for the TO_CHANGELOG built-in process table function. */
+/**
+ * Plan tests for the TO_CHANGELOG built-in process table function. Uses {@link
+ * ExplainDetail#CHANGELOG_MODE} to verify changelog mode propagation through the plan.
+ */
public class ToChangelogTest extends TableTestBase {
private TableTestUtil util;
@BeforeEach
void setup() {
util = streamTestUtil(TableConfig.getDefault());
+ }
+
+ @Test
+ void testRetractSource() {
util.tableEnv()
.executeSql(
- "CREATE TABLE t ("
+ "CREATE TABLE retract_source ("
+ " id INT,"
+ " name STRING,"
- + " val BIGINT"
- + ") WITH ('connector' = 'values')");
- util.tableEnv()
- .executeSql(
- "CREATE VIEW t_updating AS SELECT id, name, COUNT(*) AS cnt FROM t GROUP BY id, name");
- }
-
- @Test
- void testDefaultCall() {
- util.verifyRelPlan("SELECT * FROM TO_CHANGELOG(input => TABLE t_updating PARTITION BY id)");
- }
-
- @Test
- void testCustomOpName() {
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'changelog-mode' = 'I,UB,UA,D'"
+ + ")");
util.verifyRelPlan(
- "SELECT * FROM TO_CHANGELOG("
- + "input => TABLE t_updating PARTITION BY id, "
- + "op => DESCRIPTOR(op_code))");
+ "SELECT * FROM TO_CHANGELOG(input => TABLE retract_source PARTITION BY id)",
+ JavaScalaConversionUtil.toScala(
Review Comment:
When a Java class has to do things like conversion. Always feel free to update the base, to not require this anymore by overloading.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -108,18 +121,13 @@ public void eval(
final RowData input,
@Nullable final ColumnList op,
@Nullable final Map<String, String> opMapping) {
Review Comment:
Also this should be MapData. Ideally our built in function do not need any conversion of data structures. Otherwise the runtime needs to convert on every invoke.
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27777:
URL: https://github.com/apache/flink/pull/27777#issuecomment-4068930672
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "06ac092063c7a5bc5e004f87a96b5a0e59d973cb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "06ac092063c7a5bc5e004f87a96b5a0e59d973cb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 06ac092063c7a5bc5e004f87a96b5a0e59d973cb 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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2959452050
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.planner.plan.stream.sql;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.planner.utils.TableTestUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the TO_CHANGELOG built-in process table function. */
+public class ToChangelogTest extends TableTestBase {
Review Comment:
I like the ideas of having the tests in one place and like how they look like in semantic tests so moved the validation tests and only kept the base changelog tests 👍
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2959437856
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -108,18 +121,13 @@ public void eval(
final RowData input,
@Nullable final ColumnList op,
@Nullable final Map<String, String> opMapping) {
Review Comment:
Sorry, that makes sense. I was thinking about the values themselves (constants vs changing data), but the conversion overhead comes from the generated code, which doesn't distinguish between constant and per-row arguments. It applies the same type conversion logic to every argument on every invoke.
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2960439903
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.runtime.functions.ptf;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.data.utils.ProjectedRowData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.ColumnList;
+import org.apache.flink.types.RowKind;
+
+import org.apache.flink.table.functions.FunctionContext;
+
+import javax.annotation.Nullable;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Runtime implementation of {@link BuiltInFunctionDefinitions#TO_CHANGELOG}.
+ *
+ * <p>Converts each input row into an INSERT-only output row with an operation code column. The
+ * output schema is {@code [op_column, ...non_partition_columns...]} - the framework prepends
+ * partition key columns automatically.
+ *
+ * <p>Uses {@link ProjectedRowData} for zero-copy projection of non-partition columns and {@link
+ * JoinedRowData} to combine the op column with the projected input.
+ */
+@Internal
+public class ToChangelogFunction extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Map<RowKind, String> DEFAULT_OP_MAPPING =
+ Map.of(
+ RowKind.INSERT, "INSERT",
+ RowKind.UPDATE_BEFORE, "UPDATE_BEFORE",
+ RowKind.UPDATE_AFTER, "UPDATE_AFTER",
+ RowKind.DELETE, "DELETE");
+
+ private final Map<RowKind, String> opMap;
+ private final int[] nonPartitionIndices;
+
+ private transient ProjectedRowData projectedInput;
+ private transient GenericRowData opRow;
+ private transient JoinedRowData output;
+
+ @SuppressWarnings("unchecked")
+ public ToChangelogFunction(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.TO_CHANGELOG, context);
+ final CallContext callContext = context.getCallContext();
+
+ final TableSemantics semantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ final int[] partitionKeys = semantics.partitionByColumns();
+ final Set<Integer> partitionKeySet =
+ IntStream.of(partitionKeys).boxed().collect(Collectors.toSet());
+
+ final RowType inputType = (RowType) semantics.dataType().getLogicalType();
+ this.nonPartitionIndices = buildNonPartitionIndices(inputType.getFieldCount(), partitionKeySet);
+
+ final Map<String, String> opMapping =
+ callContext.getArgumentValue(2, Map.class).orElse(null);
+ this.opMap = buildOpMap(opMapping);
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedInput = ProjectedRowData.from(nonPartitionIndices);
+ opRow = new GenericRowData(1);
+ output = new JoinedRowData();
+ }
+
+ private static int[] buildNonPartitionIndices(
+ final int fieldCount, final Set<Integer> partitionKeySet) {
+ return IntStream.range(0, fieldCount)
+ .filter(i -> !partitionKeySet.contains(i))
+ .toArray();
+ }
+
+ private static Map<RowKind, String> buildOpMap(@Nullable final Map<String, String> opMapping) {
+ if (opMapping == null) {
+ return new EnumMap<>(DEFAULT_OP_MAPPING);
+ }
+ final Map<RowKind, String> map = new EnumMap<>(RowKind.class);
+ opMapping.forEach((name, code) -> map.put(RowKind.valueOf(name), code));
+ return map;
+ }
+
+ public void eval(
+ final Context ctx,
+ final RowData input,
+ @Nullable final ColumnList op,
+ @Nullable final MapData opMapping) {
+ final String opCode = opMap.get(input.getRowKind());
+ if (opCode == null) {
+ return;
+ }
+
+ opRow.setField(0, StringData.fromString(opCode));
Review Comment:
here https://github.com/apache/flink/pull/27777/commits/21f967fb2e293a6b9e6fa83d70d6a88ca26085d0
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr merged PR #27777:
URL: https://github.com/apache/flink/pull/27777
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2945162191
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/ptf/ToChangelogFunction.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.planner.functions.ptf;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.flink.types.ColumnList;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import javax.annotation.Nullable;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Runtime implementation of {@link BuiltInFunctionDefinitions#TO_CHANGELOG}.
+ *
+ * <p>Converts each input row into an INSERT-only output row with an operation code column. The
+ * output schema is {@code [op_column, ...non_partition_columns...]} - the framework prepends
+ * partition key columns automatically.
+ */
+@Internal
+public class ToChangelogFunction extends ProcessTableFunction<Row> {
Review Comment:
You probably mean we should add a new BuiltInProcessTableFunction similar to BuiltInScalarFunction as start using it as a base class?
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2941889424
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/ptf/ToChangelogFunction.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.planner.functions.ptf;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.flink.types.ColumnList;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import javax.annotation.Nullable;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Runtime implementation of {@link BuiltInFunctionDefinitions#TO_CHANGELOG}.
+ *
+ * <p>Converts each input row into an INSERT-only output row with an operation code column. The
+ * output schema is {@code [op_column, ...non_partition_columns...]} - the framework prepends
+ * partition key columns automatically.
+ */
+@Internal
+public class ToChangelogFunction extends ProcessTableFunction<Row> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Map<RowKind, String> DEFAULT_OP_MAPPING =
+ Map.of(
+ RowKind.INSERT, "INSERT",
+ RowKind.UPDATE_BEFORE, "UPDATE_BEFORE",
+ RowKind.UPDATE_AFTER, "UPDATE_AFTER",
+ RowKind.DELETE, "DELETE");
+
+ private final Map<RowKind, String> opMap;
+ private final Set<Integer> partitionKeySet;
+ private final int partitionKeyCount;
+
+ @SuppressWarnings("unchecked")
+ public ToChangelogFunction(final SpecializedContext context) {
+ final CallContext callContext = context.getCallContext();
+
+ final TableSemantics semantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ final int[] partitionKeys = semantics.partitionByColumns();
+ this.partitionKeySet = IntStream.of(partitionKeys).boxed().collect(Collectors.toSet());
+ this.partitionKeyCount = partitionKeys.length;
+
+ final Map<String, String> opMapping =
+ callContext.getArgumentValue(2, Map.class).orElse(null);
+ this.opMap = buildOpMap(opMapping);
+ }
+
+ @Override
+ public TypeInference getTypeInference(final DataTypeFactory typeFactory) {
+ return BuiltInFunctionDefinitions.TO_CHANGELOG.getTypeInference(typeFactory);
+ }
+
+ public void eval(
+ final Context ctx,
+ final Row input,
+ @Nullable final ColumnList op,
+ @Nullable final Map<String, String> opMapping) {
+
+ final String opCode = opMap.get(input.getKind());
+ if (opCode == null) {
+ // TODO Gustavo Error handling still to be implemented in following ticket
+ return;
+ }
+
+ collect(buildOutputRow(input, opCode));
+ }
+
+ private Row buildOutputRow(final Row input, final String opCode) {
Review Comment:
once you use BuiltIn stack, everything should be RowData. as efficiently as possible with as little conversion as possible
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/ptf/ToChangelogFunction.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.planner.functions.ptf;
Review Comment:
put into `org.apache.flink.table.runtime.functions.ptf` in table-runtime
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/ptf/ToChangelogFunction.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.planner.functions.ptf;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.flink.types.ColumnList;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import javax.annotation.Nullable;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Runtime implementation of {@link BuiltInFunctionDefinitions#TO_CHANGELOG}.
+ *
+ * <p>Converts each input row into an INSERT-only output row with an operation code column. The
+ * output schema is {@code [op_column, ...non_partition_columns...]} - the framework prepends
+ * partition key columns automatically.
+ */
+@Internal
+public class ToChangelogFunction extends ProcessTableFunction<Row> {
Review Comment:
BuiltInProcessTableFunction similar to BuiltInScalarFunction
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #27777:
URL: https://github.com/apache/flink/pull/27777#issuecomment-4090939447
@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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #27777:
URL: https://github.com/apache/flink/pull/27777#issuecomment-4090523761
Hi @twalthr, I've addressed the last point, take a look
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2945156122
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/CallBindingCallContext.java:
##########
@@ -354,4 +361,15 @@ private static ColumnList convertColumnList(List<SqlNode> operands) {
.collect(Collectors.toList());
return ColumnList.of(names);
}
+
+ private static Map<String, String> convertMap(SqlCall mapCall) {
+ final List<SqlNode> operands = mapCall.getOperandList();
+ final Map<String, String> map = new LinkedHashMap<>();
+ for (int i = 0; i < operands.size(); i += 2) {
+ final String key = SqlLiteral.unchain(operands.get(i)).getValueAs(String.class);
Review Comment:
Let's catch exceptions here and also check for nulls
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2959814329
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.runtime.functions.ptf;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.data.utils.ProjectedRowData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.ColumnList;
+import org.apache.flink.types.RowKind;
+
+import org.apache.flink.table.functions.FunctionContext;
+
+import javax.annotation.Nullable;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Runtime implementation of {@link BuiltInFunctionDefinitions#TO_CHANGELOG}.
+ *
+ * <p>Converts each input row into an INSERT-only output row with an operation code column. The
+ * output schema is {@code [op_column, ...non_partition_columns...]} - the framework prepends
+ * partition key columns automatically.
+ *
+ * <p>Uses {@link ProjectedRowData} for zero-copy projection of non-partition columns and {@link
+ * JoinedRowData} to combine the op column with the projected input.
+ */
+@Internal
+public class ToChangelogFunction extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Map<RowKind, String> DEFAULT_OP_MAPPING =
+ Map.of(
+ RowKind.INSERT, "INSERT",
+ RowKind.UPDATE_BEFORE, "UPDATE_BEFORE",
+ RowKind.UPDATE_AFTER, "UPDATE_AFTER",
+ RowKind.DELETE, "DELETE");
+
+ private final Map<RowKind, String> opMap;
+ private final int[] nonPartitionIndices;
+
+ private transient ProjectedRowData projectedInput;
+ private transient GenericRowData opRow;
+ private transient JoinedRowData output;
+
+ @SuppressWarnings("unchecked")
+ public ToChangelogFunction(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.TO_CHANGELOG, context);
+ final CallContext callContext = context.getCallContext();
+
+ final TableSemantics semantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ final int[] partitionKeys = semantics.partitionByColumns();
+ final Set<Integer> partitionKeySet =
+ IntStream.of(partitionKeys).boxed().collect(Collectors.toSet());
+
+ final RowType inputType = (RowType) semantics.dataType().getLogicalType();
+ this.nonPartitionIndices = buildNonPartitionIndices(inputType.getFieldCount(), partitionKeySet);
+
+ final Map<String, String> opMapping =
+ callContext.getArgumentValue(2, Map.class).orElse(null);
+ this.opMap = buildOpMap(opMapping);
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedInput = ProjectedRowData.from(nonPartitionIndices);
+ opRow = new GenericRowData(1);
+ output = new JoinedRowData();
+ }
+
+ private static int[] buildNonPartitionIndices(
+ final int fieldCount, final Set<Integer> partitionKeySet) {
+ return IntStream.range(0, fieldCount)
+ .filter(i -> !partitionKeySet.contains(i))
+ .toArray();
+ }
+
+ private static Map<RowKind, String> buildOpMap(@Nullable final Map<String, String> opMapping) {
+ if (opMapping == null) {
+ return new EnumMap<>(DEFAULT_OP_MAPPING);
+ }
+ final Map<RowKind, String> map = new EnumMap<>(RowKind.class);
+ opMapping.forEach((name, code) -> map.put(RowKind.valueOf(name), code));
+ return map;
+ }
+
+ public void eval(
+ final Context ctx,
+ final RowData input,
+ @Nullable final ColumnList op,
+ @Nullable final MapData opMapping) {
+ final String opCode = opMap.get(input.getRowKind());
+ if (opCode == null) {
+ return;
+ }
+
+ opRow.setField(0, StringData.fromString(opCode));
Review Comment:
We can store StringData directly? And avoid the conversion in every invocation.
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2959837323
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.runtime.functions.ptf;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.data.utils.ProjectedRowData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.ColumnList;
+import org.apache.flink.types.RowKind;
+
+import org.apache.flink.table.functions.FunctionContext;
+
+import javax.annotation.Nullable;
+
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Runtime implementation of {@link BuiltInFunctionDefinitions#TO_CHANGELOG}.
+ *
+ * <p>Converts each input row into an INSERT-only output row with an operation code column. The
+ * output schema is {@code [op_column, ...non_partition_columns...]} - the framework prepends
+ * partition key columns automatically.
+ *
+ * <p>Uses {@link ProjectedRowData} for zero-copy projection of non-partition columns and {@link
+ * JoinedRowData} to combine the op column with the projected input.
+ */
+@Internal
+public class ToChangelogFunction extends BuiltInProcessTableFunction<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Map<RowKind, String> DEFAULT_OP_MAPPING =
+ Map.of(
+ RowKind.INSERT, "INSERT",
+ RowKind.UPDATE_BEFORE, "UPDATE_BEFORE",
+ RowKind.UPDATE_AFTER, "UPDATE_AFTER",
+ RowKind.DELETE, "DELETE");
+
+ private final Map<RowKind, String> opMap;
+ private final int[] nonPartitionIndices;
+
+ private transient ProjectedRowData projectedInput;
+ private transient GenericRowData opRow;
+ private transient JoinedRowData output;
+
+ @SuppressWarnings("unchecked")
+ public ToChangelogFunction(final SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.TO_CHANGELOG, context);
+ final CallContext callContext = context.getCallContext();
+
+ final TableSemantics semantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
+ final int[] partitionKeys = semantics.partitionByColumns();
+ final Set<Integer> partitionKeySet =
+ IntStream.of(partitionKeys).boxed().collect(Collectors.toSet());
+
+ final RowType inputType = (RowType) semantics.dataType().getLogicalType();
+ this.nonPartitionIndices = buildNonPartitionIndices(inputType.getFieldCount(), partitionKeySet);
+
+ final Map<String, String> opMapping =
+ callContext.getArgumentValue(2, Map.class).orElse(null);
+ this.opMap = buildOpMap(opMapping);
+ }
+
+ @Override
+ public void open(final FunctionContext context) throws Exception {
+ super.open(context);
+ projectedInput = ProjectedRowData.from(nonPartitionIndices);
+ opRow = new GenericRowData(1);
+ output = new JoinedRowData();
+ }
+
+ private static int[] buildNonPartitionIndices(
+ final int fieldCount, final Set<Integer> partitionKeySet) {
+ return IntStream.range(0, fieldCount)
+ .filter(i -> !partitionKeySet.contains(i))
+ .toArray();
+ }
+
+ private static Map<RowKind, String> buildOpMap(@Nullable final Map<String, String> opMapping) {
+ if (opMapping == null) {
+ return new EnumMap<>(DEFAULT_OP_MAPPING);
+ }
+ final Map<RowKind, String> map = new EnumMap<>(RowKind.class);
+ opMapping.forEach((name, code) -> map.put(RowKind.valueOf(name), code));
+ return map;
+ }
+
+ public void eval(
+ final Context ctx,
+ final RowData input,
+ @Nullable final ColumnList op,
+ @Nullable final MapData opMapping) {
+ final String opCode = opMap.get(input.getRowKind());
+ if (opCode == null) {
+ return;
+ }
+
+ opRow.setField(0, StringData.fromString(opCode));
Review Comment:
Oh yes, that makes sense.
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2959430686
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -108,18 +121,13 @@ public void eval(
final RowData input,
@Nullable final ColumnList op,
@Nullable final Map<String, String> opMapping) {
Review Comment:
https://github.com/apache/flink/pull/27777/commits/f1ebdaf51d1
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java:
##########
@@ -18,66 +18,94 @@
package org.apache.flink.table.planner.plan.stream.sql;
+import org.apache.flink.table.api.ExplainDetail;
import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
import org.apache.flink.table.planner.utils.TableTestBase;
import org.apache.flink.table.planner.utils.TableTestUtil;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** Tests for the TO_CHANGELOG built-in process table function. */
+/**
+ * Plan tests for the TO_CHANGELOG built-in process table function. Uses {@link
+ * ExplainDetail#CHANGELOG_MODE} to verify changelog mode propagation through the plan.
+ */
public class ToChangelogTest extends TableTestBase {
private TableTestUtil util;
@BeforeEach
void setup() {
util = streamTestUtil(TableConfig.getDefault());
+ }
+
+ @Test
+ void testRetractSource() {
util.tableEnv()
.executeSql(
- "CREATE TABLE t ("
+ "CREATE TABLE retract_source ("
+ " id INT,"
+ " name STRING,"
- + " val BIGINT"
- + ") WITH ('connector' = 'values')");
- util.tableEnv()
- .executeSql(
- "CREATE VIEW t_updating AS SELECT id, name, COUNT(*) AS cnt FROM t GROUP BY id, name");
- }
-
- @Test
- void testDefaultCall() {
- util.verifyRelPlan("SELECT * FROM TO_CHANGELOG(input => TABLE t_updating PARTITION BY id)");
- }
-
- @Test
- void testCustomOpName() {
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'changelog-mode' = 'I,UB,UA,D'"
+ + ")");
util.verifyRelPlan(
- "SELECT * FROM TO_CHANGELOG("
- + "input => TABLE t_updating PARTITION BY id, "
- + "op => DESCRIPTOR(op_code))");
+ "SELECT * FROM TO_CHANGELOG(input => TABLE retract_source PARTITION BY id)",
+ JavaScalaConversionUtil.toScala(
Review Comment:
https://github.com/apache/flink/pull/27777/commits/5be5dad62e6
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.planner.plan.stream.sql;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.planner.utils.TableTestUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the TO_CHANGELOG built-in process table function. */
+public class ToChangelogTest extends TableTestBase {
Review Comment:
https://github.com/apache/flink/pull/27777/commits/5be5dad62e6
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2941601109
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/CallBindingCallContext.java:
##########
Review Comment:
update isArgumentLiteral as well? A map can only be literal if all children are also literal
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/CallBindingCallContext.java:
##########
@@ -354,4 +361,15 @@ private static ColumnList convertColumnList(List<SqlNode> operands) {
.collect(Collectors.toList());
return ColumnList.of(names);
}
+
+ private static Map<String, String> convertMap(SqlCall mapCall) {
+ final List<SqlNode> operands = mapCall.getOperandList();
+ final Map<String, String> map = new LinkedHashMap<>();
+ for (int i = 0; i < operands.size(); i += 2) {
+ final String key = SqlLiteral.unchain(operands.get(i)).getValueAs(String.class);
Review Comment:
What happens if this unchaining fails? A map can only be literal if all children are also literal, otherwise the getArgumentValue must return Optional.empty()
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/CallBindingCallContext.java:
##########
@@ -354,4 +361,15 @@ private static ColumnList convertColumnList(List<SqlNode> operands) {
.collect(Collectors.toList());
return ColumnList.of(names);
}
+
+ private static Map<String, String> convertMap(SqlCall mapCall) {
+ final List<SqlNode> operands = mapCall.getOperandList();
+ final Map<String, String> map = new LinkedHashMap<>();
+ for (int i = 0; i < operands.size(); i += 2) {
+ final String key = SqlLiteral.unchain(operands.get(i)).getValueAs(String.class);
Review Comment:
how well is null supported in keys and values? officially we support them in Flink.
--
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-39259][table] Support TO_CHANGELOG built-in process table function (upsert stream, flat mode) [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #27777:
URL: https://github.com/apache/flink/pull/27777#discussion_r2941846580
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.ConstantArgumentCount;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.types.ColumnList;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** Type strategies for the {@code TO_CHANGELOG} process table function. */
+@Internal
+public final class ToChangelogTypeStrategy {
+
+ private static final String DEFAULT_OP_COLUMN_NAME = "op";
+
+ private static final Set<String> VALID_ROW_KIND_NAMES =
+ Set.of("INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE");
+
+ // --------------------------------------------------------------------------------------------
+ // Input validation
+ // --------------------------------------------------------------------------------------------
+
+ public static final InputTypeStrategy INPUT_TYPE_STRATEGY =
+ new InputTypeStrategy() {
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return ConstantArgumentCount.between(1, 3);
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ final CallContext callContext, final boolean throwOnFailure) {
+ return validateInputs(callContext, throwOnFailure);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(final FunctionDefinition definition) {
+ return List.of(
+ Signature.of(Argument.of("input", "TABLE")),
+ Signature.of(
+ Argument.of("input", "TABLE"), Argument.of("op", "DESCRIPTOR")),
+ Signature.of(
+ Argument.of("input", "TABLE"),
+ Argument.of("op", "DESCRIPTOR"),
+ Argument.of("op_mapping", "MAP<STRING, STRING>")));
+ }
+ };
+
+ // --------------------------------------------------------------------------------------------
+ // Output type inference
+ // --------------------------------------------------------------------------------------------
+
+ public static final TypeStrategy OUTPUT_TYPE_STRATEGY =
+ callContext -> {
+ final TableSemantics semantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "First argument must be a table for TO_CHANGELOG."));
+
+ final String opColumnName = resolveOpColumnName(callContext);
+ final List<Field> inputFields = DataType.getFields(semantics.dataType());
+ final Set<Integer> partitionKeys = intArrayToSet(semantics.partitionByColumns());
+
+ final List<Field> outputFields = new ArrayList<>();
+ outputFields.add(DataTypes.FIELD(opColumnName, DataTypes.STRING()));
+ for (int i = 0; i < inputFields.size(); i++) {
+ if (!partitionKeys.contains(i)) {
+ outputFields.add(inputFields.get(i));
+ }
+ }
+
+ return Optional.of(DataTypes.ROW(outputFields));
+ };
+
+ // --------------------------------------------------------------------------------------------
+ // Helpers
+ // --------------------------------------------------------------------------------------------
+
+ private static Optional<List<DataType>> validateInputs(
+ final CallContext callContext, final boolean throwOnFailure) {
+ if (callContext.getTableSemantics(0).isEmpty()) {
+ return callContext.fail(
+ throwOnFailure, "First argument must be a table for TO_CHANGELOG.");
+ }
+
+ final Optional<ColumnList> opDescriptor = callContext.getArgumentValue(1, ColumnList.class);
+ if (opDescriptor.isPresent() && opDescriptor.get().getNames().size() != 1) {
+ return callContext.fail(
+ throwOnFailure,
+ "OP descriptor must contain exactly one column name "
Review Comment:
The descriptor for argument 'op' must contain exactly one column name
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.types.inference.strategies;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.DataTypes.Field;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.ArgumentCount;
+import org.apache.flink.table.types.inference.CallContext;
+import org.apache.flink.table.types.inference.ConstantArgumentCount;
+import org.apache.flink.table.types.inference.InputTypeStrategy;
+import org.apache.flink.table.types.inference.Signature;
+import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.types.ColumnList;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** Type strategies for the {@code TO_CHANGELOG} process table function. */
+@Internal
+public final class ToChangelogTypeStrategy {
+
+ private static final String DEFAULT_OP_COLUMN_NAME = "op";
+
+ private static final Set<String> VALID_ROW_KIND_NAMES =
+ Set.of("INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE");
+
+ // --------------------------------------------------------------------------------------------
+ // Input validation
+ // --------------------------------------------------------------------------------------------
+
+ public static final InputTypeStrategy INPUT_TYPE_STRATEGY =
+ new InputTypeStrategy() {
+ @Override
+ public ArgumentCount getArgumentCount() {
+ return ConstantArgumentCount.between(1, 3);
+ }
+
+ @Override
+ public Optional<List<DataType>> inferInputTypes(
+ final CallContext callContext, final boolean throwOnFailure) {
+ return validateInputs(callContext, throwOnFailure);
+ }
+
+ @Override
+ public List<Signature> getExpectedSignatures(final FunctionDefinition definition) {
+ return List.of(
+ Signature.of(Argument.of("input", "TABLE")),
+ Signature.of(
+ Argument.of("input", "TABLE"), Argument.of("op", "DESCRIPTOR")),
+ Signature.of(
+ Argument.of("input", "TABLE"),
+ Argument.of("op", "DESCRIPTOR"),
+ Argument.of("op_mapping", "MAP<STRING, STRING>")));
+ }
+ };
+
+ // --------------------------------------------------------------------------------------------
+ // Output type inference
+ // --------------------------------------------------------------------------------------------
+
+ public static final TypeStrategy OUTPUT_TYPE_STRATEGY =
+ callContext -> {
+ final TableSemantics semantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(
+ () ->
+ new ValidationException(
+ "First argument must be a table for TO_CHANGELOG."));
+
+ final String opColumnName = resolveOpColumnName(callContext);
+ final List<Field> inputFields = DataType.getFields(semantics.dataType());
+ final Set<Integer> partitionKeys = intArrayToSet(semantics.partitionByColumns());
+
+ final List<Field> outputFields = new ArrayList<>();
+ outputFields.add(DataTypes.FIELD(opColumnName, DataTypes.STRING()));
+ for (int i = 0; i < inputFields.size(); i++) {
+ if (!partitionKeys.contains(i)) {
+ outputFields.add(inputFields.get(i));
+ }
+ }
+
+ return Optional.of(DataTypes.ROW(outputFields));
+ };
+
+ // --------------------------------------------------------------------------------------------
+ // Helpers
+ // --------------------------------------------------------------------------------------------
+
+ private static Optional<List<DataType>> validateInputs(
+ final CallContext callContext, final boolean throwOnFailure) {
+ if (callContext.getTableSemantics(0).isEmpty()) {
+ return callContext.fail(
+ throwOnFailure, "First argument must be a table for TO_CHANGELOG.");
+ }
+
+ final Optional<ColumnList> opDescriptor = callContext.getArgumentValue(1, ColumnList.class);
+ if (opDescriptor.isPresent() && opDescriptor.get().getNames().size() != 1) {
+ return callContext.fail(
+ throwOnFailure,
+ "OP descriptor must contain exactly one column name "
+ + "to define the output operation column name, but got: "
+ + opDescriptor.get().getNames());
+ }
+
+ final Optional<Map> opMapping = callContext.getArgumentValue(2, Map.class);
+ if (opMapping.isPresent()) {
+ for (final Object key : opMapping.get().keySet()) {
+ if (!(key instanceof String) || !VALID_ROW_KIND_NAMES.contains(key)) {
+ return callContext.fail(
+ throwOnFailure,
+ "Invalid RowKind name '%s' in op_mapping. Valid names are: %s.",
Review Comment:
"Invalid target mapping for argument 'op_mapping'."
--
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