Skip to content

Commit df901ec

Browse files
Rob2Umkroll-db
andauthored
[Spark] Concurrent Identity Columns: sequence-service abstraction + in-memory backend (#7619)
<!-- Thanks for sending a pull request! Here are some tips for you: 1. If this is your first time, please read our contributor guidelines: https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md 2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP] Your PR title ...'. 3. Be sure to keep the PR description updated to reflect all changes. 4. Please write your PR title to summarize what this PR proposes. 5. If possible, provide a concise example to reproduce the issue for a faster review. 6. If applicable, include the corresponding issue number in the PR title and link it in the body. --> #### Which Delta project/connector is this regarding? <!-- Please add the component selected below to the beginning of the pull request title For example: [Spark] Title of my pull request --> - [x] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description This PR is the second out of a stack of seven PRs in total. The combined stack adds support for the Concurrent Identity Column (CIC) feature following the RFC [here](#7272). TLDR: It allows for two concurrent 'inserters' to both succeed by adding an identity sequence service that hands out disjoint ranges of id values. This way the two writers do not conflict and can both commit. **_Stack overview:_** 1. Foundation (#7618) 2. Service abstraction + In-Memory one <- this PR (#7619) 3. Transport and buffering for the service requests (#7620) 4. Table feature and the insert path (#7621) 5. DDL integration (#7622) 6. Merge support (#7623) 7. UC Client addition (#7785) This PR introduces no behaviour changes, just scaffolding. It adds: - IdentitySequenceService trait containing the methods `createSequence`, `dropSequence`, `reserve` (+ CreateSequenceRequest / DropSequenceRequest / ReserveIdsRequest / ReserveIdsResponse). - `IdentitySequenceServices.resolve(spark)` for retrieval of the service to be used. Defaults to an in-memory `LocalIdentitySequenceService`, overridable via `spark.databricks.delta.identityColumn.concurrent.serviceClassName`. ## How was this patch tested? The `LocalIdentitySequenceService` is tested using the `LocalIdentitySequenceServiceSuite.scala`. These tests ensure that the the `LocalIdentitySequenceService` matches the expected behavior exactly. This is important because we will use it later for testing the other components. <!-- If tests were added, say they were added here. Please make sure to test the changes thoroughly including negative and positive cases if possible. If the changes were tested in any way other than unit tests, please clarify how you tested step by step (ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future). If the changes were not tested, please explain why. --> ## Does this PR introduce _any_ user-facing changes? <!-- If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible. If possible, please also clarify if this is a user-facing change compared to the released Delta Lake versions or within the unreleased branches such as master. If no, write 'No'. --> No --------- Signed-off-by: Robert Weeke <weeke.robert@gmail.com> Co-authored-by: mkroll-db <marco.kroll@databricks.com>
1 parent 26a4250 commit df901ec

5 files changed

Lines changed: 761 additions & 0 deletions

File tree

Lines changed: 158 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,158 @@
1+
/*
2+
* Copyright (2026) The Delta Lake Project Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.apache.spark.sql.delta.cic
18+
19+
/**
20+
* Driver-side trait for the UC Identity Sequence Service: the backend that hands
21+
* out non-overlapping ranges of identity values for Concurrent Identity Columns (CIC).
22+
*
23+
* Three operations are exposed:
24+
* - `createSequence`: registers a sequence under a caller-generated `sequenceId`.
25+
* The driver generates a unique sequenceId up front and stamps it into the column metadata
26+
* before the DDL commit. Used by `CREATE TABLE` with the CIC feature and by the
27+
* feature opt-in `ALTER TABLE ... SET TBLPROPERTIES`. Semantics are CreateOrGet:
28+
* a repeat call with matching `start`/`step` succeeds; a mismatch is an error.
29+
* - `reserveIds`: reserve a contiguous range of identity values for an
30+
* in-flight write. Returns the sequenceId, `[rangeStart, rangeEnd]` (inclusive)
31+
* and an echo of the immutable `step`. Callers verify the step matches the
32+
* schema-declared step to detect drifted state.
33+
* - `dropSequence`: retires a sequence the table no longer references (DROP FEATURE
34+
* downgrade, REPLACE re-stamp). Idempotent.
35+
*
36+
* Driver-only. Do not capture an instance in any closure that ships to executors.
37+
*/
38+
trait IdentitySequenceService {
39+
/**
40+
* Registers the sequence identified by the caller-supplied `req.sequenceId`.
41+
* CreateOrGet: a create for an already-existing `sequenceId` succeeds when the
42+
* stored `start`/`step` match the request and is an error otherwise. `step`
43+
* must be non-zero; implementations reject a zero step.
44+
*
45+
* Returns nothing: callers already hold the requested `start`/`step` and stamp
46+
* them into the schema metadata directly. Consistency is enforced by the
47+
* service (an error on a mismatching recreate), not by inspecting a response.
48+
*/
49+
def createSequence(req: CreateSequenceRequest): Unit
50+
51+
/**
52+
* Reserves a contiguous `[rangeStart, rangeEnd]` (inclusive) range of `count`
53+
* values from an existing sequence. `count` must be positive and reserving from
54+
* an unknown `sequenceId` is an error. Implementations must never hand out a
55+
* range that overlaps a previously reserved one, including when the underlying
56+
* counter would overflow.
57+
*
58+
* The range follows the sign of `step` (descending when `step` is negative), so
59+
* consumers must materialize values as `rangeStart + i * step` for `i` in
60+
* `[0, count)`, not by iterating `rangeStart to rangeEnd`.
61+
*/
62+
def reserveIds(req: ReserveIdsRequest): ReserveIdsResponse
63+
64+
/**
65+
* Drops (retires) the sequence registered under `(req.tableId, req.sequenceId)`. Idempotent:
66+
* dropping an unknown or already-retired sequence is a no-op. Reclaims a sequence a table no
67+
* longer references (DROP FEATURE downgrade, REPLACE re-stamp) so the service-side counter does
68+
* not leak. A dropped `(tableId, sequenceId)` must never be reused for live data; a re-converted
69+
* or replaced table always mints a fresh id, so the monotonic counter can never collide.
70+
*/
71+
def dropSequence(req: DropSequenceRequest): Unit
72+
}
73+
74+
/**
75+
* Thrown by an [[IdentitySequenceService]] backend when the service is administratively disabled.
76+
*/
77+
class IdentitySequenceServiceDisabledException(message: String, cause: Throwable = null)
78+
extends RuntimeException(message, cause)
79+
80+
/**
81+
* Request to register an identity sequence.
82+
*
83+
* `sequenceId` is generated by the driver up front, stamped into the column metadata under
84+
* `delta.identity.concurrent.sequenceId`, and passed here; the service does not mint it. The
85+
* sequence is keyed by `(tableId, sequenceId)`.
86+
*
87+
* `tableId` is the table scope the sequence lives under. For a catalog-owned (UC
88+
* coordinated-commits) table this is the UC table id (resolved via
89+
* `ConcurrentIdentityColumnSchema.sequenceServiceTableId`: the
90+
* `io.unitycatalog.tableId` table property). Because sequences are keyed by
91+
* `tableId`, commands that replace the table in place (e.g. `CREATE OR REPLACE
92+
* TABLE`) generate a new `tableId` and leave the old sequences orphaned -- the
93+
* replacement table starts fresh, which is the right semantics. For testing
94+
* only, the Delta table's own `metadata.id` can be used instead.
95+
*
96+
* `columnInformation` is audit/debugging-only; the caller passes the physical column name.
97+
*
98+
* @param sequenceId driver-generated unique id identifying the sequence
99+
* @param tableId table scope (UC table id for catalog-owned tables, else the table's `metadata.id`)
100+
* @param start start of sequence
101+
* @param step step of sequence; must be non-zero
102+
* @param columnInformation audit/debugging-only; the physical column name
103+
*/
104+
case class CreateSequenceRequest(
105+
sequenceId: String,
106+
tableId: String,
107+
start: Long,
108+
step: Long,
109+
columnInformation: Option[String] = None)
110+
111+
/**
112+
* Request to reserve a contiguous range of IDs from an existing sequence.
113+
*
114+
* `(tableId, sequenceId)` names the counter, mirroring the UC wire contract: the
115+
* backend keys sequences per table scope (a CLONE replicates a sequence id into the
116+
* clone's table scope as an independent counter) and authorizes MODIFY on the table.
117+
* `tableId` must be the same value the sequence was created under.
118+
*
119+
* `step` is the schema-declared step the caller expects. It travels on the UC reserve wire
120+
* so the backend can sanity-check the granted range against it (the backend stays the owner
121+
* of the step); the UC-backed implementation also verifies the range invariant
122+
* `rangeEnd - rangeStart == step * (count - 1)` on the granted range and fills
123+
* [[ReserveIdsResponse.step]]. The local stub validates it against the stored step, so a
124+
* caller whose schema drifted from the registration fails loudly.
125+
*
126+
* @param sequenceId the caller-generated id the sequence was created under
127+
* @param tableId table scope the sequence was created under (see
128+
* [[CreateSequenceRequest.tableId]])
129+
* @param count number of identity values to reserve; must be positive
130+
* @param step the schema-declared step of the identity column
131+
*/
132+
case class ReserveIdsRequest(sequenceId: String, tableId: String, count: Long, step: Long)
133+
134+
/**
135+
* Response from a reserve-ids request.
136+
*
137+
* Callers MUST verify `step` matches the step persisted in the schema metadata.
138+
* A mismatch signals corrupted or drifted state (step is immutable at the
139+
* service) and is a hard error.
140+
*
141+
* @param sequenceId echo of the request
142+
* @param rangeStart first value in the reserved range (inclusive)
143+
* @param rangeEnd last value in the reserved range (inclusive); for a negative
144+
* `step` this is less than `rangeStart` (descending range)
145+
* @param step echo of the immutable step
146+
*/
147+
case class ReserveIdsResponse(
148+
sequenceId: String,
149+
rangeStart: Long,
150+
rangeEnd: Long,
151+
step: Long)
152+
153+
/**
154+
* Request to drop (retire) an identity sequence. `(tableId, sequenceId)` names the counter, the
155+
* same scope it was created and reserved under (see [[CreateSequenceRequest.tableId]]). Idempotent
156+
* at the service: dropping an unknown or already-retired id is a no-op.
157+
*/
158+
case class DropSequenceRequest(sequenceId: String, tableId: String)
Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
/*
2+
* Copyright (2026) The Delta Lake Project Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.apache.spark.sql.delta.cic
18+
19+
import org.apache.spark.sql.delta.sources.DeltaSQLConf
20+
import org.apache.spark.sql.delta.cic.IdentitySequenceService
21+
22+
import org.apache.spark.sql.SparkSession
23+
import org.apache.spark.util.Utils
24+
25+
/**
26+
* Resolves the [[IdentitySequenceService]] backend for a session. When
27+
* [[DeltaSQLConf.CONCURRENT_IDENTITY_COLUMN_SERVICE_CLASS_NAME]] is set, reflectively instantiates
28+
* that class (no-arg constructor); whether a table uses a sequence service at all stays
29+
* table-state-driven (the stamped sequence pointer).
30+
*
31+
* Otherwise there is no built-in backend, so it fails with a clear error directing the caller to
32+
* configure a backend via the class-name config.
33+
*/
34+
object IdentitySequenceServices {
35+
def resolve(spark: SparkSession): IdentitySequenceService = {
36+
spark.conf.get(DeltaSQLConf.CONCURRENT_IDENTITY_COLUMN_SERVICE_CLASS_NAME) match {
37+
case Some(className) if className.nonEmpty =>
38+
Utils.classForName(className).getDeclaredConstructor().newInstance()
39+
.asInstanceOf[IdentitySequenceService]
40+
case _ =>
41+
throw new UnsupportedOperationException(
42+
"No concurrent identity-sequence backend is configured. Set " +
43+
"spark.databricks.delta.identityColumn.concurrent.serviceClassName to an " +
44+
"IdentitySequenceService implementation.")
45+
}
46+
}
47+
}

‎spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3618,6 +3618,15 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils {
36183618
.checkValue(_ > 0, "maxColumnsPerTable must be positive")
36193619
.createWithDefault(50)
36203620

3621+
val CONCURRENT_IDENTITY_COLUMN_SERVICE_CLASS_NAME =
3622+
buildConf("identityColumn.concurrent.serviceClassName")
3623+
.internal()
3624+
.doc("Optional override naming an IdentitySequenceService implementation (with a no-arg " +
3625+
"constructor) that IdentitySequenceServices.resolve instantiates reflectively instead of " +
3626+
"the default backend. Unit suites point it at the LocalIdentitySequenceService fake.")
3627+
.stringConf
3628+
.createOptional
3629+
36213630
//////////////////
36223631
// GeoSpatial
36233632
//////////////////

0 commit comments

Comments
 (0)