risingwavelabs/risingwave · error · anyhow::Error

Cannot find {}

Error message

Cannot find {}

What it means

`get_index_column_index` resolves the `index_column` name to its position in the sink schema. If no schema field matches the configured name, it returns 'Cannot find index_column' (the option's constant name). Note the column is normally also checked in `validate_config` with an unwrap, so this usually surfaces when the two checks are bypassed or the lookup happens against a different schema.

Source

Thrown at src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_config.rs:297

                return Err(SinkError::Config(anyhow!(
                    "please ensure the data type of {} is varchar.",
                    routing_column
                )));
            }
        }
        Ok(())
    }

    pub fn get_index_column_index(&self, schema: &Schema) -> Result<Option<usize>> {
        let index_column_idx = self
            .index_column
            .as_ref()
            .map(|n| {
                schema
                    .fields()
                    .iter()
                    .position(|s| &s.name == n)
                    .ok_or_else(|| anyhow!("Cannot find {}", ES_OPTION_INDEX_COLUMN))
            })
            .transpose()?;
        Ok(index_column_idx)
    }

    pub fn get_routing_column_index(&self, schema: &Schema) -> Result<Option<usize>> {
        let routing_column_idx = self
            .routing_column
            .as_ref()
            .map(|n| {
                schema
                    .fields()
                    .iter()
                    .position(|s| &s.name == n)
                    .ok_or_else(|| anyhow!("Cannot find {}", ES_OPTION_ROUTING_COLUMN))
            })
            .transpose()?;
        Ok(routing_column_idx)

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the exact column name in the sink schema and use it verbatim in `index_column`
2. Check for case differences or accidental whitespace in the WITH option
3. Query the schema (e.g. DESC the materialized view) to list available columns
4. If the column was renamed upstream, update or recreate the sink

Example fix

// before
WITH (connector='elasticsearch', index_column='Topic')
// after
WITH (connector='elasticsearch', index_column='topic')

fn check_column_exists(schema: &Schema, name: &str) -> Result<(), String> {
    if schema.fields().iter().any(|f| f.name == name) {
        Ok(())
    } else {
        Err(format!("column '{}' not in schema; available: {:?}",
            name, schema.fields().iter().map(|f| f.name.clone()).collect::<Vec<_>>()))
    }
}

When it happens

Trigger: Sink creation path `new` -> `get_index_column_index` with `index_column` set to a name that does not exist verbatim in the schema fields (case mismatch, extra whitespace, renamed column).

Common situations: Case-sensitive name mismatch ('UserID' vs 'user_id'), column renamed upstream after config was written, quoting artifacts in the WITH option value.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

• Cannot find {}
• please set only one of the 'index_column' or 'index' propert
• please ensure the data type of {} is varchar.
• no value find in sink schema, index is 0
• `{}` must be {}, or {}

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