risingwavelabs/risingwave · error · SinkError::Iceberg

partition spec {} not found

Error message

partition spec {} not found

What it means

resolve_partition_spec looks up a PartitionSpec by its id in the Iceberg table metadata. If the table has no spec with the given id (e.g. the referenced spec was replaced/removed or the id comes from stale metadata), it returns this SinkError::Iceberg error.

Source

Thrown at src/connector/src/sink/iceberg/writer.rs:1019

            let truncated = truncate_datafile(f);
            Ok(SerializedDataFile::try_from(
                truncated,
                partition_type,
                format_version,
            )?)
        })
        .collect()
}

/// Resolves a partition spec by id from the table metadata, with a consolidated
/// error message. Shared spec-lookup mechanic for the pk-index merger, commit
/// coordinator, and sink commit paths.
pub fn resolve_partition_spec(table: &Table, spec_id: i32) -> Result<PartitionSpecRef> {
    table
        .metadata()
        .partition_spec_by_id(spec_id)
        .cloned()
        .ok_or_else(|| SinkError::Iceberg(anyhow!("partition spec {} not found", spec_id)))
}

/// Resolves the partition [`StructType`] for the given `spec_id` against `schema`.
///
/// `schema` is passed explicitly (rather than read from the table) so callers can
/// preserve their chosen schema, and `spec_id` is passed explicitly so callers can
/// preserve their chosen spec (e.g. a file's own spec vs. the default spec).
pub fn resolve_partition_type(table: &Table, spec_id: i32, schema: &Schema) -> Result<StructType> {
    resolve_partition_spec(table, spec_id)?
        .partition_type(schema)
        .map_err(|e| SinkError::Iceberg(anyhow!(e)))
}

/// Truncate large column statistics from `DataFile` BEFORE serialization.
///
/// This function directly modifies `DataFile`'s `lower_bounds` and `upper_bounds`
/// to remove entries that exceed `MAX_COLUMN_STAT_SIZE`.
///

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the spec_id against the current table metadata before committing (e.g. resolve via metadata.default_partition_spec_id() or partition_specs keys)
2. Recreate or rebuild the sink so its stored spec id matches the current table
3. Ensure commits read the same table snapshot/metadata that produced the data files

Example fix

// before
let spec = resolve_partition_spec(&table, file_spec_id)?;
// after
let spec_id = table.metadata().default_partition_spec_id();
let spec = resolve_partition_spec(&table, spec_id)?;

fn has_spec(table: &Table, spec_id: i32) -> bool {
    table.metadata().partition_spec_by_id(spec_id).is_some()
}
// guard: if !has_spec(&table, spec_id) { return Err(...); }

let spec = resolve_partition_spec(&table, spec_id).context("resolving partition spec for commit")?;

When it happens

Trigger: Calling resolve_partition_spec(table, spec_id) with a spec_id that is absent from table.metadata().partition_specs — typically a spec_id persisted in sink state or data files that no longer exists in the current table metadata after a partition-spec change.

Common situations: Table schema/partition evolution between sink creation and commit; restoring sink state referencing an old spec id; pointing the sink at a different table than the one that produced the data files.

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

• Can't create iceberg sink write result from empty data!
• iceberg sink metadata should be an object
• schema_id should be a u64
• Fail to convert version_hint from utf8 to string: {}
• Can't find schema by id {}

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