[Prototype] feat: Concurrent identity columns

[Rob2U]

What changes are proposed in this pull request?

We would like to integrate the feature by allowing connectors to use the following two workflows:

Condensed from the runnable examples/identity-column-demo.rs, which runs the whole flow against an InMemorySequenceClient (no external service).

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 
...
}

The resulting rows carry id = [1, 2, 3] and row_id = [1000, 1010, 1020].

Column metadata

A CIC column is marked by field metadata on an otherwise non-nullable LONG field. The keys (all under the delta.identity.v2. prefix):

Key	Type	Required	Meaning
delta.identity.v2.sequenceId	string	yes	Pointer to the UC sequence controlling this column
delta.identity.v2.start	number	yes	First value the sequence issues
delta.identity.v2.step	number	yes	Increment between successive values (non-zero)
delta.identity.v2.allowExplicitInsert	boolean	no (default false)	Whether user-supplied values are permitted (not used allowed right now)

The presence of sequenceId is what identifies a column as CIC. When it is present, start and step are also required (a missing one is an error). The sequenceId is a client-minted unique id of at most 64 characters.

Structure of the changes

Crate	Holds
delta_kernel — identity_columns.rs	IdentitySequenceState + ReservedRange, IdentityColumnInfo, detect_identity_columns, cic_column. No UC dependency.
unity-catalog-delta-client-api	The SequenceClient trait, its models, and InMemorySequenceClient. No kernel dependency.
delta-kernel-unity-catalog	Owner of client and kernel primitives: IdentityColumnWriter (write path) and register_identity_sequences (CREATE-table).

IdentitySequenceState

IdentitySequenceState is the state that contains the core logic and state of a sequence. It holds one cursor
per CIC column, a queue of ReservedRanges plus available / inflight / claimed counters, and knows nothing about any catalog.
It has the following methods:

begin_reserve(count)   -> Vec<SequenceReservationRequest>   // Mark that a certain amount of values will arrive (inflight)
complete_reserve(count, Vec<ReservedRange>) -> DeltaResult  // enqueue + validate the `ReservedRange`
fail_reserve(count, message)                                // Remove the inflight
try_claim(count)       -> DeltaResult<ClaimOutcome>         // Claimed | Reserve(deficit) | Wait
fill_engine_batch(handler, input, target_schema)           // emit values, columns in schema order (sync)

begin_reserve says what to reserve (a SequenceReservationRequest per column). The caller runs
its RPC and complete_reserve feeds the ranges back
or fail_reserve rolls the in-flight count back.
try_claim claims count for one caller so concurrent callers get disjoint ranges. It only reports
that a wait is needed (ClaimOutcome::Wait), the actual waiting is the caller's job.

ReservedRange { range_start, range_end, step } is the primitive providing the guarded enumeration logic.
It is both used by the state and FFI.

IdentityColumnWriter

IdentityColumnWriter<C: SequenceClient> is the object a connector talks to while writing. It is
an async wrapper over IdentitySequenceState (stored in a mutex) and the SequenceClient. The writer supports three
operations:

• reserve(count) (async) — runs begin_reserve under the lock, issues one ReserveIdentityRanges
  RPC covering all columns, then feeds the result back via complete_reserve / fail_reserve and
  signals the Notify. It returns a 'static future, so the engine can tokio::spawn it to prefetch
  the next chunk or await it inline.
• ensure_available(count) (async) — the pre-fill failsafe. Loops on try_claim: Claimed ->
  done. Reserve(deficit) -> await reserve(deficit). Wait -> await the Notify, then re-check.
• fill_engine_batch(...) (sync) - delegates straight to the kernel, which fills input.len()
  values per identity column and returns every column in target_schema order. Fails if a column
  is short (i.e. ensure_available was skipped).

The division is as follows: The actual logic (the queue, counters, claim decision, step validation, and
value fill) are located in the kernel, but its integration with a client is done here in the writer.

The clients

SequenceClient mirrors the UC Identity Sequence Service's three RPCs:

• create_identity_sequences — create, or idempotently get, sequences (CREATE-table time).
• reserve_identity_ranges — reserve ranges from one or more sequences; returned IdentityIdRanges
  are positional within the batch.
• drop_identity_sequences — idempotently drop sequences (DROP / REPLACE), reporting whether each existed.

A REST/HTTP implementation lives in unity-catalog-delta-rest-client and an InMemorySequenceClient serves tests.
At CREATE-table time the register_identity_sequences(client, table_id, &[IdentityColumnInfo])
helper issues one batched create for every column (called after the create
commit succeeds).

How was this tested?

All introduced components have accompanying unit tests. Integration tests live in delta-kernel-unity-catalog/tests/identity_columns.rs