risingwave: remove ON CONFLICT from static DDL and generator output

- Update risingwaveStaticSql to avoid ON CONFLICT clauses, aligning with
  expected seed behavior and tests.
- Adjust from-proto-generate-csv getStaticSQL for risingwave accordingly.
- Update inserter comments to reflect plain INSERT usage.
- Update RisingWave dialect tests to assert absence of ON CONFLICT.

Author: chriswessels

cmd/substreams-sink-sql/from_proto_generate_csv.go

@@ -524,17 +524,17 @@ CREATE TABLE IF NOT EXISTS "%s"._blocks_ (
 );`, schemaName, schemaName, schemaName, schemaName)
 
     case "risingwave":
-        // Matches db_proto/sql/risingwave/dialect.go risingwaveStaticSql
+        // Matches db_proto/sql/risingwave/dialect.go risingwaveStaticSql (no ON CONFLICT clauses)
         return fmt.Sprintf(`CREATE SCHEMA IF NOT EXISTS "%s";
 
 CREATE TABLE IF NOT EXISTS "%s"._sink_info_ (
     schema_hash VARCHAR PRIMARY KEY
-) ON CONFLICT DO NOTHING;
+);
 
 CREATE TABLE IF NOT EXISTS "%s"._cursor_ (
     name VARCHAR PRIMARY KEY,
     cursor VARCHAR
-) ON CONFLICT OVERWRITE;
+);
 
 CREATE TABLE IF NOT EXISTS "%s"._blocks_ (
     number INTEGER PRIMARY KEY,

db_proto/sql/risingwave/accumulator_inserter.go

@@ -46,10 +46,8 @@ func (i *AccumulatorInserter) init(database *Database) error {
 		query: fmt.Sprintf("INSERT INTO %s (number, hash, timestamp) VALUES ", tableName(database.schema.Name, "_blocks_")),
 	}
 
-	// RisingWave doesn't support PostgreSQL's ON CONFLICT syntax
-	// We'll use a simple INSERT for the cursor, relying on the table to handle conflicts
-	// The _cursor_ table should be created with appropriate ON CONFLICT behavior
-	cursorQuery := fmt.Sprintf("INSERT INTO %s (name, cursor) VALUES ($1, $2)", tableName(database.schema.Name, "_cursor_"))
+    // RisingWave: use a simple INSERT for the cursor; no ON CONFLICT clause in DDL
+    cursorQuery := fmt.Sprintf("INSERT INTO %s (name, cursor) VALUES ($1, $2)", tableName(database.schema.Name, "_cursor_"))
 	cs, err := database.db.Prepare(cursorQuery)
 	if err != nil {
 		return fmt.Errorf("preparing statement %q: %w", cursorQuery, err)

db_proto/sql/risingwave/dialect.go

@@ -18,12 +18,12 @@ const risingwaveStaticSql = `
 
 	CREATE TABLE IF NOT EXISTS "%s"._sink_info_ (
 		schema_hash VARCHAR PRIMARY KEY
-	) ON CONFLICT DO NOTHING;
+	);
 
 	CREATE TABLE IF NOT EXISTS "%s"._cursor_ (
 		name VARCHAR PRIMARY KEY,
 		cursor VARCHAR
-	) ON CONFLICT OVERWRITE;
+	);
 
 	CREATE TABLE IF NOT EXISTS "%s"._blocks_ (
 		number INTEGER PRIMARY KEY,

db_proto/sql/risingwave/dialect_test.go

@@ -85,12 +85,9 @@ func TestDialectRisingwave_CreateTableStaticSql(t *testing.T) {
     assert.Contains(t, sql, "_cursor_")
     assert.Contains(t, sql, "name varchar primary key")
     assert.Contains(t, sql, "cursor varchar")
-    assert.Contains(t, sql, "on conflict overwrite")
 
-    // Check conflict policies per table
-    // _sink_info_: DO NOTHING
-    assert.Contains(t, sql, "_sink_info_")
-    assert.Contains(t, sql, "on conflict do nothing")
+    // No ON CONFLICT clauses expected in static DDL
+    assert.NotContains(t, sql, "on conflict")
 
 	// Check _blocks_ table
 	assert.Contains(t, sql, "_blocks_")

db_proto/sql/risingwave/row_inserter.go

@@ -55,7 +55,7 @@ func (i *RowInserter) init(database *Database) error {
 	}
 	insertStatements["_blocks_"] = bs
 
-    // Use plain INSERT; table-level ON CONFLICT policy (OVERWRITE) is defined in CREATE TABLE
+    // Use plain INSERT; RisingWave static DDL does not use ON CONFLICT clauses.
     insertQueries["_cursor_"] = fmt.Sprintf("INSERT INTO %s (name, cursor) VALUES ($1, $2)", tableName(database.schema.Name, "_cursor_"))
 	cs, err := database.db.Prepare(insertQueries["_cursor_"])
 	if err != nil {