(fluss) branch main updated: [hotfix][flink] Use mini operation scope of MaterializedTableChange. (#1808)

[hongshun]

This is an automated email from the ASF dual-hosted git repository.

hongshun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 97bd2eb75 [hotfix][flink] Use mini operation scope of 
MaterializedTableChange. (#1808)
97bd2eb75 is described below

commit 97bd2eb75c41d46d110a4224412796470a20fe8d
Author: Hongshun Wang <125648852+loserwang1...@users.noreply.github.com>
AuthorDate: Fri Oct 17 17:04:12 2025 +0800

    [hotfix][flink] Use mini operation scope of MaterializedTableChange. (#1808)
---
 .../java/org/apache/fluss/flink/utils/FlinkConversions.java    | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git 
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
 
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
index c4624b527..3256b3f52 100644
--- 
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
+++ 
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
@@ -390,11 +390,7 @@ public class FlinkConversions {
 
     public static List<TableChange> toFlussTableChanges(
             org.apache.flink.table.catalog.TableChange tableChange) {
-        if (tableChange
-                instanceof 
org.apache.flink.table.catalog.TableChange.MaterializedTableChange) {
-            // MaterializedTableChange may produce multiple fluss TableChange,
-            return convertMaterializedTableChange(tableChange);
-        } else if (tableChange instanceof 
org.apache.flink.table.catalog.TableChange.SetOption) {
+        if (tableChange instanceof 
org.apache.flink.table.catalog.TableChange.SetOption) {
             return Collections.singletonList(
                     convertSetOption(
                             
(org.apache.flink.table.catalog.TableChange.SetOption) tableChange));
@@ -402,6 +398,10 @@ public class FlinkConversions {
             return Collections.singletonList(
                     convertResetOption(
                             
(org.apache.flink.table.catalog.TableChange.ResetOption) tableChange));
+        } else if (tableChange instanceof ModifyRefreshStatus
+                || tableChange instanceof ModifyRefreshHandler) {
+            // MaterializedTableChange may produce multiple fluss TableChange.
+            return convertMaterializedTableChange(tableChange);
         } else {
             throw new UnsupportedOperationException(
                     String.format("Unsupported flink table change: %s.", 
tableChange));