From c3624d706267e928267f87f1827618a38b8dec67 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 20:50:55 +0200 Subject: [PATCH] feat(sql): add insert and scan to DiskTable --- .../kotlin/kestrel/store/sql/DiskTable.kt | 174 ++++++++++++++++++ 1 file changed, 174 insertions(+) diff --git a/src/main/kotlin/kestrel/store/sql/DiskTable.kt b/src/main/kotlin/kestrel/store/sql/DiskTable.kt index b8b491d..753af12 100644 --- a/src/main/kotlin/kestrel/store/sql/DiskTable.kt +++ b/src/main/kotlin/kestrel/store/sql/DiskTable.kt @@ -163,3 +163,177 @@ class DiskTable internal constructor( for ((rid, row) in existing) update(txnId, rid, transform(row)) } + fun insert(txnId: Long, row: Row): RecordId { + requireArity(row) + val pkValues = primaryKeyValuesOf(row) + rejectDuplicatePrimaryKey(txnId, pkValues) + rejectDuplicateUniqueSecondary(txnId, row, excludeRid = null) + rejectDuplicateUniqueConstraints(txnId, row, excludeRid = null) + val roots = currentRoots() + val rid = heap.insert(txnId, TupleCodec.encode(row.values)) + indexRow(txnId, rid, row, pkValues) + persistIfRootsChanged(txnId, roots) + return rid + } + + fun get(txnId: Long, rid: RecordId): Row = Row(TupleCodec.decode(heap.get(txnId, rid))) + + fun scan(txnId: Long): List> { + scanCount++ + return rawScan(txnId) + } + + private fun rawScan(txnId: Long): List> = + heap.scan(txnId).map { (rid, bytes) -> rid to Row(TupleCodec.decode(bytes)) } + + fun delete(txnId: Long, rid: RecordId) { + val row = get(txnId, rid) + val roots = currentRoots() + unindexRow(txnId, rid, row) + heap.delete(txnId, rid) + persistIfRootsChanged(txnId, roots) + } + + fun update(txnId: Long, rid: RecordId, newRow: Row): RecordId { + requireArity(newRow) + val oldRow = get(txnId, rid) + val newPk = primaryKeyValuesOf(newRow) + val oldPk = primaryKeyValuesOf(oldRow) + if (newPk != oldPk) rejectDuplicatePrimaryKey(txnId, newPk) + rejectDuplicateUniqueSecondary(txnId, newRow, excludeRid = rid) + rejectDuplicateUniqueConstraints(txnId, newRow, excludeRid = rid) + val roots = currentRoots() + unindexRow(txnId, rid, oldRow) + val newRid = heap.update(txnId, rid, TupleCodec.encode(newRow.values)) + indexRow(txnId, newRid, newRow, newPk) + persistIfRootsChanged(txnId, roots) + return newRid + } + + /** Convenience for a single-column PRIMARY KEY; delegates to the general, composite-capable overload below. */ + fun findByPrimaryKey(txnId: Long, key: Value): Pair? = findByPrimaryKey(txnId, listOf(key)) + + /** + * Point lookup by the table's PRIMARY KEY, whether inline-single-column or table-level + * composite - [keys] must be given in [TableSchema.primaryKeyColumns] order. [KeyCodec.encodeKey] + * concatenates each value's own order-preserving encoding, so a single-element list here + * produces byte-identical output to [KeyCodec.encode] on that same value, keeping this the + * one lookup path both PRIMARY KEY shapes share. [keys] is folded through + * [encodePrimaryKey] first, so a search literal on a `COLLATE NOCASE`/`RTRIM`-declared PRIMARY + * KEY column finds a case-/trailing-space-variant row the same way a plain `=` comparison does. + */ + fun findByPrimaryKey(txnId: Long, keys: List): Pair? { + pointLookupCount++ + val index = primaryIndex ?: throw DiskTableException("table ${schema.name} has no primary key index") + val ridBits = index.search(txnId, encodePrimaryKey(keys)) ?: return null + val rid = RecordId.decode(ridBits) + return rid to get(txnId, rid) + } + + /** + * [DiskIndexAccessPlanner] plans a range against either a single-column PRIMARY KEY ([prefix] + * empty, see [DiskAccessPlan.PrimaryKeyRange]'s own doc) or the LAST column of a table-level + * composite one, after an equality prefix on every column before it ([prefix] non-empty, see + * [DiskAccessPlan.CompositePrimaryKeyRange]'s own doc) - either way, [prefix] plus [lower]/ + * [upper] always names every PRIMARY KEY column, so [encodePrimaryKey]'s own per-position + * collation folding ([kestrel.engine.collationFoldedUniquenessKey]) applies directly and always + * produces a COMPLETE, full-arity key: unlike [rangeBySecondary], no RID-suffix bracketing + * trick is needed here, since a PRIMARY KEY entry is a bare `encodeKey(values)` with nothing + * after it to disambiguate. + */ + fun rangeByPrimaryKey( + txnId: Long, + lower: Value? = null, + lowerInclusive: Boolean = true, + upper: Value? = null, + upperInclusive: Boolean = true, + prefix: List = emptyList() + ): List> { + rangeScanCount++ + val index = primaryIndex ?: throw DiskTableException("table ${schema.name} has no primary key index") + fun encodeBound(value: Value): ByteArray = + KeyCodec.encodeKey(collationFoldedUniquenessKey(schema, schema.primaryKeyIndices, prefix + value)) + val lowerBytes = lower?.let(::encodeBound)?.let { if (lowerInclusive) it else strictlyAfter(it) } + val upperBytes = upper?.let(::encodeBound) + return index.rangeScan(txnId, lowerBytes, upperBytes, toInclusive = upperInclusive).map { (_, ridBits) -> + val rid = RecordId.decode(ridBits) + rid to get(txnId, rid) + } + } + + /** Convenience for a single-column secondary index; delegates to the general, composite-capable overload below. */ + fun findBySecondary(txnId: Long, indexName: String, key: Value): List> = + findBySecondary(txnId, indexName, listOf(key)) + + /** + * Equality lookup on a secondary index, whether single-column or composite - [keys] must be + * given in [SecondaryIndexHandle.columnNames] order. A single-element list here produces a + * byte-identical prefix to the single-[Value] overload, keeping this the one lookup path both + * shapes share (mirrors [findByPrimaryKey]). [keys] is folded per-column through + * [equalRangeFor], so equality against a declared-collation column finds a case-/trailing- + * space-variant row instead of missing it. + */ + fun findBySecondary(txnId: Long, indexName: String, keys: List): List> { + secondaryLookupCount++ + val handle = secondaryIndexHandles.firstOrNull { it.name.equals(indexName, ignoreCase = true) } + ?: throw DiskTableException("no such index: $indexName on table ${schema.name}") + val (lower, upper) = equalRangeFor(handle.columnIndices, keys) + return handle.index.rangeScan(txnId, lower, upper, toInclusive = true).map { (_, ridBits) -> + val rid = RecordId.decode(ridBits) + rid to get(txnId, rid) + } + } + + /** + * [DiskIndexAccessPlanner] plans a range against a secondary index's column right after + * [prefix]'s equality run (see [kestrel.store.sql.DiskAccessPlan.SecondaryRange]'s own doc) - + * [prefix] is empty for a single-column index or a composite one ranged on its very first + * column. Unlike [rangeByPrimaryKey] (whose B+Tree entries are bare `encodeKey(values)`, no RID + * suffix, since a PRIMARY KEY is unique), a secondary index entry is `encodeKey(values) ++ rid` + * (see the class doc), so a bound naming just [lower]/[upper]'s own encoded bytes would sit + * strictly *before* every real entry sharing that value (any real key, being longer, sorts + * after its own prefix) - reusing [equalRangeFor]'s (all-zero-suffix, all-one-suffix) pair for + * `prefix + value` sidesteps this: the all-zero-suffix bound is a valid inclusive lower + * bound for "this value or greater" (it sits at or before every RID-suffixed entry of that + * exact value), and the all-one-suffix bound is a valid inclusive upper bound for "this value + * or less" (it sits at or after every RID-suffixed entry of that exact value) - swapped for the + * opposite (exclusive) side of each bound, so e.g. `> X` starts scanning right after the last + * possible entry of exactly `X`. When [prefix] is non-empty and one side has no explicit bound + * (e.g. `WHERE a = 5 AND b < 10`, no lower bound on `b`), that side falls back to + * [equalRangeFor]'s own bound for [prefix] alone, so the scan still stays within the `a = 5` + * group instead of running off into rows from a smaller/larger `a` - [prefix]'s own encoding is + * a byte-string prefix of every real entry sharing it (each value's encoding is self-delimited), + * so this bound is exact regardless of how many values [prefix] holds relative to the index's + * full column count. + */ + fun rangeBySecondary( + txnId: Long, + indexName: String, + prefix: List = emptyList(), + lower: Value? = null, + lowerInclusive: Boolean = true, + upper: Value? = null, + upperInclusive: Boolean = true + ): List> { + rangeScanCount++ + val handle = secondaryIndexHandles.firstOrNull { it.name.equals(indexName, ignoreCase = true) } + ?: throw DiskTableException("no such index: $indexName on table ${schema.name}") + val prefixBounds = if (prefix.isNotEmpty()) equalRangeFor(handle.columnIndices, prefix) else null + val lowerBytes = if (lower != null) { + val (inclusiveBound, exclusiveBound) = equalRangeFor(handle.columnIndices, prefix + lower) + if (lowerInclusive) inclusiveBound else exclusiveBound + } else { + prefixBounds?.first + } + val upperBytes = if (upper != null) { + val (exclusiveBound, inclusiveBound) = equalRangeFor(handle.columnIndices, prefix + upper) + if (upperInclusive) inclusiveBound else exclusiveBound + } else { + prefixBounds?.second + } + return handle.index.rangeScan(txnId, lowerBytes, upperBytes, toInclusive = true).map { (_, ridBits) -> + val rid = RecordId.decode(ridBits) + rid to get(txnId, rid) + } + } +