risingwave: align docs and implementation with latest risingwave behaviour for ON CONFLICT

Author: chriswessels

cmd/substreams-sink-sql/from_proto_generate_csv.go

@@ -391,10 +391,10 @@ func exportSQLSchema(dialect sql.Dialect, sqlSchema *schema.Schema, useConstrain
         b.WriteString(seed)
         b.WriteString("\n")
     case "risingwave":
-        // RisingWave: emulate DO NOTHING semantics with WHERE NOT EXISTS
+        // RisingWave: rely on table-level ON CONFLICT policy (DO NOTHING) defined in CREATE TABLE
         seed := fmt.Sprintf(
-            "INSERT INTO \"%s\".\"_sink_info_\" (schema_hash) SELECT '%s' WHERE NOT EXISTS (SELECT 1 FROM \"%s\".\"_sink_info_\" WHERE schema_hash = '%s');",
-            sqlSchema.Name, dialect.SchemaHash(), sqlSchema.Name, dialect.SchemaHash(),
+            "INSERT INTO \"%s\".\"_sink_info_\" (schema_hash) VALUES ('%s');",
+            sqlSchema.Name, dialect.SchemaHash(),
         )
         b.WriteString("-- Seed\n")
         b.WriteString(seed)
@@ -529,7 +529,7 @@ CREATE TABLE IF NOT EXISTS "%s"._blocks_ (
 
 CREATE TABLE IF NOT EXISTS "%s"._sink_info_ (
     schema_hash VARCHAR PRIMARY KEY
-) ON CONFLICT OVERWRITE;
+) ON CONFLICT DO NOTHING;
 
 CREATE TABLE IF NOT EXISTS "%s"._cursor_ (
     name VARCHAR PRIMARY KEY,

cmd/substreams-sink-sql/from_proto_generate_csv_test.go

@@ -185,13 +185,13 @@ func TestSchemaSeed_RisingWave(t *testing.T) {
     err = exportSQLSchema(dialect, testSchema, true, "risingwave", f.Name())
     require.NoError(t, err)
 
-    // Read and assert WHERE NOT EXISTS pattern
+    // Read and assert RisingWave plain INSERT (table-level policy is in DDL)
     data, err := os.ReadFile(f.Name())
     require.NoError(t, err)
     sqlText := string(data)
 
-    assert.Contains(t, sqlText, "INSERT INTO \"test_schema\".\"_sink_info_\" (schema_hash) SELECT ")
-    assert.Contains(t, sqlText, "WHERE NOT EXISTS (")
+    assert.Contains(t, sqlText, "INSERT INTO \"test_schema\".\"_sink_info_\" (schema_hash) VALUES (")
+    assert.NotContains(t, sqlText, "ON CONFLICT")
 }
 
 // TestCSVColumnOrder ensures CSV columns are in the exact order from-proto expects

db_proto/sql/risingwave/database.go

@@ -279,11 +279,12 @@ func (d *Database) FetchSinkInfo(schemaName string) (*sql.SinkInfo, error) {
 }
 
 func (d *Database) StoreSinkInfo(schemaName string, schemaHash string) error {
-	_, err := d.execSql(fmt.Sprintf("INSERT INTO %s._sink_info_ (schema_hash) VALUES ($1)", schemaName), schemaHash)
-	if err != nil {
-		return fmt.Errorf("storing schema hash: %w", err)
-	}
-	return nil
+    // Use plain INSERT; table-level ON CONFLICT DO NOTHING is defined in CREATE TABLE
+    _, err := d.execSql(fmt.Sprintf("INSERT INTO %s._sink_info_ (schema_hash) VALUES ($1)", schemaName), schemaHash)
+    if err != nil {
+        return fmt.Errorf("storing schema hash: %w", err)
+    }
+    return nil
 }
 
 func (d *Database) UpdateSinkInfoHash(schemaName string, newHash string) error {

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;
+	) ON CONFLICT DO NOTHING;
 
 	CREATE TABLE IF NOT EXISTS "%s"._cursor_ (
 		name VARCHAR PRIMARY KEY,

db_proto/sql/risingwave/dialect_test.go

@@ -85,12 +85,12 @@ 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", "RisingWave should use ON CONFLICT OVERWRITE")
+    assert.Contains(t, sql, "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 conflict policies per table
+    // _sink_info_: DO NOTHING
+    assert.Contains(t, sql, "_sink_info_")
+    assert.Contains(t, sql, "on conflict do nothing")
 
 	// Check _blocks_ table
 	assert.Contains(t, sql, "_blocks_")

db_proto/sql/risingwave/row_inserter.go

@@ -55,9 +55,8 @@ func (i *RowInserter) init(database *Database) error {
 	}
 	insertStatements["_blocks_"] = bs
 
-	// RisingWave doesn't support PostgreSQL's ON CONFLICT syntax in INSERT statements
-	// The _cursor_ table is created with ON CONFLICT OVERWRITE to handle this automatically
-	insertQueries["_cursor_"] = fmt.Sprintf("INSERT INTO %s (name, cursor) VALUES ($1, $2)", tableName(database.schema.Name, "_cursor_"))
+    // Use plain INSERT; table-level ON CONFLICT policy (OVERWRITE) is defined in CREATE TABLE
+    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 {
 		return fmt.Errorf("preparing statement %q: %w", insertQueries["_cursor_"], err)

docs/FROM_PROTO_GENERATE_CSV_README.md

@@ -74,16 +74,14 @@ Notes:
     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:
+    - Conflict policy is defined at CREATE TABLE (table-level), not per INSERT.
+    - We set: `_sink_info_` → `ON CONFLICT DO NOTHING`; `_cursor_` → `ON CONFLICT OVERWRITE`.
+    - The seed uses a plain INSERT and relies on the table-level policy:
     ```sql
     INSERT INTO "<schema>"."_sink_info_" (schema_hash)
-    SELECT '<hash>'
-    WHERE NOT EXISTS (
-      SELECT 1 FROM "<schema>"."_sink_info_" WHERE schema_hash = '<hash>'
-    );
+    VALUES ('<hash>');
     ```
-  This lets `from-proto` start without additional migrations.
+  This lets `from-proto` start without additional migrations and keeps the seed idempotent.
 
 ## End‑to‑End Workflow (PostgreSQL)
 
@@ -130,7 +128,10 @@ substreams-sink-sql from-proto "$PSQL_DSN" "$MANIFEST" "$MODULE" \
 ## Dialect Notes
 
 - 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.
+- RisingWave: Conflict policy is defined on CREATE TABLE (not per INSERT). We set:
+  - `_sink_info_`: `ON CONFLICT DO NOTHING` (idempotent seed)
+  - `_cursor_`: `ON CONFLICT OVERWRITE` (upsert semantics for the single cursor row)
+  The seed is a plain `INSERT`, relying on table-level policies.
 - 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