risingwavelabs/risingwave · error · ConnectorError

unsupported primary key data type

Error message

unsupported primary key data type: {}

What it means

When performing a snapshot read, the connector must convert the previous primary-key value (sentinel row) into a MySQL query parameter. The match on the upstream column type has no arm for that type, so it bails with 'unsupported primary key data type'. It signals the table's PK uses a data type this CDC connector does not yet support for key-range pagination.

Solutions

  1. Change the upstream table's primary key to a supported type (INT, BIGINT, VARCHAR, TIMESTAMP, etc.).
  2. If you control the schema, add a surrogate auto-increment integer PK and keep the exotic column as a unique key.
  3. Create the RisingWave CDC table on a table with a conventional PK type.
  4. Extend the match in `snapshot_read_inner` to support the missing MySQL type if you maintain the connector.

Example fix

// before: exotic JSON primary key in MySQL
CREATE TABLE events (payload JSON PRIMARY KEY, ...);
// after: surrogate integer PK, JSON as unique key
CREATE TABLE events (id BIGINT AUTO_INCREMENT PRIMARY KEY, payload JSON, UNIQUE KEY (payload));
Defensive patterns

Strategy: validation

Validate before calling

// Check the upstream PK column type is supported before enabling CDC
SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS
 WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND COLUMN_KEY = 'PRI';
-- DATA_TYPE must be one of: int, bigint, smallint, varchar, char, timestamp, date

Try / catch

match snapshot_read(...).await {
    Err(e) if e.to_string().contains("unsupported primary key data type") => {
        // fail fast and surface a schema-fix hint to the operator
    },
    r => r?,
}

Prevention

When it happens

Trigger: `snapshot_read_inner` (called from `snapshot_read`) reaches the `_ =>` arm of the PK value conversion match — i.e. the previous PK value is non-null but its MySQL type is not one of the explicitly handled numeric/string/timestamp types.

Common situations: Tables with exotic PK types (BLOB, JSON, BIT, ENUM, SET, large DECIMAL) used as primary keys; using a generated or binary column as PK in MySQL; schema drift changing a PK column's type to an unsupported one.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/84a53397a76fa05a. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/source/cdc/external/mysql.rs:759

                                    Value::from(int64_val)
                                }
                            }
                            DataType::Float32 => Value::from(value.into_float32().into_inner()),
                            DataType::Float64 => Value::from(value.into_float64().into_inner()),
                            DataType::Varchar => Value::from(String::from(value.into_utf8())),
                            DataType::Date => Value::from(value.into_date().0),
                            DataType::Time => Value::from(value.into_time().0),
                            DataType::Timestamp => Value::from(value.into_timestamp().0),
                            DataType::Decimal => Value::from(value.into_decimal().to_string()),
                            DataType::Timestamptz => {
                                // Convert timestamptz to NaiveDateTime for MySQL TIMESTAMP comparison
                                // MySQL expects NaiveDateTime for TIMESTAMP parameters
                                let ts = value.into_timestamptz();
                                let datetime_utc = ts.to_datetime_utc();
                                let naive_datetime = datetime_utc.naive_utc();
                                Value::from(naive_datetime)
                            }
                            _ => bail!("unsupported primary key data type: {}", ty),
                        };
                        ConnectorResult::Ok((pk.to_lowercase(), val))
                    } else {
                        bail!("primary key {} cannot be null", pk);
                    }
                })
                .try_collect::<_, _, ConnectorError>()?;

            tracing::debug!("snapshot read params: {:?}", &params);
            let rs_stream = sql
                .with(Params::from(params))
                .stream::<mysql_async::Row, _>(&mut conn)
                .await?;

            let row_stream = rs_stream.map(|row| {
                // convert mysql row into OwnedRow
                let mut row = row?;
                mysql_row_to_owned_row_with_strict_pk(&mut row, &self.rw_schema, &self.pk_indices)

View on GitHub (pinned to 6469eb736d)