{"record":{"id":"e4adc7ff812d0298","repo":"risingwavelabs/risingwave","slug":"only-es-sink-supports-struct-got","errorCode":null,"errorMessage":"Only Es sink supports struct, got {:?}: {:?}","messagePattern":"Only Es sink supports struct, got (.+?): (.+?)","errorType":"validation","errorClass":"SinkError::Remote","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/remote.rs","lineNumber":227,"sourceCode":"                    | DataType::Interval\n                    | DataType::Jsonb\n                    | DataType::Bytea => Ok(()),\n            DataType::List(list) => {\n                if is_remote_es_sink(sink_name) || matches!(list.elem(), DataType::Int16 | DataType::Int32 | DataType::Int64 | DataType::Float32 | DataType::Float64 | DataType::Varchar){\n                    Ok(())\n                } else{\n                    Err(SinkError::Remote(anyhow!(\n                        \"Remote sink only supports list<int16, int32, int64, float, double, varchar>, got {:?}: {:?}\",\n                        col.name,\n                        col.data_type,\n                    )))\n                }\n            },\n            DataType::Struct(_) => {\n                if is_remote_es_sink(sink_name){\n                    Ok(())\n                }else{\n                    Err(SinkError::Remote(anyhow!(\n                        \"Only Es sink supports struct, got {:?}: {:?}\",\n                        col.name,\n                        col.data_type,\n                    )))\n                }\n            },\n            DataType::Vector(_) |\n            DataType::Serial | DataType::Int256 | DataType::Map(_) | DataType::Variant => Err(SinkError::Remote(anyhow!(\n                            \"remote sink supports Int16, Int32, Int64, Float32, Float64, Boolean, Decimal, Time, Date, Interval, Jsonb, Timestamp, Timestamptz, Bytea, List and Varchar, (Es sink support Struct) got {:?}: {:?}\",\n                            col.name,\n                            col.data_type,\n                        )))}})?;\n\n    let jvm = Jvm::get_or_init()?;\n    let sink_param = param.to_proto();\n\n    spawn_blocking(move || -> anyhow::Result<()> {\n        execute_with_jni_env(jvm, |env| {","sourceCodeStart":209,"sourceCodeEnd":245,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/remote.rs#L209-L245","documentation":"During remote sink column validation, RisingWave rejects any column whose type is Struct unless the remote sink is an Elasticsearch (Es) sink. The JVM-based remote sink protocol can only serialize struct columns for the ES connector, so any other remote sink with a Struct column fails validation before the sink is created.","triggerScenarios":"Creating/validating a remote (JVM) sink whose schema contains a DataType::Struct column while sink_name is not recognized as a remote ES sink (is_remote_es_sink returns false), via CREATE SINK with a struct-typed column.","commonSituations":"Users sink a table/materialized view containing a nested struct (e.g. from JSON ingestion) into a non-ES remote sink like Kafka via JDBC; changing a column to STRUCT after originally targeting ES.","solutions":["Use an Elasticsearch remote sink if the data genuinely contains struct columns.","Flatten the struct column into primitive columns (e.g. via flatten or explicit field selection) before sinking.","Cast the struct column to a JSONB/VARCHAR column, which remote sinks support.","If the sink actually is ES, verify the sink name/connector matches what is_remote_es_sink expects."],"exampleFix":"-- before\nCREATE SINK s FROM mv WITH (connector = 'jdbc', ...); -- mv has STRUCT column\n-- after (flatten)\nCREATE SINK s AS SELECT (col).field_a AS field_a, (col).field_b AS field_b FROM mv WITH (connector = 'jdbc', ...);","handlingStrategy":"validation","validationCode":"-- before CREATE SINK: check for struct columns\nSELECT column_name, data_type\nFROM rw_catalog.rw_columns\nWHERE relation_id = 'your_mv_id'\n  AND data_type LIKE 'struct%';\n-- if rows returned, use an ES sink or flatten/cast those columns","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Check the sink schema for STRUCT columns when using non-ES remote sinks","Prefer flattening nested structs into primitives at the MV layer","Validate the sink plan with EXPLAIN or a dry-run before creating it"],"tags":["sink","validation","data-type","connector"],"backgroundTag":"incompatible-source-type","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}