risingwavelabs/risingwave · error · SinkError::Config

collection.name.field {} not found

Error message

collection.name.field {} not found

What it means

For dynamic collection routing, `collection.name.field` must name an existing column of the sink's schema; `validate` searches the schema fields and fails if none matches. This ensures the sink can resolve the per-row collection name at write time.

Source

Thrown at src/connector/src/sink/mongodb.rs:351

                "collection.name.field must be specified when collection.name.field.drop is enabled"
            )));
        }

        // checking dynamic collection name settings
        if let Some(coll_field) = &self.config.collection_name_field {
            let fields = self.schema.fields();

            let coll_field_index = fields
                .iter()
                .enumerate()
                .find_map(|(index, field)| {
                    if &field.name == coll_field {
                        Some(index)
                    } else {
                        None
                    }
                })
                .ok_or(SinkError::Config(anyhow!(
                    "collection.name.field {} not found",
                    coll_field
                )))?;

            if fields[coll_field_index].data_type() != risingwave_common::types::DataType::Varchar {
                return Err(SinkError::Config(anyhow!(
                    "the type of collection.name.field {} must be varchar",
                    coll_field
                )));
            }

            if !self.is_append_only && self.pk_indices.contains(&coll_field_index) {
                return Err(SinkError::Config(anyhow!(
                    "collection.name.field {} must not be equal to the primary key field",
                    coll_field
                )));
            }
        }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set `collection.name.field` to an exact existing column name (case-sensitive).
2. If the routing key doesn't exist, add it to the upstream query (e.g., a computed column) and refresh the sink.
3. Re-check the current schema with `DESCRIBE` on the source/materialized view.

Example fix

-- before
CREATE SINK s FROM mv INTO mongodb WITH (
  connector='mongodb', collection='db.c', collection.name.field='coll'
); -- mv has column `coll_name`
-- after
CREATE SINK s FROM mv INTO mongodb WITH (
  connector='mongodb', collection='db.c', collection.name.field='coll_name'
);

const collField = options['collection.name.field'];
if (collField != null && !schema.fields.some(f => f.name === collField)) {
  throw new Error(`collection.name.field '${collField}' not found in sink schema`);
}

When it happens

Trigger: CREATE SINK into mongodb with `collection.name.field='X'` where no column named `X` exists in the sink's output schema (typo, renamed column, wrong case).

Common situations: Typos or case-sensitivity mismatches; upstream schema changed (column renamed/dropped) after the sink config was written; confusing the field name with the collection name.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• collection.name.field must be specified when collection.name
• the type of collection.name.field {} must be varchar
• collection.name.field {} must not be equal to the primary ke
• If you want to use upsert, please set the keysType of doris
• Can't find data

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