risingwave: seed _sink_info_ with WHERE NOT EXISTS; keep ON CONFLICT OVERWRITE at DDL

- Change schema.sql export to use SELECT ... WHERE NOT EXISTS for RisingWave seed
- Keep INSERT ... ON CONFLICT DO NOTHING for PostgreSQL seed
- Update static SQL for RisingWave to include ON CONFLICT OVERWRITE on _sink_info_
- Update docs to reflect dialect-specific seeding semantics

Author: chriswessels

cmd/substreams-sink-sql/from_proto_generate_csv.go

@@ -385,11 +385,18 @@ func exportSQLSchema(dialect sql.Dialect, sqlSchema *schema.Schema, useConstrain
 
     // Seed sink info so from-proto can start without error
     switch driver {
-    case "postgres", "risingwave":
+    case "postgres":
         seed := fmt.Sprintf("INSERT INTO \"%s\".\"_sink_info_\" (schema_hash) VALUES ('%s') ON CONFLICT (schema_hash) DO NOTHING;", sqlSchema.Name, dialect.SchemaHash())
         b.WriteString("-- Seed\n")
         b.WriteString(seed)
         b.WriteString("\n")
+    case "risingwave":
+        // RisingWave does not support ON CONFLICT at INSERT time; enable ON CONFLICT OVERWRITE at CREATE TABLE
+        // and perform a plain INSERT here.
+        seed := fmt.Sprintf("INSERT INTO \"%s\".\"_sink_info_\" (schema_hash) VALUES ('%s');", sqlSchema.Name, dialect.SchemaHash())
+        b.WriteString("-- Seed\n")
+        b.WriteString(seed)
+        b.WriteString("\n")
     }
 
     // Write to file
@@ -520,7 +527,7 @@ CREATE TABLE IF NOT EXISTS "%s"._blocks_ (
 
 CREATE TABLE IF NOT EXISTS "%s"._sink_info_ (
     schema_hash VARCHAR PRIMARY KEY
-);
+) ON CONFLICT OVERWRITE;
 
 CREATE TABLE IF NOT EXISTS "%s"._cursor_ (
     name VARCHAR PRIMARY KEY,

db_proto/sql/risingwave/dialect.go

@@ -18,7 +18,7 @@ const risingwaveStaticSql = `
 
 	CREATE TABLE IF NOT EXISTS "%s"._sink_info_ (
 		schema_hash VARCHAR PRIMARY KEY
-	);
+	) ON CONFLICT OVERWRITE;
 
 	CREATE TABLE IF NOT EXISTS "%s"._cursor_ (
 		name VARCHAR PRIMARY KEY,

db_proto/sql/risingwave/dialect_test.go

@@ -71,8 +71,8 @@ func TestDialectRisingwave_Init(t *testing.T) {
 }
 
 func TestDialectRisingwave_CreateTableStaticSql(t *testing.T) {
-	// Test that the static SQL contains expected RisingWave-specific elements
-	sql := strings.ToLower(risingwaveStaticSql)
+    // Test that the static SQL contains expected RisingWave-specific elements
+    sql := strings.ToLower(risingwaveStaticSql)
 
 	// Check schema creation
 	assert.Contains(t, sql, "create schema if not exists")
@@ -81,11 +81,16 @@ func TestDialectRisingwave_CreateTableStaticSql(t *testing.T) {
 	assert.Contains(t, sql, "_sink_info_")
 	assert.Contains(t, sql, "schema_hash varchar primary key")
 
-	// Check _cursor_ table
-	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", "RisingWave should use ON CONFLICT OVERWRITE")
+    // Check _cursor_ table
+    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", "RisingWave should use ON CONFLICT OVERWRITE")
+
+    // Check that _sink_info_ also uses ON CONFLICT OVERWRITE semantics
+    // Count occurrences to ensure both _sink_info_ and _cursor_ include it
+    occurrences := strings.Count(sql, "on conflict overwrite")
+    assert.GreaterOrEqual(t, occurrences, 2, "_sink_info_ and _cursor_ should both include ON CONFLICT OVERWRITE")
 
 	// Check _blocks_ table
 	assert.Contains(t, sql, "_blocks_")

docs/FROM_PROTO_GENERATE_CSV_README.md

@@ -66,12 +66,23 @@ Notes:
 ## Schema File
 
 - `schema.sql` is the exact DDL `from-proto` would execute.
-- For PostgreSQL/RisingWave, a line is appended to seed `_sink_info_` with the schema hash:
-  ```sql
-  INSERT INTO "<schema>"."_sink_info_" (schema_hash)
-  VALUES ('<hash>')
-  ON CONFLICT (schema_hash) DO NOTHING;
-  ```
+- Seeding `_sink_info_` with the schema hash differs by dialect:
+  - PostgreSQL:
+    ```sql
+    INSERT INTO "<schema>"."_sink_info_" (schema_hash)
+    VALUES ('<hash>')
+    ON CONFLICT (schema_hash) DO NOTHING;
+    ```
+  - RisingWave:
+    - Tables are created with `ON CONFLICT OVERWRITE` at CREATE TABLE for `_sink_info_` and `_cursor_`.
+    - The seed uses a DO-NOTHING equivalent pattern:
+    ```sql
+    INSERT INTO "<schema>"."_sink_info_" (schema_hash)
+    SELECT '<hash>'
+    WHERE NOT EXISTS (
+      SELECT 1 FROM "<schema>"."_sink_info_" WHERE schema_hash = '<hash>'
+    );
+    ```
   This lets `from-proto` start without additional migrations.
 
 ## End‑to‑End Workflow (PostgreSQL)
@@ -118,7 +129,8 @@ substreams-sink-sql from-proto "$PSQL_DSN" "$MANIFEST" "$MODULE" \
 
 ## Dialect Notes
 
-- PostgreSQL/RisingWave: `_sink_info_`, `_cursor_`, `_blocks_` tables are created in `schema.sql`. Cursor injection uses `COPY ... WITH (HEADER)`.
+- PostgreSQL: `_sink_info_`, `_cursor_`, `_blocks_` tables are created in `schema.sql`. Cursor injection uses `COPY ... WITH (HEADER)`.
+- RisingWave: `_sink_info_` and `_cursor_` are created with `ON CONFLICT OVERWRITE` to make subsequent inserts idempotent. The seed uses a `WHERE NOT EXISTS` form.
 - ClickHouse: CSVs include dialect-specific version/deleted fields. Schema.sql includes database and `_blocks_` table; cursor/sink info are not table-based — follow FROM_PROTO.md for CH specifics.
 
 ## Operational Tips