risingwave: make DDL column order consistent with from-proto inserts

- In CREATE TABLE, emit system columns (block_number, block_timestamp) before PK
- Keeps inline PRIMARY KEY for explicit PK; composite PK unchanged
- Adds test to assert block_number appears before id in emitted SQL

Author: chriswessels

cmd/substreams-sink-sql/from_proto_generate_csv_test.go

@@ -147,6 +147,14 @@ func TestFromProtoGenerateCSVCompatibility_RisingWave(t *testing.T) {
     assert.NotContains(t, fullSQL, "_block_number_")
     assert.NotContains(t, fullSQL, "_block_timestamp_")
 
+    // Ensure ordering: system columns appear before id column
+    // (block_number before id)
+    idxBN := strings.Index(fullSQL, "block_number")
+    idxID := strings.Index(fullSQL, "id BIGINT PRIMARY KEY")
+    if idxBN >= 0 && idxID >= 0 {
+        assert.Less(t, idxBN, idxID, "block_number should appear before id in column list")
+    }
+
     // User columns with correct types (id unquoted, others quoted by dialect)
     assert.Contains(t, fullSQL, "id BIGINT PRIMARY KEY")
     assert.Contains(t, fullSQL, `"name" CHARACTER VARYING`)

db_proto/sql/risingwave/dialect.go

@@ -71,27 +71,27 @@ func (d *DialectRisingwave) init() error {
 }
 
 func (d *DialectRisingwave) createTable(table *schema.Table) error {
-	var sb strings.Builder
-	addedColumns := make(map[string]struct{})
-
-	tableName := d.FullTableName(table)
-
-	sb.WriteString(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (", tableName))
-
-	// Add primary key if it exists
-	var primaryKeyFieldName string
-	if table.PrimaryKey != nil {
-		pk := table.PrimaryKey
-		primaryKeyFieldName = pk.Name
-		sb.WriteString(fmt.Sprintf("%s %s PRIMARY KEY,", pk.Name, MapFieldType(pk.FieldDescriptor)))
-		addedColumns[pk.Name] = struct{}{}
-	}
-
-	// Always add block metadata columns
-	sb.WriteString(" block_number INTEGER,")
-	sb.WriteString(" block_timestamp TIMESTAMP WITH TIME ZONE,")
-	addedColumns["block_number"] = struct{}{}
-	addedColumns["block_timestamp"] = struct{}{}
+    var sb strings.Builder
+    addedColumns := make(map[string]struct{})
+
+    tableName := d.FullTableName(table)
+
+    sb.WriteString(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (", tableName))
+
+    // Always add block metadata columns first (align column order across dialects)
+    sb.WriteString(" block_number INTEGER,")
+    sb.WriteString(" block_timestamp TIMESTAMP WITH TIME ZONE,")
+    addedColumns["block_number"] = struct{}{}
+    addedColumns["block_timestamp"] = struct{}{}
+
+    // Add primary key column next if it exists (inline PRIMARY KEY)
+    var primaryKeyFieldName string
+    if table.PrimaryKey != nil {
+        pk := table.PrimaryKey
+        primaryKeyFieldName = pk.Name
+        sb.WriteString(fmt.Sprintf("%s %s PRIMARY KEY,", pk.Name, MapFieldType(pk.FieldDescriptor)))
+        addedColumns[pk.Name] = struct{}{}
+    }
 
 	// Add parent key for child tables
 	var parentKeyColumns []string