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/05/06 14:16:43 UTC
[PR] [FLINK-39614][table] Support TO_CHANGELOG: retract stream -> retract … [flink]
gustavodemorais opened a new pull request, #28120:
URL: https://github.com/apache/flink/pull/28120
## What is the purpose of the change
`TO_CHANGELOG` was locked to row semantics. `PARTITION BY` silently duplicated the partition key in the output. This PR honors the conditional `SET_SEMANTIC_TABLE` trait declared in FLINK-39392: when `PARTITION BY` is provided, the framework prepends the partition key and the function emits the rest. Output: `[partition_key, op, ...other input fields]`.
## Brief change log
- Added `ChangelogTypeStrategyUtils#computeOutputIndices` helper (shared with `FROM_CHANGELOG` once FLINK-39537 lands).
- Wired `ToChangelogTypeStrategy` and `ToChangelogFunction` to drop partition keys from the declared/emitted schema.
- Added semantic tests for leading and middle-column partition keys; updated the existing plan test golden XML.
## Verifying this change
Covered by `ToChangelogSemanticTests` (two new test programs) and the updated `ToChangelogTest#testRetractPartitionBy` plan test.
## Does this pull request potentially affect one of the following parts:
- Dependencies: no
- The public API: no
- The serializers: no
- The runtime per-record code paths: no
- Deployment / recovery: no
- S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no (fixes broken behavior)
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes (please specify the tool below)
2.1.117 (Claude Code)
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #28120:
URL: https://github.com/apache/flink/pull/28120#issuecomment-4389918913
I've addressed your comments, @raminqaf and @snuyanzin. 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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3197048532
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns (the PTF framework prepends them when the input has set semantics).
+ */
+ public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
+ final int inputFieldCount = DataType.getFieldNames(tableSemantics.dataType()).size();
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ return filterIndices(inputFieldCount, excluded);
+ }
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns and the operation column matching {@code opColumnName}.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final String opColumnName) {
+ final List<String> inputFieldNames = DataType.getFieldNames(tableSemantics.dataType());
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ final int opIndex = inputFieldNames.indexOf(opColumnName);
+ if (opIndex >= 0) {
+ excluded.add(opIndex);
+ }
+ return filterIndices(inputFieldNames.size(), excluded);
+ }
+
+ private static Set<Integer> collectPartitionKeyIndices(final TableSemantics tableSemantics) {
+ final Set<Integer> indices = new HashSet<>();
+ for (final int idx : tableSemantics.partitionByColumns()) {
+ indices.add(idx);
+ }
+ return indices;
Review Comment:
```suggestion
final Set<Integer> indices = new HashSet<>(Arrays.stream(tableSemantics.partitionByColumns()).boxed().toList());
```
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3197076350
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns (the PTF framework prepends them when the input has set semantics).
+ */
+ public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
+ final int inputFieldCount = DataType.getFieldNames(tableSemantics.dataType()).size();
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ return filterIndices(inputFieldCount, excluded);
+ }
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns and the operation column matching {@code opColumnName}.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final String opColumnName) {
+ final List<String> inputFieldNames = DataType.getFieldNames(tableSemantics.dataType());
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ final int opIndex = inputFieldNames.indexOf(opColumnName);
+ if (opIndex >= 0) {
+ excluded.add(opIndex);
+ }
Review Comment:
this method and the one above are differ only by this code snippet
```java
final int opIndex = inputFieldNames.indexOf(opColumnName);
if (opIndex >= 0) {
excluded.add(opIndex);
}
```
why can not we reuse the logic then?
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196276701
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.Field;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding: -
+ * Partition keys, if present (the PTF framework prepends them when the input has set semantics)
+ * - The operation column {@code opColumnName} when non-null and present.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final @Nullable String opColumnName) {
Review Comment:
I would rather create an override of the method one for TO_CHANGELOG and one for FROM_CHANGELOG.
```suggestion
public static int[] computeOutputIndices(
final TableSemantics tableSemantics, final String opColumnName) {
// ...
}
public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
// ....
}
```
Alternatively, it is good to explain why opColumnName is nullable.
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #28120:
URL: https://github.com/apache/flink/pull/28120#issuecomment-4395800482
Thanks @snuyanzin, I think the PR is in a good shape now. 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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196472166
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.Field;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding: -
+ * Partition keys, if present (the PTF framework prepends them when the input has set semantics)
+ * - The operation column {@code opColumnName} when non-null and present.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final @Nullable String opColumnName) {
Review Comment:
added overload
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196980865
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns (the PTF framework prepends them when the input has set semantics).
+ */
+ public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
+ final int inputFieldCount = DataType.getFieldNames(tableSemantics.dataType()).size();
Review Comment:
```suggestion
final int inputFieldCount = DataType.getFieldCount();
```
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #28120:
URL: https://github.com/apache/flink/pull/28120
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3200264561
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns (the PTF framework prepends them when the input has set semantics).
+ */
+ public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
+ final int inputFieldCount = DataType.getFieldNames(tableSemantics.dataType()).size();
Review Comment:
done
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196472000
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -70,6 +75,14 @@ public ToChangelogFunction(final SpecializedContext context) {
final Map<String, String> opMapping =
callContext.getArgumentValue(2, Map.class).orElse(null);
this.rawOpMap = buildOpMap(opMapping);
+
+ // Drop partition keys when set semantics: the framework prepends them automatically.
+ // For row semantics, partitionByColumns is empty and this is a no-op identity projection.
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
Review Comment:
removed + comment
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196980865
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns (the PTF framework prepends them when the input has set semantics).
+ */
+ public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
+ final int inputFieldCount = DataType.getFieldNames(tableSemantics.dataType()).size();
Review Comment:
```suggestion
final int inputFieldCount = DataType.getFieldCount(tableSemantics.dataType());
```
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #28120:
URL: https://github.com/apache/flink/pull/28120#issuecomment-4389851427
> if need fields names only for field names and size, why do not use DataType#getFieldsNames then? seems code comment do not work for me for whatever reason
That's a good callout, @snuyanzin, thanks
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196276701
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.Field;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding: -
+ * Partition keys, if present (the PTF framework prepends them when the input has set semantics)
+ * - The operation column {@code opColumnName} when non-null and present.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final @Nullable String opColumnName) {
Review Comment:
I would rather create an override of the method one for TO_CHANGELOG and one for FROM_CHANGELOG.
```suggestion
public static int[] computeOutputIndices(
final TableSemantics tableSemantics, final String opColumnName) {
// ...
}
public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
// ....
}
```
Alternatively, it is good to explain why opColumnName is nullable.
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #28120:
URL: https://github.com/apache/flink/pull/28120#issuecomment-4389792609
if need it only for field names and size, why do not use DataType#getFieldsNames then?
seems code comment do not work for me for whatever reason
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28120:
URL: https://github.com/apache/flink/pull/28120#issuecomment-4389033137
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9d6af4026e8039655419ea4d1e55fa9eb529c44c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9d6af4026e8039655419ea4d1e55fa9eb529c44c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9d6af4026e8039655419ea4d1e55fa9eb529c44c 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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3200266810
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns (the PTF framework prepends them when the input has set semantics).
+ */
+ public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
+ final int inputFieldCount = DataType.getFieldNames(tableSemantics.dataType()).size();
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ return filterIndices(inputFieldCount, excluded);
+ }
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns and the operation column matching {@code opColumnName}.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final String opColumnName) {
+ final List<String> inputFieldNames = DataType.getFieldNames(tableSemantics.dataType());
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ final int opIndex = inputFieldNames.indexOf(opColumnName);
+ if (opIndex >= 0) {
+ excluded.add(opIndex);
+ }
+ return filterIndices(inputFieldNames.size(), excluded);
+ }
+
+ private static Set<Integer> collectPartitionKeyIndices(final TableSemantics tableSemantics) {
+ final Set<Integer> indices = new HashSet<>();
+ for (final int idx : tableSemantics.partitionByColumns()) {
+ indices.add(idx);
+ }
+ return indices;
Review Comment:
done
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196245007
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java:
##########
@@ -96,10 +97,14 @@ public List<Signature> getExpectedSignatures(final FunctionDefinition definition
final String opColumnName = resolveOpColumnName(callContext);
final List<Field> inputFields = DataType.getFields(semantics.dataType());
+ // Excludes partition keys when set semantics; the framework prepends them so
+ // including them again here would duplicate the columns.
Review Comment:
nit: not really needed since the java docs of the method explain already what it does
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.Field;
+import org.apache.flink.table.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding: -
+ * Partition keys, if present (the PTF framework prepends them when the input has set semantics)
+ * - The operation column {@code opColumnName} when non-null and present.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final @Nullable String opColumnName) {
Review Comment:
I would rather create an override of the method one for TO_CHANGELOG and one for FROM_CHANGELOG.
```suggestion
public static int[] computeOutputIndices(
final TableSemantics tableSemantics, final @Nullable String opColumnName) {
// ...
}
public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
// ....
}
```
Alternatively, it is good to explain why opColumnName is nullable.
##########
flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml:
##########
@@ -35,42 +35,42 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(), DEFAULT(), D
]]>
</Resource>
</TestCase>
- <TestCase name="testRetractSource">
+ <TestCase name="testRetractPartitionBy">
<Resource name="sql">
- <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE retract_source)]]>
+ <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE retract_source PARTITION BY id)]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalProject(op=[$0], id=[$1], name=[$2])
-+- LogicalTableFunctionScan(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) op, INTEGER id, VARCHAR(2147483647) name)])
+LogicalProject(id=[$0], op=[$1], name=[$2])
Review Comment:
Just to point out:
- If we use `PARTITION BY id` the LogicalProject would be id, op, ...
- If we avoid `PARTITION BY` clause the LogicalProject would be op, id,...
is this intended?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -70,6 +75,14 @@ public ToChangelogFunction(final SpecializedContext context) {
final Map<String, String> opMapping =
callContext.getArgumentValue(2, Map.class).orElse(null);
this.rawOpMap = buildOpMap(opMapping);
+
+ // Drop partition keys when set semantics: the framework prepends them automatically.
+ // For row semantics, partitionByColumns is empty and this is a no-op identity projection.
+ final TableSemantics tableSemantics =
+ callContext
+ .getTableSemantics(0)
+ .orElseThrow(() -> new IllegalStateException("Table argument expected."));
Review Comment:
Needed? Should be caught in the TypeStrategy as Validation error during planning
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3196472251
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java:
##########
@@ -96,10 +97,14 @@ public List<Signature> getExpectedSignatures(final FunctionDefinition definition
final String opColumnName = resolveOpColumnName(callContext);
final List<Field> inputFields = DataType.getFields(semantics.dataType());
+ // Excludes partition keys when set semantics; the framework prepends them so
+ // including them again here would duplicate the columns.
Review Comment:
removed
##########
flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml:
##########
@@ -35,42 +35,42 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(), DEFAULT(), D
]]>
</Resource>
</TestCase>
- <TestCase name="testRetractSource">
+ <TestCase name="testRetractPartitionBy">
<Resource name="sql">
- <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE retract_source)]]>
+ <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE retract_source PARTITION BY id)]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalProject(op=[$0], id=[$1], name=[$2])
-+- LogicalTableFunctionScan(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) op, INTEGER id, VARCHAR(2147483647) name)])
+LogicalProject(id=[$0], op=[$1], name=[$2])
Review Comment:
yeah, that checks out for me
--
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-39614][table] Support TO_CHANGELOG: retract stream -> retract with set semantics [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28120:
URL: https://github.com/apache/flink/pull/28120#discussion_r3200273760
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ChangelogTypeStrategyUtils.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.functions.TableSemantics;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+/** Shared helpers for changelog-style PTFs ({@code TO_CHANGELOG}, {@code FROM_CHANGELOG}). */
+@Internal
+public final class ChangelogTypeStrategyUtils {
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns (the PTF framework prepends them when the input has set semantics).
+ */
+ public static int[] computeOutputIndices(final TableSemantics tableSemantics) {
+ final int inputFieldCount = DataType.getFieldNames(tableSemantics.dataType()).size();
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ return filterIndices(inputFieldCount, excluded);
+ }
+
+ /**
+ * Returns the input column indices that pass through to the function's output, excluding the
+ * partition key columns and the operation column matching {@code opColumnName}.
+ */
+ public static int[] computeOutputIndices(
+ final TableSemantics tableSemantics, final String opColumnName) {
+ final List<String> inputFieldNames = DataType.getFieldNames(tableSemantics.dataType());
+ final Set<Integer> excluded = collectPartitionKeyIndices(tableSemantics);
+ final int opIndex = inputFieldNames.indexOf(opColumnName);
+ if (opIndex >= 0) {
+ excluded.add(opIndex);
+ }
Review Comment:
I've adjusted the overloads so they don't have any duplicate code, see https://github.com/apache/flink/pull/28120/commits/b5b1722df09031a6758491e1403da6973e285b58
--
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