[core]support conf for flink

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkTableFactory.java

@@ -52,6 +52,13 @@ public String factoryIdentifier() {
 
     @Override
     public DynamicTableSource createDynamicTableSource(Context context) {
+        CatalogTable table = context.getCatalogTable().getOrigin();
+        if (table instanceof FormatCatalogTable) {
+            CoreOptions options = new CoreOptions(((FormatCatalogTable) table).table().options());
+            if (!options.formatTableImplementationIsPaimon()) {
+                return ((FormatCatalogTable) table).createTableSource(context);
+            }
+        }
         createTableIfNeeded(context);
         return super.createDynamicTableSource(context);
     }
@@ -60,7 +67,13 @@ public DynamicTableSource createDynamicTableSource(Context context) {
     public DynamicTableSink createDynamicTableSink(Context context) {
         CatalogTable table = context.getCatalogTable().getOrigin();
         if (table instanceof FormatCatalogTable) {
-            return ((FormatCatalogTable) table).createTableSink(context);
+            CoreOptions options = new CoreOptions(((FormatCatalogTable) table).table().options());
+            if (!options.formatTableImplementationIsPaimon()) {
+                return ((FormatCatalogTable) table).createTableSink(context);
+            } else {
+                throw new UnsupportedOperationException(
+                        "Format table's sink is not supported yet.");
+            }
         }
         createTableIfNeeded(context);
         return super.createDynamicTableSink(context);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FormatCatalogTable.java

@@ -23,6 +23,7 @@
 import org.apache.flink.table.api.Schema;
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.factories.DynamicTableFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.types.logical.RowType;
@@ -127,6 +128,17 @@ public Optional<String> getDetailedDescription() {
         return getDescription();
     }
 
+    public DynamicTableSource createTableSource(DynamicTableFactory.Context context) {
+        return FactoryUtil.createDynamicTableSource(
+                null,
+                context.getObjectIdentifier(),
+                context.getCatalogTable(),
+                new HashMap<>(),
+                context.getConfiguration(),
+                context.getClassLoader(),
+                context.isTemporary());
+    }
+
     public DynamicTableSink createTableSink(DynamicTableFactory.Context context) {
         return FactoryUtil.createDynamicTableSink(
                 null,

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableReadITCaseBase.java → paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/PaimonFormatTableTest.java

@@ -47,9 +47,10 @@
 import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** ITCase for format table. */
-public class FormatTableReadITCaseBase extends RESTCatalogITCaseBase {
+public class PaimonFormatTableTest extends RESTCatalogITCaseBase {
 
     @Test
     public void testParquetFileFormat() throws IOException {
@@ -69,7 +70,11 @@ public void testParquetFileFormat() throws IOException {
         datas[1] = GenericRow.of(2, 2, 2);
         String tableName = "format_table_test";
         sql(
-                "CREATE TABLE %s (a INT, b INT, c INT) WITH ('file.format'='parquet', 'type'='format-table')",
+                "CREATE TABLE %s (a INT, b INT, c INT) WITH ("
+                        + "'file.format'='parquet',"
+                        + "'type'='format-table',"
+                        + "'format-table.implementation'='paimon'"
+                        + ")",
                 tableName);
         write(
                 factory,
@@ -84,6 +89,8 @@ public void testParquetFileFormat() throws IOException {
         restCatalogServer.setDataToken(identifier, expiredDataToken);
         assertThat(sql("SELECT a FROM %s", tableName))
                 .containsExactlyInAnyOrder(Row.of(1), Row.of(2));
+        assertThatThrownBy(() -> sql("INSERT INTO %s VALUES (3, 3, 3)", tableName))
+                .isInstanceOf(RuntimeException.class);
         sql("Drop TABLE %s", tableName);
     }