You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "raminqaf (via GitHub)" <gi...@apache.org> on 2026/05/08 11:06:13 UTC
[PR] [FLINK-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
raminqaf opened a new pull request, #28128:
URL: https://github.com/apache/flink/pull/28128
### What is the purpose of the change
Adds `fromChangelog()` and `toChangelog()` convenience methods to PartitionedTable so users can invoke the built-in `FROM_CHANGELOG` and `TO_CHANGELOG PTF`s with set semantics from the Table API. Today this requires the verbose `partitionBy(...).process("FROM_CHANGELOG", ...)` form.
### Brief change log
- Added `fromChangelog(Expression...)` and `toChangelog(Expression...)` to PartitionedTable.
- Implemented them in PartitionedTableImpl by delegating to `process(...)` on the partitioned operation tree.
- Cross-linked the new methods from the row-semantics overloads in Table.java.
- Added `TABLE_API_RETRACT_PARTITION_BY` test programs to both `FromChangelogTestPrograms` and `ToChangelogTestPrograms`, registered in the corresponding *SemanticTests.
- Updated `docs/content/docs/sql/reference/queries/changelog.md` with the new Table API entry points.
### Verifying this change
This change added tests and can be verified as follows:
- `FromChangelogSemanticTests#TABLE_API_RETRACT_PARTITION_BY` exercises `cdcStream.partitionBy($("id")).fromChangelog()` with a retract input.
- `ToChangelogSemanticTests#TABLE_API_RETRACT_PARTITION_BY` exercises `myTable.partitionBy($("id")).toChangelog()` with a retract input where the partition key is a non-leading column.
- Both verify the optimized plan and the produced changelog.
### Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes. Two new methods on `@PublicEvolving` PartitionedTable.
- 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 and JavaDocs
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change the checkbox below to `[X]` followed by the name of the tool, and uncomment the
"Generated-by" line. See the ASF Generative Tooling Guidance for details:
https://www.apache.org/legal/generative-tooling.html
You are responsible for the quality and correctness of every change in this PR
regardless of the tooling used. Low-effort AI-generated PRs will be closed. See
AGENTS.md for the full guidance.
-->
- [x] Yes (please specify the tool below)
Generated-by: Opus 4.7 (for docs only)
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28128:
URL: https://github.com/apache/flink/pull/28128#issuecomment-4405944617
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "6620c12c1412583cf29f1bbfaef8302aaa8832e1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6620c12c1412583cf29f1bbfaef8302aaa8832e1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6620c12c1412583cf29f1bbfaef8302aaa8832e1 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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske merged PR #28128:
URL: https://github.com/apache/flink/pull/28128
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3225814572
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,96 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Default: adds 'op' column and supports all changelog modes
+ * Table result = table.partitionBy($("id")).toChangelog();
+ *
+ * // Custom op column name and mapping
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("op_code").asArgument("op"),
+ * map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
+ * );
+ *
+ * // Deletion flag pattern: comma-separated keys map multiple change operations to the same code
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("deleted").asArgument("op"),
+ * map("INSERT, UPDATE_AFTER", "false", "DELETE", "true").asArgument("op_mapping")
+ * );
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op} and {@code op_mapping}
+ * @return an append-only {@link Table} ordered as {@code [partition_keys, op,
Review Comment:
what do you mean by "ordered"?
The output isn't explicitly ordered, right?
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,96 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Default: adds 'op' column and supports all changelog modes
+ * Table result = table.partitionBy($("id")).toChangelog();
+ *
+ * // Custom op column name and mapping
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("op_code").asArgument("op"),
+ * map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
+ * );
+ *
+ * // Deletion flag pattern: comma-separated keys map multiple change operations to the same code
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("deleted").asArgument("op"),
+ * map("INSERT, UPDATE_AFTER", "false", "DELETE", "true").asArgument("op_mapping")
+ * );
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op} and {@code op_mapping}
+ * @return an append-only {@link Table} ordered as {@code [partition_keys, op,
+ * non_partition_input_columns]}
+ * @see Table#toChangelog(Expression...)
+ */
+ Table toChangelog(Expression... arguments);
+
+ /**
+ * Converts this partitioned append-only table with an explicit operation code column into a
+ * (potentially updating) dynamic table using the built-in {@code FROM_CHANGELOG} process table
+ * function with set semantics.
+ *
+ * <p>Each input row is expected to have a string column that indicates the change operation.
+ * The operation column is interpreted by the engine and removed from the output. With set
+ * semantics, rows for the same partition key are co-located in the same parallel operator
+ * instance, which is required when downstream operators are keyed on that column.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#fromChangelog} on
+ * the unpartitioned table.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Default: reads 'op' column with standard change operation names
+ * Table result = cdcStream.partitionBy($("id")).fromChangelog();
+ *
+ * // With custom op column name
+ * Table result = cdcStream
+ * .partitionBy($("id"))
+ * .fromChangelog(descriptor("operation").asArgument("op"));
+ *
+ * // With custom op_mapping
+ * Table result = cdcStream
+ * .partitionBy($("id"))
+ * .fromChangelog(
+ * descriptor("op").asArgument("op"),
+ * map("c, r", "INSERT",
+ * "ub", "UPDATE_BEFORE",
+ * "ua", "UPDATE_AFTER",
+ * "d", "DELETE").asArgument("op_mapping")
+ * );
+ *
+ * // Silently skip rows with NULL or unmapped op codes instead of failing
+ * Table result = cdcStream
+ * .partitionBy($("id"))
+ * .fromChangelog(lit("SKIP").asArgument("error_handling"));
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op}, {@code op_mapping}, and {@code
+ * error_handling}
+ * @return a dynamic {@link Table} ordered as {@code [partition_keys,
Review Comment:
same as above
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3224229924
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,70 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("op_code").asArgument("op"),
+ * map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
+ * );
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op} and {@code op_mapping}
+ * @return an append-only {@link Table} with an {@code op} column and the partition keys
+ * prepended to the input columns
Review Comment:
Addressed
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java:
##########
@@ -1431,6 +1431,17 @@ default TableResult executeInsert(
* INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
* UPDATE_AFTER, DELETE, etc.).
*
+ * <p>By default, the input is processed with row semantics (each row independently). To
+ * co-locate rows with the same key in the same parallel operator instance, partition the input
+ * first via {@link #partitionBy(Expression...)} and invoke {@link
+ * PartitionedTable#toChangelog(Expression...)}:
Review Comment:
Addressed
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3227545596
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,96 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Default: adds 'op' column and supports all changelog modes
+ * Table result = table.partitionBy($("id")).toChangelog();
+ *
+ * // Custom op column name and mapping
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("op_code").asArgument("op"),
+ * map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
+ * );
+ *
+ * // Deletion flag pattern: comma-separated keys map multiple change operations to the same code
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("deleted").asArgument("op"),
+ * map("INSERT, UPDATE_AFTER", "false", "DELETE", "true").asArgument("op_mapping")
+ * );
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op} and {@code op_mapping}
+ * @return an append-only {@link Table} ordered as {@code [partition_keys, op,
Review Comment:
I think `with output schema` would be a better rephrase here. I will address this.
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3224230917
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,70 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * Table result = table
+ * .partitionBy($("id"))
Review Comment:
Addressed
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3208870396
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java:
##########
@@ -1431,6 +1431,17 @@ default TableResult executeInsert(
* INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
* UPDATE_AFTER, DELETE, etc.).
*
+ * <p>By default, the input is processed with row semantics (each row independently). To
+ * co-locate rows with the same key in the same parallel operator instance, partition the input
+ * first via {@link #partitionBy(Expression...)} and invoke {@link
+ * PartitionedTable#toChangelog(Expression...)}:
Review Comment:
nit: The new partitioned reference reads *"invoke `PartitionedTable#toChangelog(Expression...)`"* — `fromChangelog` (line 1483) has the same sentence with the *"with set semantics"* qualifier. Add it here too for symmetry.
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3208871244
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,70 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * Table result = table
+ * .partitionBy($("id"))
Review Comment:
nit: The unpartitioned `Table#toChangelog` lead-in example shows the no-arg form first (`table.toChangelog()`). The new javadoc jumps straight to the with-args case. Adding the minimal form first matches the docs and shows the most common usage:
```java
// Default: adds 'op' column and supports all changelog modes
Table result = table.partitionBy($("id")).toChangelog();
// With custom parameters
Table result = table
.partitionBy($("id"))
.toChangelog(
descriptor("op_code").asArgument("op"),
map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
);
```
Same fix for `fromChangelog`. Actually, the regular table has multiple examples. Any reason why not to copy the javadocs why all the examples and only adjust them with partition by?
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3208868588
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,70 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("op_code").asArgument("op"),
+ * map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
+ * );
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op} and {@code op_mapping}
+ * @return an append-only {@link Table} with an {@code op} column and the partition keys
+ * prepended to the input columns
Review Comment:
nit: The `@return` says "partition keys prepended to the input columns" but the actual output drops the partition columns from the trailing portion (FLINK-39614). Worth matching the docs' explicit shape so readers don't expect duplicates:
```suggestion
* @return an append-only {@link Table} ordered as `[partition_keys, op, non_partition_input_columns]`
```
Same fix for `fromChangelog`.
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3208871244
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,70 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * Table result = table
+ * .partitionBy($("id"))
Review Comment:
nit: The unpartitioned `Table#toChangelog` lead-in example shows the no-arg form first (`table.toChangelog()`). The new javadoc jumps straight to the with-args case. Adding the minimal form first matches the docs and shows the most common usage:
```java
// Default: adds 'op' column and supports all changelog modes
Table result = table.partitionBy($("id")).toChangelog();
// With custom parameters
Table result = table
.partitionBy($("id"))
.toChangelog(
descriptor("op_code").asArgument("op"),
map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
);
```
Same fix for `fromChangelog`.
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3208870396
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java:
##########
@@ -1431,6 +1431,17 @@ default TableResult executeInsert(
* INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
* UPDATE_AFTER, DELETE, etc.).
*
+ * <p>By default, the input is processed with row semantics (each row independently). To
+ * co-locate rows with the same key in the same parallel operator instance, partition the input
+ * first via {@link #partitionBy(Expression...)} and invoke {@link
+ * PartitionedTable#toChangelog(Expression...)}:
Review Comment:
nit: The new partitioned reference reads *"invoke `PartitionedTable#toChangelog(Expression...)`"* — `fromChangelog` (line 1483) has the same sentence with the *"with set semantics"* qualifier. Add it here too for symmetry
--
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-39632][table] Add fromChangelog() / toChangelog() convenience methods to PartitionedTable [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28128:
URL: https://github.com/apache/flink/pull/28128#discussion_r3227557887
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -169,4 +169,96 @@ public interface PartitionedTable {
* @see ProcessTableFunction
*/
Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
+
+ /**
+ * Converts this partitioned dynamic table into an append-only table with an explicit operation
+ * code column using the built-in {@code TO_CHANGELOG} process table function with set
+ * semantics.
+ *
+ * <p>Each input row - regardless of its original change operation - is emitted as an
+ * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
+ * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+ * co-located in the same parallel operator instance.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
+ * unpartitioned table.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Default: adds 'op' column and supports all changelog modes
+ * Table result = table.partitionBy($("id")).toChangelog();
+ *
+ * // Custom op column name and mapping
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("op_code").asArgument("op"),
+ * map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
+ * );
+ *
+ * // Deletion flag pattern: comma-separated keys map multiple change operations to the same code
+ * Table result = table
+ * .partitionBy($("id"))
+ * .toChangelog(
+ * descriptor("deleted").asArgument("op"),
+ * map("INSERT, UPDATE_AFTER", "false", "DELETE", "true").asArgument("op_mapping")
+ * );
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op} and {@code op_mapping}
+ * @return an append-only {@link Table} ordered as {@code [partition_keys, op,
+ * non_partition_input_columns]}
+ * @see Table#toChangelog(Expression...)
+ */
+ Table toChangelog(Expression... arguments);
+
+ /**
+ * Converts this partitioned append-only table with an explicit operation code column into a
+ * (potentially updating) dynamic table using the built-in {@code FROM_CHANGELOG} process table
+ * function with set semantics.
+ *
+ * <p>Each input row is expected to have a string column that indicates the change operation.
+ * The operation column is interpreted by the engine and removed from the output. With set
+ * semantics, rows for the same partition key are co-located in the same parallel operator
+ * instance, which is required when downstream operators are keyed on that column.
+ *
+ * <p>For row semantics (each row processed independently), use {@link Table#fromChangelog} on
+ * the unpartitioned table.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * // Default: reads 'op' column with standard change operation names
+ * Table result = cdcStream.partitionBy($("id")).fromChangelog();
+ *
+ * // With custom op column name
+ * Table result = cdcStream
+ * .partitionBy($("id"))
+ * .fromChangelog(descriptor("operation").asArgument("op"));
+ *
+ * // With custom op_mapping
+ * Table result = cdcStream
+ * .partitionBy($("id"))
+ * .fromChangelog(
+ * descriptor("op").asArgument("op"),
+ * map("c, r", "INSERT",
+ * "ub", "UPDATE_BEFORE",
+ * "ua", "UPDATE_AFTER",
+ * "d", "DELETE").asArgument("op_mapping")
+ * );
+ *
+ * // Silently skip rows with NULL or unmapped op codes instead of failing
+ * Table result = cdcStream
+ * .partitionBy($("id"))
+ * .fromChangelog(lit("SKIP").asArgument("error_handling"));
+ * }</pre>
+ *
+ * @param arguments optional named arguments for {@code op}, {@code op_mapping}, and {@code
+ * error_handling}
+ * @return a dynamic {@link Table} ordered as {@code [partition_keys,
Review Comment:
Addressed :)
--
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