Integration of Concurrent Identity Columns (CIC)

[Rob2U]

Please describe why this is necessary.

Currently, the kernel does not support identity columns. The problem of regular identity columns is that two concurrent "inserters" race leading to a failure and retry of the losing one. Both try to modify the "highWaterMark" metadata field of the identity column.
Concurrent Identity Columns (CIC) circumvent this problem. They rely on an external sequence service to reserve disjoint ranges of values to insert into their identity columns (for more information see "Additional context").
Implementing kernel support would allow connectors to use CIC without reimplementing the protocol logic.

Describe the functionality you are proposing.

The connectors should see the following workflows:

• Create: The connector mints a sequence ID for each identity column and asks Kernel to add the required CIC metadata. After the table creation commits, the connector registers the sequences with the external service via a single synchronous method call.
• Write: The connector creates an identity-column writer from the table schema, ensures enough values are reserved, and asks Kernel to fill the identity columns. The resulting batch is written through the normal append path.

Additional context

The accompanying RFC of this feature (for delta) can be found here and the (general) design-document here.

This is the accompanying prototype implementation: #3290

[Rob2U]

In my current prototype the kernel holds mostly primitives e.g. cic_column for adding CIC to the schema.

A connector would mainly use register_identity_sequences and IdentityColumnWriter. Both live in the delta-kernel-unity-catalog crate. Because they depend on the unity-catalog-delta-rest-client.

This is where I have two questions:

1. The IdentityColumnWriter implements quite a lot of behavior that can safely live inside the kernel. Therefore, I introduce this IdentitySequenceState. Is this the right approach (its interface is very much designed to fit the writers functionality).
2. The current connector interface looks like this:

Create

Mint a sequenceId per identity column, stamp them into the schema, commit CREATE TABLE, and only
then register the sequences with the service.

// 1. Mint a sequence_id per identity column (the caller owns minting).
let infos = [
    IdentityColumnInfo {
        column_name: "id".into(),
        sequence_id: Uuid::new_v4().to_string(),
        start: 1,
        step: 1,
        allow_explicit_insert: false,
    },
    IdentityColumnInfo {
        column_name: "row_id".into(),
        sequence_id: Uuid::new_v4().to_string(),
        start: 1000,
        step: 10,
        allow_explicit_insert: false,
    },
];

// 2. Stamp the ids into the schema, then commit CREATE TABLE *first*.
let schema = Arc::new(StructType::try_new(vec![
    cic_column(&infos[0].column_name, &infos[0].sequence_id, infos[0].start, infos[0].step),
    StructField::new("payload", DataType::STRING, true),
    cic_column(&infos[1].column_name, &infos[1].sequence_id, infos[1].start, infos[1].step),
])?);

create_table(table_path, schema, "cic-demo/0.1")
    .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
    .commit(engine.as_ref())?;

// 3. Only after the commit, register the sequences (one batched RPC). If the commit had failed,
//    no sequence would have been orphaned in the catalog.
register_identity_sequences(client.as_ref(), table_id, &infos).await?;

Write

Reload the snapshot (the CIC metadata is now on its schema), build one IdentityColumnWriter,
ensure_available before filling, then write the filled batch as an ordinary append. The
engine-side batch has only the non-identity columns. The writer fills id and row_id and
returns every column in schema order, ready to write.

let snapshot = Snapshot::builder_for(table_url).build(engine.as_ref())?;
let schema = snapshot.schema();

let writer = IdentityColumnWriter::new(&schema, client.clone(), table_id)?;

const BATCH_ROWS: u64 = 3;
// Ensure that the fill of all rows can be covered by the current reservations.
writer.ensure_available(BATCH_ROWS).await?;

// Kernel fills the identity columns.
let payload: ArrayRef = Arc::new(StringArray::from(vec!["hello", "world", "!"]));
let input = RecordBatch::try_new(
    Arc::new(ArrowSchema::new(vec![ArrowField::new("payload", ArrowDataType::Utf8, true)])),
    vec![payload],
)?;
// fill_engine_batch fills the identity columns and returns every column in `schema` order, ready
// to write. The engine's (synchronous) evaluation handler places the columns. It returns
// Box<dyn EngineData>, but the default engine's write_parquet wants a concrete ArrowEngineData,
// so recover it.
let filled = writer.fill_engine_batch(
    engine.evaluation_handler().as_ref(),
    &ArrowEngineData::new(input),
    &schema,
)?;
let filled = ArrowEngineData::try_from_engine_data(filled)?;

// Write `filled` and commit 
...
}

Do you have any comments or feedback on this?