risingwavelabs/risingwave · error · SinkError::Config

collection.name.field {} must not be equal to the primary ke

Error message

collection.name.field {} must not be equal to the primary key field

What it means

In upsert mode, the primary key is written as the document's `_id`; the dynamic collection-name column must survive in the document per row, so it cannot be part of the primary key (otherwise it would be consumed into `_id`, breaking per-collection routing). `validate` rejects this combination for non-append-only sinks.

Source

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

                        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
                )));
            }
        }

        Ok(())
    }

    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        Ok(MongodbSinkWriter::new(
            format!("{}-{}", writer_param.executor_id, self.param.sink_name),
            self.config.clone(),
            self.schema.clone(),
            self.pk_indices.clone(),
            self.is_append_only,
        )
        .await?

View on GitHub (pinned to 6469eb736d)

Solutions

1. Remove the collection-name column from `primary_key` and pick a different PK (e.g., add a synthetic id column and use `primary_key='id'`).
2. If the collection name column must be the key, keep the sink append-only (no updates) so the constraint doesn't apply.
3. Duplicate the value: route on a plain string column and use a different column set as PK.

Example fix

-- before
CREATE SINK s FROM mv INTO mongodb WITH (
  connector='mongodb', collection='db.c',
  primary_key='tenant', collection.name.field='tenant'
);
-- after
CREATE SINK s FROM mv INTO mongodb WITH (
  connector='mongodb', collection='db.c',
  primary_key='id', collection.name.field='tenant'
);

const pk = new Set(sinkOptions.primary_key.split(',').map(s => s.trim()));
const cf = options['collection.name.field'];
if (!isAppendOnly && cf != null && pk.has(cf)) {
  throw new Error("collection.name.field must not be part of primary_key in upsert mongodb sink");
}

When it happens

Trigger: CREATE SINK into mongodb (upsert mode) where `collection.name.field` references a column that is also listed in `primary_key`.

Common situations: Routing by tenant column while also using tenant as part of the upsert PK — a common but unsupported combo here.

Understand the failure class

Background: Conflicting config options: "cannot be used together" — configuration validation errors across open-source libraries — this error's family across 162 libraries.

Related errors

• Primary key not defined for upsert mongodb sink (please defi
• Primary key not defined for upsert clickhouse sink (please d
• primary key must be specified for upsert iceberg sink
• VARIANT column `{}` cannot be used as the primary key of an
• collection.name.field must be specified when collection.name

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