hi all! I’m following the Flink Table API doc for 1.7.1 and trying to 1) read data from MySQL CDC (this seems to work ✅), 2) do basic transforms on it (this seems to work when I do Table.execute().print() ✅), and 3) write the transformed table back to a new MySQL table on the same source database <--- having problem here 😭 My java program runs something like this

TableResult input = tableEnv.executeSql(input_sql_table);
		TableResult test_output = tableEnv.executeSql(output_sql_table);

		Table input_raw = tableEnv.from("input"); // when this prints, it is correct
		Table input_aggregate = simple_aggregate(input_raw); // when this prints, it is correct
		TablePipeline tablePipeline = input_aggregate.insertInto("kim_test");
		TableResult tableResult = tablePipeline.execute();
		tableResult.await();

The output_sql_table string I have is:

String output_sql_table = "CREATE TABLE IF NOT EXISTS kim_test (\n" +
      "     account_id STRING,\n" +
      "     txn_aggregate_amount INT,\n" +
      "     txn_aggregate_count INT,\n" +
      "     PRIMARY KEY(account_id) NOT ENFORCED\n" +
      "     ) WITH (\n" +
      "     'connector' = 'jdbc',\n" +
      "     'url'        = 'jdbc:<mysql://ip>:port/db',\n" +
      "     'table-name' = 'kim_test',\n" +
      "     'driver'     = 'com.mysql.jdbc.Driver',\n" +
      "     'username' = '" + username + "',\n" +
      "     'password' = '" + password + "'"+
      ")";

Which runs into this error:

Caused by: java.sql.SQLSyntaxErrorException: Table 'db.kim_test' doesn't exist

I can manually go into the mysql db and can confirm that the db exists, and i’m able to read from db.input (one of the table I got my CDC from). db.kim_test doesn’t exist of course (because I want to create it as the MySQL output/sink). I’m tearing my hair out as how to proceed, and would love if anyone has any pointers!

JDBC connector doesn’t create tables if they don’t exist at the moment: https://issues.apache.org/jira/browse/FLINK-16294

this is super helpful! Thank you! I knew I was missing something obvious!!!

Tables in Flink SQL are descriptions of how Flink SQL should interpret data stored in external systems; these Flink tables don’t actually store any data, they’re just metadata.

@David Anderson can you please expand on this a little bit in terms of open table format, when I create these tables - doesn't flink add metadata to backing catalog store like HMS about the table? otherwise it could be overwritten at runtime by anyone? - Maybe your answer points to the direction I should be looking at w.r.t to my questions as well. any thoughts on that? https://apache-flink.slack.com/archives/C03G7LJTS2G/p1691101487099469