[flink] Fix not compatible with Flink 1.18 and 1.19 (java.lang.NoSuchFieldError: MATERIALIZED_TABLE) (#2083)

Author: sd4324530

fluss-flink/fluss-flink-1.18/src/main/java/org/apache/flink/table/catalog/CatalogBaseTable.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.flink.adapter;
 
 import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
 import org.apache.flink.table.catalog.CatalogTable;
 
 import javax.annotation.Nullable;
@@ -26,8 +27,8 @@
 import java.util.Map;
 
 /**
- * A adapter for {@link CatalogTable} constructor. TODO: remove this class when no longer support
- * flink 1.18 and 1.19.
+ * A adapter for {@link CatalogTable} constructor, and adapter MATERIALIZED_TABLE for {@link
+ * CatalogBaseTable.TableKind} TODO: remove this class when no longer support flink 1.18 and 1.19.
  */
 public class CatalogTableAdapter {
     public static CatalogTable toCatalogTable(
@@ -37,4 +38,9 @@ public static CatalogTable toCatalogTable(
             Map<String, String> options) {
         return CatalogTable.of(schema, comment, partitionKeys, options);
     }
+
+    public static boolean isMaterializedTable(CatalogBaseTable.TableKind tableKind) {
+        // flink 1.18 not support MaterializedTable
+        return false;
+    }
 }

fluss-flink/fluss-flink-1.18/src/main/java/org/apache/fluss/flink/adapter/CatalogTableAdapter.java

@@ -0,0 +1,35 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link CatalogTableAdapter} in flink 1.18. */
+public class Flink118CatalogTableTest extends FlinkCatalogTableTest {
+
+    @Test
+    public void testIsMaterializedTable() {
+        assertThat(CatalogTableAdapter.isMaterializedTable(CatalogBaseTable.TableKind.VIEW))
+                .isEqualTo(false);
+        assertThat(CatalogTableAdapter.isMaterializedTable(CatalogBaseTable.TableKind.TABLE))
+                .isEqualTo(false);
+    }
+}

fluss-flink/fluss-flink-1.18/src/test/java/org/apache/fluss/flink/adapter/Flink118CatalogTableTest.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.flink.adapter;
 
 import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
 import org.apache.flink.table.catalog.CatalogTable;
 
 import javax.annotation.Nullable;
@@ -26,8 +27,8 @@
 import java.util.Map;
 
 /**
- * A adapter for {@link CatalogTable} constructor. TODO: remove this class when no longer support
- * flink 1.18 and 1.19.
+ * A adapter for {@link CatalogTable} constructor, and adapter MATERIALIZED_TABLE for {@link
+ * CatalogBaseTable.TableKind} TODO: remove this class when no longer support flink 1.18 and 1.19.
  */
 public class CatalogTableAdapter {
     public static CatalogTable toCatalogTable(
@@ -37,4 +38,9 @@ public static CatalogTable toCatalogTable(
             Map<String, String> options) {
         return CatalogTable.of(schema, comment, partitionKeys, options);
     }
+
+    public static boolean isMaterializedTable(CatalogBaseTable.TableKind tableKind) {
+        // flink 1.19 not support MaterializedTable
+        return false;
+    }
 }

fluss-flink/fluss-flink-1.19/src/main/java/org/apache/flink/table/catalog/CatalogBaseTable.java

@@ -0,0 +1,35 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link CatalogTableAdapter} in flink 1.19. */
+public class Flink119CatalogTableTest extends FlinkCatalogTableTest {
+
+    @Test
+    public void testIsMaterializedTable() {
+        assertThat(CatalogTableAdapter.isMaterializedTable(CatalogBaseTable.TableKind.VIEW))
+                .isEqualTo(false);
+        assertThat(CatalogTableAdapter.isMaterializedTable(CatalogBaseTable.TableKind.TABLE))
+                .isEqualTo(false);
+    }
+}

fluss-flink/fluss-flink-1.19/src/main/java/org/apache/fluss/flink/adapter/CatalogTableAdapter.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.flink.adapter;
 
 import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
 import org.apache.flink.table.catalog.CatalogTable;
 
 import javax.annotation.Nullable;
@@ -26,8 +27,8 @@
 import java.util.Map;
 
 /**
- * A adapter for {@link CatalogTable} constructor. TODO: remove this class when no longer support
- * flink 1.18 and 1.19.
+ * A adapter for {@link CatalogTable} constructor, and adapter MATERIALIZED_TABLE for {@link
+ * CatalogBaseTable.TableKind} TODO: remove this class when no longer support flink 1.18 and 1.19.
  */
 public class CatalogTableAdapter {
     public static CatalogTable toCatalogTable(
@@ -42,4 +43,8 @@ public static CatalogTable toCatalogTable(
                 .options(options)
                 .build();
     }
+
+    public static boolean isMaterializedTable(CatalogBaseTable.TableKind tableKind) {
+        return CatalogBaseTable.TableKind.MATERIALIZED_TABLE == tableKind;
+    }
 }