diff --git a/.gitignore b/.gitignore index ac6ace56..b6747baa 100644 --- a/.gitignore +++ b/.gitignore @@ -84,3 +84,4 @@ engine/target/ # Native library build artifacts native/src/c/*.dylib native/src/c/*.so +native/slatedb/ diff --git a/conventions/src/main/kotlin/actionbase/BaseConventionsPlugin.kt b/conventions/src/main/kotlin/actionbase/BaseConventionsPlugin.kt index 3821a912..5875dabc 100644 --- a/conventions/src/main/kotlin/actionbase/BaseConventionsPlugin.kt +++ b/conventions/src/main/kotlin/actionbase/BaseConventionsPlugin.kt @@ -24,6 +24,7 @@ class BaseConventionsPlugin : Plugin { project.version = project.rootProject.version // Configure repositories + project.repositories.mavenLocal() project.repositories.mavenCentral() // Configure dependencies diff --git a/engine/build.gradle.kts b/engine/build.gradle.kts index c69b7cba..390cdb46 100644 --- a/engine/build.gradle.kts +++ b/engine/build.gradle.kts @@ -41,6 +41,9 @@ dependencies { implementation(Dependencies.Logging.SLF4J_API) implementation(Dependencies.Logging.LOGBACK_CLASSIC) + // SlateDB + implementation("io.slatedb:slatedb:0.1.0-SNAPSHOT") + // HBase implementation(Dependencies.HBase.CLIENT) implementation(Dependencies.HBase.MAPREDUCE) diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt index fc1dae98..dd56cf2f 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/Graph.kt @@ -62,6 +62,8 @@ import com.kakao.actionbase.v2.engine.sql.toRowFlux import com.kakao.actionbase.v2.engine.storage.hbase.HBaseConnections import com.kakao.actionbase.v2.engine.storage.hbase.HBaseOptions import com.kakao.actionbase.v2.engine.storage.jdbc.MetadataTable +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbConnections +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbOptions import com.kakao.actionbase.v2.engine.util.getLogger import com.kakao.actionbase.v2.engine.wal.Wal import com.kakao.actionbase.v2.engine.wal.WalFactory @@ -616,6 +618,10 @@ class Graph( val options = storage.materialize().options as HBaseOptions options.checkConnection() } + StorageType.SLATEDB -> { + val options = storage.materialize().options as SlateDbOptions + options.checkConnection() + } else -> Mono.just(false) } @@ -892,6 +898,7 @@ class Graph( intervalDisposable?.dispose() log.info("Disposed Flux.interval for reloading metastore - {}", intervalDisposable) HBaseConnections.closeConnections().block() + SlateDbConnections.closeConnections().block() DefaultHBaseCluster.INSTANCE.close() } diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt index 4fa10b50..028ed29a 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/LabelEntity.kt @@ -21,12 +21,15 @@ import com.kakao.actionbase.v2.engine.label.hbase.HBaseIndexedLabel import com.kakao.actionbase.v2.engine.label.metastore.JdbcHashLabel import com.kakao.actionbase.v2.engine.label.metastore.LocalBackedJdbcHashLabel import com.kakao.actionbase.v2.engine.label.nil.NilLabel +import com.kakao.actionbase.v2.engine.label.slatedb.SlateDbHashLabel +import com.kakao.actionbase.v2.engine.label.slatedb.SlateDbIndexedLabel import com.kakao.actionbase.v2.engine.service.ddl.LabelCreateRequest import com.kakao.actionbase.v2.engine.sql.RowWithSchema import com.kakao.actionbase.v2.engine.storage.DatastoreStorage import com.kakao.actionbase.v2.engine.storage.hbase.HBaseStorage import com.kakao.actionbase.v2.engine.storage.jdbc.JdbcStorage import com.kakao.actionbase.v2.engine.storage.local.LocalStorage +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbStorage import com.kakao.actionbase.v2.engine.util.getLogger import org.slf4j.Logger @@ -79,6 +82,7 @@ data class LabelEntity( is JdbcStorage -> JdbcHashLabel.create(this, graph, storage, block) is HBaseStorage -> HBaseHashLabel.create(this, graph, storage) is DatastoreStorage -> DatastoreHashLabel.create(this, graph, block) + is SlateDbStorage -> SlateDbHashLabel.create(this, graph, storage) else -> { logger.error( "{} supports only Local, Jdbc, HBase storage types. {} is not supported. Fallback to NilLabel", @@ -99,9 +103,10 @@ data class LabelEntity( when (storage) { is HBaseStorage -> HBaseIndexedLabel.create(this, graph, storage) is DatastoreStorage -> DatastoreIndexedLabel.create(this, graph, block) + is SlateDbStorage -> SlateDbIndexedLabel.create(this, graph, storage) else -> { logger.error( - "{} supports only Jdbc, HBase storage types. {} is not supported. Fallback to NilLabel", + "{} supports only Jdbc, HBase, SlateDb storage types. {} is not supported. Fallback to NilLabel", type, storage, ) diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/StorageEntity.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/StorageEntity.kt index 5b0cdec1..2f689668 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/StorageEntity.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/entity/StorageEntity.kt @@ -11,6 +11,7 @@ import com.kakao.actionbase.v2.engine.storage.hbase.HBaseStorage import com.kakao.actionbase.v2.engine.storage.jdbc.JdbcStorage import com.kakao.actionbase.v2.engine.storage.local.LocalStorage import com.kakao.actionbase.v2.engine.storage.nil.NilStorage +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbStorage import com.kakao.actionbase.v2.engine.util.getLogger import org.slf4j.Logger @@ -41,6 +42,9 @@ data class StorageEntity( StorageType.HBASE -> { HBaseStorage(this) } + StorageType.SLATEDB -> { + SlateDbStorage(this) + } StorageType.DATASTORE -> { DatastoreStorage } diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbHashLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbHashLabel.kt new file mode 100644 index 00000000..b958cd82 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbHashLabel.kt @@ -0,0 +1,288 @@ +package com.kakao.actionbase.v2.engine.label.slatedb + +import com.kakao.actionbase.v2.core.code.EdgeEncoder +import com.kakao.actionbase.v2.core.code.EncodedKey +import com.kakao.actionbase.v2.core.code.IdEdgeEncoder +import com.kakao.actionbase.v2.core.code.KeyFieldValue +import com.kakao.actionbase.v2.core.code.KeyValue +import com.kakao.actionbase.v2.core.edge.Edge +import com.kakao.actionbase.v2.core.edge.SchemaEdge +import com.kakao.actionbase.v2.core.metadata.Direction +import com.kakao.actionbase.v2.engine.GraphDefaults +import com.kakao.actionbase.v2.engine.edge.decodeByteArray +import com.kakao.actionbase.v2.engine.edge.toRow +import com.kakao.actionbase.v2.engine.entity.LabelEntity +import com.kakao.actionbase.v2.engine.label.AbstractLabel +import com.kakao.actionbase.v2.engine.label.LabelFactory +import com.kakao.actionbase.v2.engine.sql.DataFrame +import com.kakao.actionbase.v2.engine.sql.Row +import com.kakao.actionbase.v2.engine.sql.StatKey +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbStorage +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbTable + +import java.nio.ByteBuffer +import java.util.Arrays + +import reactor.core.publisher.Mono + +open class SlateDbHashLabel( + entity: LabelEntity, + coder: EdgeEncoder, + private val table: Mono, +) : AbstractLabel(entity, coder) { + override fun findHashEdge(keyField: EncodedKey): Mono { + require(keyField.field == null) { "field must be null" } + return table.flatMap { it.get(keyField.key) } + } + + override fun create( + keyField: EncodedKey, + value: ByteArray, + ): Mono> { + require(keyField.field == null) { "field must be null" } + return table + .flatMap { it.put(keyField.key, value) } + .thenReturn(emptyList()) + } + + override fun update( + keyField: EncodedKey, + value: ByteArray, + ): Mono> = create(keyField, value) + + override fun delete(keyField: EncodedKey): Mono> { + require(keyField.field == null) { "field must be null" } + return table + .flatMap { it.delete(keyField.key) } + .thenReturn(emptyList()) + } + + override fun setnx( + keyField: EncodedKey, + value: ByteArray, + ): Mono { + require(keyField.field == null) { "field must be null" } + return table.flatMap { tbl -> + tbl + .get(keyField.key) + .hasElement() + .flatMap { exists -> + if (exists) { + Mono.just(false) + } else { + tbl.put(keyField.key, value).thenReturn(true) + } + } + } + } + + override fun cad( + keyField: EncodedKey, + value: ByteArray, + ): Mono { + require(keyField.field == null) { "field must be null" } + return table.flatMap { tbl -> + tbl + .get(keyField.key) + .flatMap { existingValue -> + if (Arrays.equals(existingValue, value)) { + tbl.delete(keyField.key).thenReturn(1L) + } else { + Mono.just(0L) + } + }.defaultIfEmpty(0L) + } + } + + override fun findLockValue(keyField: EncodedKey): Mono { + require(keyField.field == null) { "field must be null" } + return table.flatMap { it.get(keyField.key) } + } + + override fun incrby( + key: ByteArray, + acc: Long, + ): Mono> = + table.flatMap { tbl -> + tbl + .get(key) + .map { bytes -> ByteBuffer.wrap(bytes).getLong() } + .defaultIfEmpty(0L) + .flatMap { current -> + val newValue = ByteBuffer.allocate(8).putLong(current + acc).array() + tbl.put(key, newValue).thenReturn(emptyList()) + } + } + + override fun scanStorage( + prefix: EncodedKey, + limit: Int, + start: EncodedKey?, + end: EncodedKey?, + ): Mono>> = + table + .flatMap { it.scanPrefix(prefix.key, limit + 1) } + .map { results -> + results + // Filter by start key (exclusive) + .dropWhile { (key, _) -> + start?.key?.let { startKey -> Arrays.compareUnsigned(startKey, key) >= 0 } ?: false + } + // Filter by end key (exclusive) + .dropLastWhile { (key, _) -> + end?.key?.let { endKey -> Arrays.compareUnsigned(endKey, key) < 0 } ?: false + }.take(limit) + .map { (key, value) -> KeyFieldValue(key, value) } + } + + override fun encodedEdgeToSchemaEdge(keyFieldValue: KeyFieldValue): SchemaEdge = entity.schema.decodeByteArray(keyFieldValue) + + override fun deleteOnLock(keyField: KeyValue): Mono = cad(EncodedKey(keyField.key), keyField.value).map { it > 0 } + + override fun getSelf( + src: List, + stats: Set, + idEdgeEncoder: IdEdgeEncoder, + ): Mono { + val withAll = stats.contains(StatKey.WITH_ALL) + val withEdgeId = withAll || stats.contains(StatKey.EDGE_ID) + + val keysMono = + Mono.just( + src.map { + val edge = Edge(0L, it, it).ensureType(entity.schema) + coder.encodeHashEdgeKey(edge, entity.id) + }, + ) + + return keysMono + .flatMap { keys -> + table.flatMap { tbl -> + Mono.zip( + keys.map { key -> tbl.get(key.key).map { key to it } }, + ) { results -> + results + .filterIsInstance, ByteArray>>() + .mapNotNull { (key, value) -> + try { + encodedEdgeToSchemaEdge(KeyFieldValue(key.key, value)) + } catch (e: Exception) { + null + } + }.filter { withAll || it.isActive } + .map { + if (withEdgeId) { + it.toRow(withAll, idEdgeEncoder) + } else { + it.toRow(withAll, null) + } + } + } + } + }.map { rows -> + DataFrame( + rows, + if (withAll) { + entity.schema.allStructType + } else if (withEdgeId) { + entity.schema.edgeIdStructType + } else { + entity.schema.structType + }, + ) + }.defaultIfEmpty(DataFrame.empty(entity.schema.allStructType)) + } + + override fun get( + src: Any, + tgt: List, + dir: Direction, + stats: Set, + idEdgeEncoder: IdEdgeEncoder, + ): Mono { + val withAll = stats.contains(StatKey.WITH_ALL) + val withEdgeId = withAll || stats.contains(StatKey.EDGE_ID) + + val keys = + tgt.map { + val edge = Edge(0L, src, it).ensureType(entity.schema) + coder.encodeHashEdgeKey(edge, entity.id) + } + + return table + .flatMap { tbl -> + if (keys.isEmpty()) { + Mono.just(emptyList()) + } else { + Mono.zip( + keys.map { key -> tbl.get(key.key).map { key to it }.defaultIfEmpty(key to ByteArray(0)) }, + ) { results -> + results + .filterIsInstance, ByteArray>>() + .filter { it.second.isNotEmpty() } + .mapNotNull { (key, value) -> + try { + encodedEdgeToSchemaEdge(KeyFieldValue(key.key, value)) + } catch (e: Exception) { + null + } + }.filter { withAll || it.isActive } + .map { + if (withEdgeId) { + it.toRow(withAll, idEdgeEncoder, isMultiEdge) + } else { + it.toRow(withAll, null, isMultiEdge) + } + } + } + } + }.map { rows -> + DataFrame( + rows, + if (withAll) { + entity.schema.allStructType + } else if (withEdgeId) { + entity.schema.edgeIdStructType + } else { + entity.schema.structType + }, + ) + }.defaultIfEmpty(DataFrame.empty(entity.schema.allStructType)) + } + + override fun getCountRows( + srcAndKeys: List>, + dir: Direction, + ): Mono> = + table.flatMap { tbl -> + if (srcAndKeys.isEmpty()) { + Mono.just(emptyList()) + } else { + Mono.zip( + srcAndKeys.map { (src, key) -> + tbl + .get(key) + .map { bytes -> ByteBuffer.wrap(bytes).getLong() } + .defaultIfEmpty(0L) + .map { count -> Row(arrayOf(src, count, dir)) } + }, + ) { results -> results.filterIsInstance() } + } + } + + companion object : LabelFactory { + override fun create( + entity: LabelEntity, + graph: GraphDefaults, + storage: SlateDbStorage, + block: SlateDbHashLabel.() -> Unit, + ): SlateDbHashLabel { + val table = storage.options.getTable() + return SlateDbHashLabel( + entity = entity, + coder = graph.edgeEncoderFactory.bytesKeyValueEncoder, + table = table, + ).apply(block) + } + } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbIndexedLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbIndexedLabel.kt new file mode 100644 index 00000000..953e3431 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbIndexedLabel.kt @@ -0,0 +1,69 @@ +package com.kakao.actionbase.v2.engine.label.slatedb + +import com.kakao.actionbase.v2.core.code.EdgeEncoder +import com.kakao.actionbase.v2.core.code.IdEdgeEncoder +import com.kakao.actionbase.v2.core.code.Index +import com.kakao.actionbase.v2.engine.GraphDefaults +import com.kakao.actionbase.v2.engine.cdc.CdcContext +import com.kakao.actionbase.v2.engine.entity.LabelEntity +import com.kakao.actionbase.v2.engine.label.AbstractLabel +import com.kakao.actionbase.v2.engine.label.LabelFactory +import com.kakao.actionbase.v2.engine.label.mixin.IndexedLabelMixin +import com.kakao.actionbase.v2.engine.sql.DataFrame +import com.kakao.actionbase.v2.engine.sql.ScanFilter +import com.kakao.actionbase.v2.engine.sql.StatKey +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbStorage +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbTable + +import reactor.core.publisher.Mono + +/** + * Manages IndexedEdgeEncoder in SlateDB + */ +class SlateDbIndexedLabel( + entity: LabelEntity, + coder: EdgeEncoder, + override val indices: List, + override val indexNameToIndex: Map, + table: Mono, +) : SlateDbHashLabel( + entity = entity, + coder = coder, + table = table, + ), + IndexedLabelMixin { + override val self: AbstractLabel = this + + override fun finalizeEdgeMutationUnderLock(context: CdcContext): Mono> = mutateIndexedEdges(context) + + override fun scan( + scanFilter: ScanFilter, + stats: Set, + idEdgeEncoder: IdEdgeEncoder, + ): Mono = + scanIndexedEdges( + scanFilter, + stats, + idEdgeEncoder, + ) + + companion object : LabelFactory { + override fun create( + entity: LabelEntity, + graph: GraphDefaults, + storage: SlateDbStorage, + block: SlateDbIndexedLabel.() -> Unit, + ): SlateDbIndexedLabel { + val table = storage.options.getTable() + val indices: List = entity.indices + val indexNameToId = indices.associateBy { it.name } + return SlateDbIndexedLabel( + entity = entity, + coder = graph.edgeEncoderFactory.bytesKeyValueEncoder, + indices = indices, + indexNameToIndex = indexNameToId, + table = table, + ) + } + } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/StorageType.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/StorageType.kt index 55c968f5..f278f38b 100644 --- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/StorageType.kt +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metadata/StorageType.kt @@ -5,6 +5,7 @@ enum class StorageType { LOCAL, JDBC, HBASE, + SLATEDB, DATASTORE, ; diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbConnections.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbConnections.kt new file mode 100644 index 00000000..c3cccd4a --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbConnections.kt @@ -0,0 +1,71 @@ +package com.kakao.actionbase.v2.engine.storage.slatedb + +import com.kakao.actionbase.v2.engine.util.getLogger + +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean + +import io.slatedb.SlateDb +import io.slatedb.SlateDbConfig +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers + +object SlateDbConnections { + private val logger = getLogger() + + private val initialized = AtomicBoolean(false) + private val connections: ConcurrentHashMap> = ConcurrentHashMap() + + fun ensureInitialized() { + if (initialized.compareAndSet(false, true)) { + logger.info("Initializing SlateDB (native library loaded from JAR classpath)") + SlateDb.initLogging(SlateDbConfig.LogLevel.INFO) + } + } + + fun getConnection( + dbPath: String, + url: String, + ): Mono { + val cacheKey = getCacheKey(dbPath, url) + + return connections.computeIfAbsent(cacheKey) { key -> + Mono + .fromCallable { + ensureInitialized() + val db = SlateDb.open(dbPath, url, null) + SlateDbTable.create(db) + }.subscribeOn(Schedulers.boundedElastic()) + .doOnSuccess { + logger.info("Successfully opened SlateDB connection for cacheKey: {}", key) + }.doOnError { error -> + logger.error("Failed to open SlateDB connection for cacheKey: {}", key, error) + connections.remove(key) + }.cache() + } + } + + private fun getCacheKey( + dbPath: String, + url: String, + ): String = "$url/$dbPath" + + fun closeConnections(): Mono { + val closeMonos = + connections.entries.map { (key, tableMono) -> + tableMono + .flatMap { table -> + Mono + .fromRunnable { + try { + table.close() + logger.info("Closed SlateDB connection for cacheKey: {}", key) + } catch (e: Exception) { + logger.error("Error closing SlateDB connection for cacheKey: {}", key, e) + } + }.subscribeOn(Schedulers.boundedElastic()) + } + } + return Mono.`when`(closeMonos).doFinally { connections.clear() } + } +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbOptions.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbOptions.kt new file mode 100644 index 00000000..0864a311 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbOptions.kt @@ -0,0 +1,24 @@ +package com.kakao.actionbase.v2.engine.storage.slatedb + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties + +import reactor.core.publisher.Mono + +@JsonIgnoreProperties(ignoreUnknown = true) +data class SlateDbOptions( + val path: String = "data", + val url: String = "", +) { + fun checkConnection(): Mono = + if (url.isBlank()) { + Mono.just(false) + } else { + Mono.just(true) + } + + fun getTable(): Mono = + SlateDbConnections.getConnection( + dbPath = path, + url = url, + ) +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbStorage.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbStorage.kt new file mode 100644 index 00000000..ea6546b4 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbStorage.kt @@ -0,0 +1,11 @@ +package com.kakao.actionbase.v2.engine.storage.slatedb + +import com.kakao.actionbase.v2.engine.entity.StorageEntity +import com.kakao.actionbase.v2.engine.storage.Storage +import com.kakao.actionbase.v2.engine.storage.Storage.Companion.parseOptions + +class SlateDbStorage( + override val entity: StorageEntity, +) : Storage { + override val options: SlateDbOptions = parseOptions(entity.conf) +} diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTable.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTable.kt new file mode 100644 index 00000000..38d26d03 --- /dev/null +++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTable.kt @@ -0,0 +1,125 @@ +package com.kakao.actionbase.v2.engine.storage.slatedb + +import java.nio.ByteBuffer + +import io.slatedb.SlateDb +import io.slatedb.SlateDbKeyValue +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers + +sealed class BatchOperation { + data class Put( + val key: ByteArray, + val value: ByteArray, + ) : BatchOperation() + + data class Delete( + val key: ByteArray, + ) : BatchOperation() + + /** Not yet supported - waiting for merge operator in slatedb-c (#1250) */ + data class Increment( + val key: ByteArray, + val delta: Long, + ) : BatchOperation() +} + +interface SlateDbTable : AutoCloseable { + fun get(key: ByteArray): Mono + + fun put( + key: ByteArray, + value: ByteArray, + ): Mono + + fun delete(key: ByteArray): Mono + + fun flush(): Mono + + fun scanPrefix( + prefix: ByteArray, + limit: Int, + ): Mono>> + + fun batch(operations: List): Mono + + companion object { + fun create(db: SlateDb): SlateDbTable = SlateDbTableImpl(db) + } +} + +internal class SlateDbTableImpl( + private val db: SlateDb, +) : SlateDbTable { + override fun get(key: ByteArray): Mono = + Mono + .fromCallable { db.get(key) } + .flatMap { Mono.justOrEmpty(it) } + .subscribeOn(Schedulers.boundedElastic()) + + override fun put( + key: ByteArray, + value: ByteArray, + ): Mono = + Mono + .fromCallable { db.put(key, value) } + .subscribeOn(Schedulers.boundedElastic()) + .then() + + override fun delete(key: ByteArray): Mono = + Mono + .fromCallable { db.delete(key) } + .subscribeOn(Schedulers.boundedElastic()) + .then() + + override fun flush(): Mono = + Mono + .fromCallable { db.flush() } + .subscribeOn(Schedulers.boundedElastic()) + .then() + + override fun scanPrefix( + prefix: ByteArray, + limit: Int, + ): Mono>> = + Mono + .fromCallable { + val results = mutableListOf>() + db.scanPrefix(prefix).use { iterator -> + var kv: SlateDbKeyValue? = iterator.next() + var count = 0 + while (kv != null && count < limit) { + results.add(kv.key() to kv.value()) + count++ + kv = iterator.next() + } + } + results.toList() + }.subscribeOn(Schedulers.boundedElastic()) + + override fun batch(operations: List): Mono = + Mono + .fromCallable { + SlateDb.newWriteBatch().use { batch -> + operations.forEach { op -> + when (op) { + is BatchOperation.Put -> batch.put(op.key, op.value) + is BatchOperation.Delete -> batch.delete(op.key) + is BatchOperation.Increment -> { + // TODO: Replace with batch.merge() once slatedb#1250 is resolved + // WARNING: This is NOT atomic - race condition possible + val currentValue = db.get(op.key)?.let { ByteBuffer.wrap(it).long } ?: 0L + val newBytes = ByteBuffer.allocate(Long.SIZE_BYTES).putLong(currentValue + op.delta).array() + batch.put(op.key, newBytes) + } + } + } + db.write(batch) + } + }.subscribeOn(Schedulers.boundedElastic()) + .then() + + override fun close() { + db.close() + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/DatastoreCompatibilityTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/DatastoreCompatibilityTest.kt new file mode 100644 index 00000000..379160da --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/DatastoreCompatibilityTest.kt @@ -0,0 +1,285 @@ +package com.kakao.actionbase.engine.datastore + +import java.nio.ByteBuffer +import java.nio.ByteOrder +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicInteger + +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Assumptions.assumeTrue +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test + +/** + * Abstract compatibility test for storage backends. + * + * Required operations: get, scan, put, delete, increment, batch, checkAndMutate. + * + * @see Storage Backends + */ +abstract class DatastoreCompatibilityTest { + protected abstract fun createStore(): StorageOperations + + protected open fun cleanup() {} + + protected open fun supportsCheckAndMutate(): Boolean = true + + protected open fun supportsScanLimit(): Boolean = true + + private lateinit var store: StorageOperations + + @BeforeEach fun setUp() { + store = createStore() + } + + @AfterEach fun tearDown() { + cleanup() + } + + @Nested + @DisplayName("get") + inner class GetTest { + @Test fun `returns value when key exists`() { + store.put(b("key"), b("value")) + assert(store.get(b("key"))?.contentEquals(b("value")) == true) + } + + @Test fun `returns null when key not exists`() { + assert(store.get(b("missing")) == null) + } + + @Test fun `getAll returns matching records`() { + store.put(b("k1"), b("v1")) + store.put(b("k2"), b("v2")) + assert(store.getAll(listOf(b("k1"), b("k2"))).size == 2) + } + + @Test fun `getAll skips missing keys`() { + store.put(b("exists"), b("v")) + assert(store.getAll(listOf(b("exists"), b("missing"))).size == 1) + } + } + + @Nested + @DisplayName("scan") + inner class ScanTest { + @BeforeEach fun setup() { + listOf("user:001:a", "user:001:b", "user:002:a", "post:001").forEach { + store.put(b(it), b("v")) + } + } + + @Test fun `returns matching prefix`() { + val results = store.scan(b("user:001"), 100) + assert(results.size == 2) + assert(results.all { String(it.first).startsWith("user:001") }) + } + + @Test fun `returns empty for non-matching prefix`() { + assert(store.scan(b("nonexistent"), 100).isEmpty()) + } + + @Test fun `returns sorted keys`() { + val keys = store.scan(b("user:"), 100).map { String(it.first) } + assert(keys == keys.sorted()) + } + + @Test fun `respects limit`() { + assumeTrue(supportsScanLimit()) + assert(store.scan(b("user:"), 2).size == 2) + } + } + + @Nested + @DisplayName("put") + inner class PutTest { + @Test fun `stores value`() { + store.put(b("k"), b("v")) + assert(store.get(b("k"))?.contentEquals(b("v")) == true) + } + + @Test fun `overwrites existing`() { + store.put(b("k"), b("old")) + store.put(b("k"), b("new")) + assert(String(store.get(b("k"))!!) == "new") + } + } + + @Nested + @DisplayName("delete") + inner class DeleteTest { + @Test fun `removes key`() { + store.put(b("k"), b("v")) + store.delete(b("k")) + assert(store.get(b("k")) == null) + } + + @Test fun `silently succeeds for missing key`() { + store.delete(b("nonexistent")) + } + } + + @Nested + @DisplayName("increment") + inner class IncrementTest { + @Test fun `creates counter if not exists`() { + assert(store.increment(b("cnt"), 10) == 10L) + } + + @Test fun `updates existing counter`() { + store.put(b("cnt"), longToBytes(100)) + assert(store.increment(b("cnt"), 50) == 150L) + } + + @Test fun `decrements with negative delta`() { + store.put(b("cnt"), longToBytes(100)) + assert(store.increment(b("cnt"), -30) == 70L) + } + } + + @Nested + @DisplayName("batch") + inner class BatchTest { + @Test fun `executes puts`() { + store.batch(listOf(Mutation.Put(b("b1"), b("v1")), Mutation.Put(b("b2"), b("v2")))) + assert(store.getAll(listOf(b("b1"), b("b2"))).size == 2) + } + + @Test fun `executes deletes`() { + store.put(b("d1"), b("v")) + store.put(b("d2"), b("v")) + store.batch(listOf(Mutation.Delete(b("d1")), Mutation.Delete(b("d2")))) + assert(store.getAll(listOf(b("d1"), b("d2"))).isEmpty()) + } + + @Test fun `executes increments`() { + store.batch(listOf(Mutation.Increment(b("c1"), 10), Mutation.Increment(b("c2"), 20))) + assert(bytesToLong(store.get(b("c1"))!!) == 10L) + assert(bytesToLong(store.get(b("c2"))!!) == 20L) + } + + @Test fun `executes mixed mutations`() { + store.put(b("to-delete"), b("v")) + store.batch( + listOf( + Mutation.Put(b("new"), b("v")), + Mutation.Delete(b("to-delete")), + Mutation.Increment(b("cnt"), 100), + ), + ) + assert(store.get(b("new")) != null) + assert(store.get(b("to-delete")) == null) + assert(bytesToLong(store.get(b("cnt"))!!) == 100L) + } + } + + @Nested + @DisplayName("checkAndMutate") + inner class CheckAndMutateTest { + @BeforeEach fun checkSupport() { + assumeTrue(supportsCheckAndMutate()) + } + + @Nested + @DisplayName("setIfNotExists") + inner class SetIfNotExistsTest { + @Test fun `succeeds when key not exists`() { + assert(store.setIfNotExists(b("lock"), b("owner"))) + assert(store.get(b("lock"))?.contentEquals(b("owner")) == true) + } + + @Test fun `fails when key exists`() { + store.put(b("lock"), b("existing")) + assert(!store.setIfNotExists(b("lock"), b("new"))) + assert(String(store.get(b("lock"))!!) == "existing") + } + } + + @Nested + @DisplayName("deleteIfEquals") + inner class DeleteIfEqualsTest { + @Test fun `succeeds when value matches`() { + store.put(b("lock"), b("owner")) + assert(store.deleteIfEquals(b("lock"), b("owner"))) + assert(store.get(b("lock")) == null) + } + + @Test fun `fails when value differs`() { + store.put(b("lock"), b("owner")) + assert(!store.deleteIfEquals(b("lock"), b("different"))) + assert(store.get(b("lock")) != null) + } + + @Test fun `fails when key not exists`() { + assert(!store.deleteIfEquals(b("missing"), b("v"))) + } + } + + @Nested + @DisplayName("concurrent") + inner class ConcurrentTest { + @Test fun `only one thread acquires lock`() { + val threads = 10 + val acquired = AtomicInteger(0) + val latch = CountDownLatch(threads) + val executor = Executors.newFixedThreadPool(threads) + + repeat(threads) { i -> + executor.submit { + try { + if (store.setIfNotExists(b("lock"), b("owner-$i"))) { + acquired.incrementAndGet() + } + } finally { + latch.countDown() + } + } + } + + latch.await() + executor.shutdown() + assert(acquired.get() == 1) { "Expected 1 but got ${acquired.get()}" } + } + + @Test fun `only owner releases lock`() { + store.put(b("lock"), b("owner-0")) + val threads = 10 + val released = AtomicInteger(0) + val latch = CountDownLatch(threads) + val executor = Executors.newFixedThreadPool(threads) + + repeat(threads) { i -> + executor.submit { + try { + if (store.deleteIfEquals(b("lock"), b("owner-$i"))) { + released.incrementAndGet() + } + } finally { + latch.countDown() + } + } + } + + latch.await() + executor.shutdown() + assert(released.get() == 1) { "Expected 1 but got ${released.get()}" } + } + } + } + + companion object { + fun b(s: String): ByteArray = s.toByteArray() + + fun longToBytes(v: Long): ByteArray = + ByteBuffer + .allocate(8) + .order(ByteOrder.BIG_ENDIAN) + .putLong(v) + .array() + + fun bytesToLong(b: ByteArray): Long = ByteBuffer.wrap(b).order(ByteOrder.BIG_ENDIAN).long + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/HBaseDatastoreCompatibilityTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/HBaseDatastoreCompatibilityTest.kt new file mode 100644 index 00000000..b7d768bb --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/HBaseDatastoreCompatibilityTest.kt @@ -0,0 +1,140 @@ +package com.kakao.actionbase.engine.datastore + +import com.kakao.actionbase.test.hbase.HBaseTestingCluster +import com.kakao.actionbase.v2.engine.storage.hbase.HBaseConnections + +import java.nio.ByteBuffer +import java.nio.ByteOrder + +import org.apache.hadoop.hbase.NamespaceDescriptor +import org.apache.hadoop.hbase.TableName +import org.apache.hadoop.hbase.client.Admin +import org.apache.hadoop.hbase.client.CheckAndMutate +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder +import org.apache.hadoop.hbase.client.Delete +import org.apache.hadoop.hbase.client.Get +import org.apache.hadoop.hbase.client.Increment +import org.apache.hadoop.hbase.client.Put +import org.apache.hadoop.hbase.client.Scan +import org.apache.hadoop.hbase.client.Table +import org.apache.hadoop.hbase.client.TableDescriptorBuilder +import org.apache.hadoop.hbase.filter.PrefixFilter +import org.junit.jupiter.api.AfterAll +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.TestInstance + +/** + * HBase compatibility test. Default: MockConnection. Set HBASE_MINI_CLUSTER=true for mini cluster. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +class HBaseDatastoreCompatibilityTest : DatastoreCompatibilityTest() { + private lateinit var table: Table + private val tableName = TableName.valueOf("test", "compatibility_test") + private val cf = "f".toByteArray() + private val useMiniCluster = System.getenv("HBASE_MINI_CLUSTER") == "true" + + @BeforeAll + fun setUpHBase() { + val connection = + if (useMiniCluster) { + HBaseTestingCluster.startIfNeeded() + HBaseTestingCluster.connection.also { createTableIfNeeded(it.admin) } + } else { + HBaseConnections.getMockConnection("test") + } + table = connection.getTable(tableName) + } + + @AfterAll + fun tearDownHBase() { + table.close() + if (useMiniCluster) HBaseTestingCluster.stopIfNeeded() + } + + override fun createStore(): StorageOperations = HBaseOps(table, cf) + + override fun supportsCheckAndMutate() = useMiniCluster + + override fun supportsScanLimit() = useMiniCluster + + override fun cleanup() { + table.getScanner(Scan()).use { s -> + s.map { Delete(it.row) }.takeIf { it.isNotEmpty() }?.let { table.delete(it) } + } + } + + private fun createTableIfNeeded(admin: Admin) { + val ns = tableName.namespaceAsString + if (admin.listNamespaceDescriptors().none { it.name == ns }) { + admin.createNamespace(NamespaceDescriptor.create(ns).build()) + } + if (!admin.tableExists(tableName)) { + admin.createTable( + TableDescriptorBuilder + .newBuilder(tableName) + .setColumnFamily(ColumnFamilyDescriptorBuilder.of(cf)) + .build(), + ) + } + } + + private class HBaseOps( + private val t: Table, + private val cf: ByteArray, + ) : StorageOperations { + private val q = "v".toByteArray() + + override fun get(key: ByteArray) = t.get(Get(key).addColumn(cf, q)).getValue(cf, q) + + override fun getAll(keys: List) = t.get(keys.map { Get(it).addColumn(cf, q) }).mapNotNull { r -> r.getValue(cf, q)?.let { r.row to it } } + + override fun scan( + prefix: ByteArray, + limit: Int, + ) = t + .getScanner(Scan().setFilter(PrefixFilter(prefix)).addColumn(cf, q).setLimit(limit)) + .use { s -> s.mapNotNull { r -> r.getValue(cf, q)?.let { r.row to it } } } + + override fun put( + key: ByteArray, + value: ByteArray, + ) { + t.put(Put(key).addColumn(cf, q, value)) + } + + override fun delete(key: ByteArray) { + t.delete(Delete(key)) + } + + override fun increment( + key: ByteArray, + delta: Long, + ) = ByteBuffer + .wrap(t.increment(Increment(key).addColumn(cf, q, delta)).getValue(cf, q)) + .order(ByteOrder.BIG_ENDIAN) + .long + + override fun batch(mutations: List) { + if (mutations.isEmpty()) return + val actions = + mutations.map { m -> + when (m) { + is Mutation.Put -> Put(m.key).addColumn(cf, q, m.value) + is Mutation.Delete -> Delete(m.key) + is Mutation.Increment -> Increment(m.key).addColumn(cf, q, m.delta) + } + } + t.batch(actions, arrayOfNulls(actions.size)) + } + + override fun setIfNotExists( + key: ByteArray, + value: ByteArray, + ) = t.checkAndMutate(CheckAndMutate.newBuilder(key).ifNotExists(cf, q).build(Put(key).addColumn(cf, q, value))).isSuccess + + override fun deleteIfEquals( + key: ByteArray, + expectedValue: ByteArray, + ) = t.checkAndMutate(CheckAndMutate.newBuilder(key).ifEquals(cf, q, expectedValue).build(Delete(key))).isSuccess + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/MemoryDatastoreCompatibilityTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/MemoryDatastoreCompatibilityTest.kt new file mode 100644 index 00000000..fa47beeb --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/MemoryDatastoreCompatibilityTest.kt @@ -0,0 +1,61 @@ +package com.kakao.actionbase.engine.datastore + +import com.kakao.actionbase.engine.datastore.impl.ByteArrayStore + +/** Memory (ByteArrayStore) compatibility test. */ +class MemoryDatastoreCompatibilityTest : DatastoreCompatibilityTest() { + private lateinit var store: ByteArrayStore + + override fun createStore(): StorageOperations { + store = ByteArrayStore() + return MemoryOps(store) + } + + private class MemoryOps( + private val s: ByteArrayStore, + ) : StorageOperations { + override fun get(key: ByteArray) = s[key] + + override fun getAll(keys: List) = keys.mapNotNull { k -> s[k]?.let { k to it } } + + override fun scan( + prefix: ByteArray, + limit: Int, + ) = s.prefixScan(prefix).take(limit).map { it.key to it.value } + + override fun put( + key: ByteArray, + value: ByteArray, + ) { + s[key] = value + } + + override fun delete(key: ByteArray) { + s.remove(key) + } + + override fun increment( + key: ByteArray, + delta: Long, + ) = s.increment(key, delta) + + override fun batch(mutations: List) = + mutations.forEach { m -> + when (m) { + is Mutation.Put -> s[m.key] = m.value + is Mutation.Delete -> s.remove(m.key) + is Mutation.Increment -> s.increment(m.key, m.delta) + } + } + + override fun setIfNotExists( + key: ByteArray, + value: ByteArray, + ) = s.checkAndSet(key, null, value) + + override fun deleteIfEquals( + key: ByteArray, + expectedValue: ByteArray, + ) = s.checkAndSet(key, expectedValue, null) + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/SlateDBDatastoreCompatibilityTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/SlateDBDatastoreCompatibilityTest.kt new file mode 100644 index 00000000..cb6a7f52 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/SlateDBDatastoreCompatibilityTest.kt @@ -0,0 +1,126 @@ +package com.kakao.actionbase.engine.datastore + +import com.kakao.actionbase.v2.engine.storage.slatedb.BatchOperation +import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbTable + +import java.nio.ByteBuffer +import java.nio.ByteOrder +import java.nio.file.Path + +import org.junit.jupiter.api.AfterAll +import org.junit.jupiter.api.Assumptions.assumeTrue +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.TestInstance +import org.junit.jupiter.api.io.TempDir + +import io.slatedb.SlateDb +import io.slatedb.SlateDbConfig + +/** + * SlateDB compatibility test. + * + * Disabled by default. Set SLATEDB_TEST=true to run. + * + * To run: + * SLATEDB_TEST=true ./gradlew :engine:test --tests "*SlateDBDatastoreCompatibilityTest*" + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +class SlateDBDatastoreCompatibilityTest : DatastoreCompatibilityTest() { + private var table: SlateDbTable? = null + private lateinit var tempDir: Path + + private val enabled = System.getenv("SLATEDB_TEST") == "true" + + @BeforeAll + fun setUpSlateDB( + @TempDir dir: Path, + ) { + assumeTrue(enabled, "SLATEDB_TEST=true not set") + tempDir = dir + SlateDb.initLogging(SlateDbConfig.LogLevel.INFO) + val db = SlateDb.open("data", "file://${tempDir.toAbsolutePath()}", null) + table = SlateDbTable.create(db) + } + + @AfterAll + fun tearDownSlateDB() { + table?.close() + } + + override fun createStore(): StorageOperations = SlateDBOperations(table!!) + + override fun supportsCheckAndMutate() = false + + override fun cleanup() { + table?.let { t -> + t.scanPrefix(ByteArray(0), Int.MAX_VALUE).block()?.forEach { (key, _) -> + t.delete(key).block() + } + } + } + + private class SlateDBOperations( + private val table: SlateDbTable, + ) : StorageOperations { + override fun get(key: ByteArray): ByteArray? = table.get(key).block() + + override fun getAll(keys: List) = keys.mapNotNull { k -> table.get(k).block()?.let { k to it } } + + override fun scan( + prefix: ByteArray, + limit: Int, + ) = table.scanPrefix(prefix, limit).block() ?: emptyList() + + override fun put( + key: ByteArray, + value: ByteArray, + ) { + table.put(key, value).block() + } + + override fun delete(key: ByteArray) { + table.delete(key).block() + } + + override fun increment( + key: ByteArray, + delta: Long, + ): Long { + val current = + table.get(key).block()?.let { + ByteBuffer.wrap(it).order(ByteOrder.BIG_ENDIAN).long + } ?: 0L + val newValue = current + delta + val bytes = + ByteBuffer + .allocate(8) + .order(ByteOrder.BIG_ENDIAN) + .putLong(newValue) + .array() + table.put(key, bytes).block() + return newValue + } + + override fun batch(mutations: List) { + val ops = + mutations.map { m -> + when (m) { + is Mutation.Put -> BatchOperation.Put(m.key, m.value) + is Mutation.Delete -> BatchOperation.Delete(m.key) + is Mutation.Increment -> BatchOperation.Increment(m.key, m.delta) + } + } + table.batch(ops).block() + } + + override fun setIfNotExists( + key: ByteArray, + value: ByteArray, + ): Boolean = throw UnsupportedOperationException("SlateDB does not support checkAndMutate") + + override fun deleteIfEquals( + key: ByteArray, + expectedValue: ByteArray, + ): Boolean = throw UnsupportedOperationException("SlateDB does not support checkAndMutate") + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/StorageOperations.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/StorageOperations.kt new file mode 100644 index 00000000..142c417c --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/StorageOperations.kt @@ -0,0 +1,57 @@ +package com.kakao.actionbase.engine.datastore + +/** + * Storage operations interface for compatibility testing. + * + * Defines the minimal operations required for Actionbase storage backends. + */ +interface StorageOperations { + fun get(key: ByteArray): ByteArray? + + fun getAll(keys: List): List> + + fun scan( + prefix: ByteArray, + limit: Int, + ): List> + + fun put( + key: ByteArray, + value: ByteArray, + ) + + fun delete(key: ByteArray) + + fun increment( + key: ByteArray, + delta: Long, + ): Long + + fun batch(mutations: List) + + fun setIfNotExists( + key: ByteArray, + value: ByteArray, + ): Boolean + + fun deleteIfEquals( + key: ByteArray, + expectedValue: ByteArray, + ): Boolean +} + +sealed class Mutation { + class Put( + val key: ByteArray, + val value: ByteArray, + ) : Mutation() + + class Delete( + val key: ByteArray, + ) : Mutation() + + class Increment( + val key: ByteArray, + val delta: Long, + ) : Mutation() +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbHashLabelTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbHashLabelTest.kt new file mode 100644 index 00000000..fa6e4d56 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbHashLabelTest.kt @@ -0,0 +1,228 @@ +package com.kakao.actionbase.v2.engine.label.slatedb + +import com.kakao.actionbase.v2.core.metadata.DirectionType +import com.kakao.actionbase.v2.core.metadata.EdgeOperation +import com.kakao.actionbase.v2.core.metadata.LabelType +import com.kakao.actionbase.v2.core.types.DataType +import com.kakao.actionbase.v2.core.types.EdgeSchema +import com.kakao.actionbase.v2.core.types.Field +import com.kakao.actionbase.v2.core.types.VertexField +import com.kakao.actionbase.v2.core.types.VertexType +import com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.GraphConfig +import com.kakao.actionbase.v2.engine.client.kafka.impl.DefaultKafkaClientFactory +import com.kakao.actionbase.v2.engine.client.web.impl.DefaultWebClientFactory +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.entity.LabelEntity +import com.kakao.actionbase.v2.engine.label.EdgeOperationStatus +import com.kakao.actionbase.v2.engine.metadata.StorageType +import com.kakao.actionbase.v2.engine.service.ddl.DdlStatus +import com.kakao.actionbase.v2.engine.service.ddl.ServiceCreateRequest +import com.kakao.actionbase.v2.engine.service.ddl.StorageCreateRequest +import com.kakao.actionbase.v2.engine.test.cdc.InMemoryCdcFactory +import com.kakao.actionbase.v2.engine.test.wal.InMemoryWalFactory + +import java.nio.file.Path +import java.util.UUID + +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper + +import io.kotest.matchers.shouldBe +import reactor.kotlin.test.test + +class SlateDbHashLabelTest { + @TempDir + lateinit var tempDir: Path + + private lateinit var graph: Graph + private val serviceName = "slatedb_test_service" + private val storageName = "slatedb_storage" + private val labelName = "slatedb_label" + + private fun createGraph(): Graph { + val config = + GraphConfig + .Builder() + .withMetastoreUrl("jdbc:h2:mem:${UUID.randomUUID()};DB_CLOSE_DELAY=-1;MODE=MYSQL") + .build() + return Graph.create(config, InMemoryWalFactory, InMemoryCdcFactory, DefaultKafkaClientFactory, DefaultWebClientFactory) + } + + @BeforeEach + fun setUp() { + graph = createGraph() + graph.updateAllMetadata().block() + + // Create service + graph.serviceDdl + .create(EntityName.fromOrigin(serviceName), ServiceCreateRequest(desc = "test service")) + .test() + .assertNext { it.status shouldBe DdlStatus.Status.CREATED } + .verifyComplete() + + // Create SlateDB storage + val conf = + jacksonObjectMapper().createObjectNode().apply { + put("path", "test-data") + put("url", "file://${tempDir.toAbsolutePath()}") + } + + graph.storageDdl + .create(EntityName.fromOrigin(storageName), StorageCreateRequest(desc = "slatedb storage", type = StorageType.SLATEDB, conf = conf)) + .test() + .assertNext { it.status shouldBe DdlStatus.Status.CREATED } + .verifyComplete() + + // Create label with SlateDB storage + val labelEntity = + LabelEntity( + active = true, + name = EntityName(serviceName, labelName), + desc = "test slatedb label", + type = LabelType.HASH, + schema = + EdgeSchema( + VertexField(VertexType.LONG), + VertexField(VertexType.LONG), + listOf( + Field("score", DataType.LONG, false), + Field("memo", DataType.STRING, true), + ), + ), + dirType = DirectionType.OUT, + storage = storageName, + ) + + graph.labelDdl + .create(labelEntity.name, labelEntity.toCreateRequest()) + .test() + .assertNext { it.status shouldBe DdlStatus.Status.CREATED } + .verifyComplete() + + graph.updateAllMetadata().block() + } + + @AfterEach + fun tearDown() { + graph.close() + } + + @Test + fun `insert and get edge`() { + val label = graph.getLabel(EntityName(serviceName, labelName)) + val edge = + com.kakao.actionbase.v2.core.edge.Edge( + System.currentTimeMillis(), + 100L, + 200L, + mapOf("score" to 42L, "memo" to "hello"), + ) + + // Insert + label + .mutate(edge.toTraceEdge(), EdgeOperation.INSERT) + .test() + .assertNext { context -> + context.status shouldBe EdgeOperationStatus.CREATED + }.verifyComplete() + + // Get + graph + .queryGet( + EntityName(serviceName, labelName), + 100L, + 200L, + ).test() + .assertNext { df -> + df.rows.size shouldBe 1 + // tgt is at index 1 in the schema (src, tgt, ts, ...) + val row = df.toRowWithSchema().first() + row.getLong("tgt") shouldBe 200L + }.verifyComplete() + } + + @Test + fun `delete edge`() { + val label = graph.getLabel(EntityName(serviceName, labelName)) + val edge = + com.kakao.actionbase.v2.core.edge.Edge( + System.currentTimeMillis(), + 101L, + 201L, + mapOf("score" to 100L), + ) + + // Insert + label.mutate(edge.toTraceEdge(), EdgeOperation.INSERT).block() + + // Delete + label + .mutate(edge.toTraceEdge(), EdgeOperation.DELETE) + .test() + .assertNext { context -> + context.status shouldBe EdgeOperationStatus.DELETED + }.verifyComplete() + + // Verify deleted (should return empty) + graph + .queryGet( + EntityName(serviceName, labelName), + 101L, + 201L, + ).test() + .assertNext { df -> + df.rows.size shouldBe 0 + }.verifyComplete() + } + + @Test + fun `update edge`() { + val label = graph.getLabel(EntityName(serviceName, labelName)) + val ts = System.currentTimeMillis() + + val edge1 = + com.kakao.actionbase.v2.core.edge.Edge( + ts, + 102L, + 202L, + mapOf("score" to 50L, "memo" to "original"), + ) + + val edge2 = + com.kakao.actionbase.v2.core.edge.Edge( + ts + 1, + 102L, + 202L, + mapOf("score" to 100L, "memo" to "updated"), + ) + + // Insert + label.mutate(edge1.toTraceEdge(), EdgeOperation.INSERT).block() + + // Update + label + .mutate(edge2.toTraceEdge(), EdgeOperation.INSERT) + .test() + .assertNext { context -> + context.status shouldBe EdgeOperationStatus.UPDATED + }.verifyComplete() + + // Verify updated + graph + .queryGet( + EntityName(serviceName, labelName), + 102L, + 202L, + ).test() + .assertNext { df -> + df.rows.size shouldBe 1 + val row = df.toRowWithSchema().first() + row.getString("memo") shouldBe "updated" + }.verifyComplete() + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbIndexedLabelTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbIndexedLabelTest.kt new file mode 100644 index 00000000..470ae134 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbIndexedLabelTest.kt @@ -0,0 +1,284 @@ +package com.kakao.actionbase.v2.engine.label.slatedb + +import com.kakao.actionbase.v2.core.code.Index +import com.kakao.actionbase.v2.core.code.hbase.Order +import com.kakao.actionbase.v2.core.metadata.DirectionType +import com.kakao.actionbase.v2.core.metadata.EdgeOperation +import com.kakao.actionbase.v2.core.metadata.LabelType +import com.kakao.actionbase.v2.core.types.DataType +import com.kakao.actionbase.v2.core.types.EdgeSchema +import com.kakao.actionbase.v2.core.types.Field +import com.kakao.actionbase.v2.core.types.VertexField +import com.kakao.actionbase.v2.core.types.VertexType +import com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.GraphConfig +import com.kakao.actionbase.v2.engine.client.kafka.impl.DefaultKafkaClientFactory +import com.kakao.actionbase.v2.engine.client.web.impl.DefaultWebClientFactory +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.entity.LabelEntity +import com.kakao.actionbase.v2.engine.label.EdgeOperationStatus +import com.kakao.actionbase.v2.engine.metadata.StorageType +import com.kakao.actionbase.v2.engine.service.ddl.DdlStatus +import com.kakao.actionbase.v2.engine.service.ddl.ServiceCreateRequest +import com.kakao.actionbase.v2.engine.service.ddl.StorageCreateRequest +import com.kakao.actionbase.v2.engine.test.cdc.InMemoryCdcFactory +import com.kakao.actionbase.v2.engine.test.wal.InMemoryWalFactory + +import java.nio.file.Path +import java.util.UUID + +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper + +import io.kotest.matchers.shouldBe +import reactor.kotlin.test.test + +class SlateDbIndexedLabelTest { + @TempDir + lateinit var tempDir: Path + + private lateinit var graph: Graph + private val serviceName = "slatedb_indexed_test_service" + private val storageName = "slatedb_indexed_storage" + private val labelName = "slatedb_indexed_label" + + private fun createGraph(): Graph { + val config = + GraphConfig + .Builder() + .withMetastoreUrl("jdbc:h2:mem:${UUID.randomUUID()};DB_CLOSE_DELAY=-1;MODE=MYSQL") + .build() + return Graph.create(config, InMemoryWalFactory, InMemoryCdcFactory, DefaultKafkaClientFactory, DefaultWebClientFactory) + } + + @BeforeEach + fun setUp() { + graph = createGraph() + graph.updateAllMetadata().block() + + // Create service + graph.serviceDdl + .create(EntityName.fromOrigin(serviceName), ServiceCreateRequest(desc = "test service")) + .test() + .assertNext { it.status shouldBe DdlStatus.Status.CREATED } + .verifyComplete() + + // Create SlateDB storage + val conf = + jacksonObjectMapper().createObjectNode().apply { + put("path", "test-data") + put("url", "file://${tempDir.toAbsolutePath()}") + } + + graph.storageDdl + .create(EntityName.fromOrigin(storageName), StorageCreateRequest(desc = "slatedb indexed storage", type = StorageType.SLATEDB, conf = conf)) + .test() + .assertNext { it.status shouldBe DdlStatus.Status.CREATED } + .verifyComplete() + + // Create INDEXED label with SlateDB storage + val labelEntity = + LabelEntity( + active = true, + name = EntityName(serviceName, labelName), + desc = "test slatedb indexed label", + type = LabelType.INDEXED, + schema = + EdgeSchema( + VertexField(VertexType.LONG), + VertexField(VertexType.LONG), + listOf( + Field("score", DataType.LONG, false), + Field("memo", DataType.STRING, true), + ), + ), + dirType = DirectionType.OUT, + storage = storageName, + indices = + listOf( + Index( + "score_desc", + listOf(Index.Field("score", Order.DESC)), + "Score descending index", + ), + ), + ) + + graph.labelDdl + .create(labelEntity.name, labelEntity.toCreateRequest()) + .test() + .assertNext { it.status shouldBe DdlStatus.Status.CREATED } + .verifyComplete() + + graph.updateAllMetadata().block() + } + + @AfterEach + fun tearDown() { + graph.close() + } + + @Test + fun `insert and get edge with index`() { + val label = graph.getLabel(EntityName(serviceName, labelName)) + val edge = + com.kakao.actionbase.v2.core.edge.Edge( + System.currentTimeMillis(), + 100L, + 200L, + mapOf("score" to 42L, "memo" to "hello"), + ) + + // Insert + label + .mutate(edge.toTraceEdge(), EdgeOperation.INSERT) + .test() + .assertNext { context -> + context.status shouldBe EdgeOperationStatus.CREATED + }.verifyComplete() + + // Get + graph + .queryGet( + EntityName(serviceName, labelName), + 100L, + 200L, + ).test() + .assertNext { df -> + df.rows.size shouldBe 1 + val row = df.toRowWithSchema().first() + row.getLong("tgt") shouldBe 200L + row.getLong("score") shouldBe 42L + }.verifyComplete() + } + + @Test + fun `insert multiple edges and verify index order`() { + val label = graph.getLabel(EntityName(serviceName, labelName)) + val ts = System.currentTimeMillis() + + // Insert edges with different scores + val edges = + listOf( + com.kakao.actionbase.v2.core.edge + .Edge(ts, 100L, 201L, mapOf("score" to 10L, "memo" to "low")), + com.kakao.actionbase.v2.core.edge + .Edge(ts + 1, 100L, 202L, mapOf("score" to 50L, "memo" to "medium")), + com.kakao.actionbase.v2.core.edge + .Edge(ts + 2, 100L, 203L, mapOf("score" to 100L, "memo" to "high")), + ) + + edges.forEach { edge -> + label.mutate(edge.toTraceEdge(), EdgeOperation.INSERT).block() + } + + // Verify all edges exist + edges.forEach { edge -> + graph + .queryGet( + EntityName(serviceName, labelName), + edge.src, + edge.tgt, + ).test() + .assertNext { df -> + df.rows.size shouldBe 1 + }.verifyComplete() + } + } + + @Test + fun `delete edge removes from index`() { + val label = graph.getLabel(EntityName(serviceName, labelName)) + val edge = + com.kakao.actionbase.v2.core.edge.Edge( + System.currentTimeMillis(), + 101L, + 201L, + mapOf("score" to 100L), + ) + + // Insert + label.mutate(edge.toTraceEdge(), EdgeOperation.INSERT).block() + + // Verify exists + graph + .queryGet( + EntityName(serviceName, labelName), + 101L, + 201L, + ).test() + .assertNext { df -> + df.rows.size shouldBe 1 + }.verifyComplete() + + // Delete + label + .mutate(edge.toTraceEdge(), EdgeOperation.DELETE) + .test() + .assertNext { context -> + context.status shouldBe EdgeOperationStatus.DELETED + }.verifyComplete() + + // Verify deleted + graph + .queryGet( + EntityName(serviceName, labelName), + 101L, + 201L, + ).test() + .assertNext { df -> + df.rows.size shouldBe 0 + }.verifyComplete() + } + + @Test + fun `update edge updates index`() { + val label = graph.getLabel(EntityName(serviceName, labelName)) + val ts = System.currentTimeMillis() + + val edge1 = + com.kakao.actionbase.v2.core.edge.Edge( + ts, + 102L, + 202L, + mapOf("score" to 50L, "memo" to "original"), + ) + + val edge2 = + com.kakao.actionbase.v2.core.edge.Edge( + ts + 1, + 102L, + 202L, + mapOf("score" to 100L, "memo" to "updated"), + ) + + // Insert + label.mutate(edge1.toTraceEdge(), EdgeOperation.INSERT).block() + + // Update + label + .mutate(edge2.toTraceEdge(), EdgeOperation.INSERT) + .test() + .assertNext { context -> + context.status shouldBe EdgeOperationStatus.UPDATED + }.verifyComplete() + + // Verify updated + graph + .queryGet( + EntityName(serviceName, labelName), + 102L, + 202L, + ).test() + .assertNext { df -> + df.rows.size shouldBe 1 + val row = df.toRowWithSchema().first() + row.getLong("score") shouldBe 100L + row.getString("memo") shouldBe "updated" + }.verifyComplete() + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbStorageTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbStorageTest.kt new file mode 100644 index 00000000..54799982 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbStorageTest.kt @@ -0,0 +1,64 @@ +package com.kakao.actionbase.v2.engine.storage.slatedb + +import com.kakao.actionbase.v2.engine.Graph +import com.kakao.actionbase.v2.engine.GraphConfig +import com.kakao.actionbase.v2.engine.client.kafka.impl.DefaultKafkaClientFactory +import com.kakao.actionbase.v2.engine.client.web.impl.DefaultWebClientFactory +import com.kakao.actionbase.v2.engine.entity.EntityName +import com.kakao.actionbase.v2.engine.metadata.StorageType +import com.kakao.actionbase.v2.engine.test.GraphFixtures +import com.kakao.actionbase.v2.engine.test.cdc.InMemoryCdcFactory +import com.kakao.actionbase.v2.engine.test.wal.InMemoryWalFactory + +import java.util.UUID + +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test + +import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper + +import reactor.kotlin.test.test + +class SlateDbStorageTest { + private lateinit var graph: Graph + + private fun createGraph(): Graph { + val config = + GraphConfig + .Builder() + .withMetastoreUrl("jdbc:h2:mem:${UUID.randomUUID()};DB_CLOSE_DELAY=-1;MODE=MYSQL") + .build() + return Graph.create(config, InMemoryWalFactory, InMemoryCdcFactory, DefaultKafkaClientFactory, DefaultWebClientFactory) + } + + @BeforeEach + fun setUp() { + graph = createGraph() + graph.updateAllMetadata().block() + } + + @AfterEach + fun tearDown() { + graph.close() + } + + @Test + fun `create SlateDB storage entity`() { + val conf = + jacksonObjectMapper().createObjectNode().apply { + put("path", "test-data") + put("url", "file:///tmp/slatedb-test") + } + + GraphFixtures.createStorage(graph, "slatedb_test", StorageType.SLATEDB, conf) + + graph.storageDdl + .getSingle(EntityName.fromOrigin("slatedb_test")) + .test() + .assertNext { storage -> + assert(storage.type == StorageType.SLATEDB) + assert(storage.active) + }.verifyComplete() + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTableTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTableTest.kt new file mode 100644 index 00000000..9cd26215 --- /dev/null +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTableTest.kt @@ -0,0 +1,172 @@ +package com.kakao.actionbase.v2.engine.storage.slatedb + +import java.nio.ByteBuffer +import java.nio.charset.StandardCharsets +import java.nio.file.Path + +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir + +import io.slatedb.SlateDb +import io.slatedb.SlateDbConfig +import reactor.test.StepVerifier + +class SlateDbTableTest { + @TempDir + lateinit var tempDir: Path + + private lateinit var table: SlateDbTable + + @BeforeEach + fun setUp() { + SlateDb.initLogging(SlateDbConfig.LogLevel.INFO) + + val fileUrl = "file://${tempDir.toAbsolutePath()}" + val dbPath = "data" + val db = SlateDb.open(dbPath, fileUrl, null) + table = SlateDbTable.create(db) + } + + @AfterEach + fun tearDown() { + table.close() + } + + @Test + fun `put and get with reactive API`() { + val key = "hello".toByteArray(StandardCharsets.UTF_8) + val value = "world".toByteArray(StandardCharsets.UTF_8) + + StepVerifier + .create( + table.put(key, value).then(table.get(key)), + ).expectNextMatches { it != null && String(it, StandardCharsets.UTF_8) == "world" } + .verifyComplete() + } + + @Test + fun `get non-existent key returns empty`() { + val key = "nonexistent".toByteArray(StandardCharsets.UTF_8) + + StepVerifier + .create(table.get(key)) + .verifyComplete() + } + + @Test + fun `delete removes key`() { + val key = "to-delete".toByteArray(StandardCharsets.UTF_8) + val value = "value".toByteArray(StandardCharsets.UTF_8) + + StepVerifier + .create( + table + .put(key, value) + .then(table.delete(key)) + .then(table.get(key)), + ).verifyComplete() + } + + @Test + fun `flush persists data`() { + val key = "flush-key".toByteArray(StandardCharsets.UTF_8) + val value = "flush-value".toByteArray(StandardCharsets.UTF_8) + + StepVerifier + .create( + table + .put(key, value) + .then(table.flush()) + .then(table.get(key)), + ).expectNextMatches { it != null && String(it, StandardCharsets.UTF_8) == "flush-value" } + .verifyComplete() + } + + @Test + fun `batch writes multiple keys atomically`() { + val key1 = "batch-key-1".toByteArray(StandardCharsets.UTF_8) + val value1 = "batch-value-1".toByteArray(StandardCharsets.UTF_8) + val key2 = "batch-key-2".toByteArray(StandardCharsets.UTF_8) + val value2 = "batch-value-2".toByteArray(StandardCharsets.UTF_8) + val key3 = "batch-key-3".toByteArray(StandardCharsets.UTF_8) + val value3 = "batch-value-3".toByteArray(StandardCharsets.UTF_8) + + val operations = + listOf( + BatchOperation.Put(key1, value1), + BatchOperation.Put(key2, value2), + BatchOperation.Put(key3, value3), + ) + + StepVerifier + .create( + table + .batch(operations) + .then(table.get(key1)), + ).expectNextMatches { String(it, StandardCharsets.UTF_8) == "batch-value-1" } + .verifyComplete() + + StepVerifier + .create(table.get(key2)) + .expectNextMatches { String(it, StandardCharsets.UTF_8) == "batch-value-2" } + .verifyComplete() + } + + @Test + fun `batch with put and delete`() { + val key1 = "batch-put".toByteArray(StandardCharsets.UTF_8) + val value1 = "value".toByteArray(StandardCharsets.UTF_8) + val key2 = "batch-delete".toByteArray(StandardCharsets.UTF_8) + val value2 = "to-be-deleted".toByteArray(StandardCharsets.UTF_8) + + // First, put key2 + StepVerifier + .create(table.put(key2, value2)) + .verifyComplete() + + // Then batch: put key1, delete key2 + val operations = + listOf( + BatchOperation.Put(key1, value1), + BatchOperation.Delete(key2), + ) + + StepVerifier + .create( + table + .batch(operations) + .then(table.get(key1)), + ).expectNextMatches { String(it, StandardCharsets.UTF_8) == "value" } + .verifyComplete() + + // key2 should be deleted + StepVerifier + .create(table.get(key2)) + .verifyComplete() + } + + @Test + fun `batch increment adds delta to value`() { + val key = "counter".toByteArray(StandardCharsets.UTF_8) + + // Increment non-existent key (starts at 0) + StepVerifier + .create( + table + .batch(listOf(BatchOperation.Increment(key, 5))) + .then(table.get(key)), + ).expectNextMatches { ByteBuffer.wrap(it).long == 5L } + .verifyComplete() + + // Increment existing key + StepVerifier + .create( + table + .batch(listOf(BatchOperation.Increment(key, 3))) + .then(table.get(key)), + ).expectNextMatches { ByteBuffer.wrap(it).long == 8L } + .verifyComplete() + } +} diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt index 0bd7c839..ee963bea 100644 --- a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt +++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/test/GraphFixtures.kt @@ -48,7 +48,6 @@ import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper import io.kotest.matchers.collections.shouldContain import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder import io.kotest.matchers.shouldBe -import reactor.blockhound.BlockHound import reactor.core.publisher.Flux import reactor.kotlin.core.publisher.toFlux import reactor.kotlin.test.test @@ -221,11 +220,6 @@ object GraphFixtures { } fun create(withTestData: Boolean = true): Graph { - BlockHound - .builder() - .allowBlockingCallsInside("org.apache.hadoop.hbase.client.mock.MockHTable", "mutateRow") - .install() - val config = GraphConfig .Builder() diff --git a/native/build-slatedb-java.sh b/native/build-slatedb-java.sh new file mode 100755 index 00000000..4033fb4b --- /dev/null +++ b/native/build-slatedb-java.sh @@ -0,0 +1,123 @@ +#!/bin/bash +# +# Build slatedb-java from source and publish to Maven local. +# +# This script clones the slatedb repository (java-include-libs branch), +# builds the Java JAR with bundled native libraries, and publishes to +# ~/.m2/repository so Gradle can resolve it as a local dependency. +# +# Requirements: +# - Rust toolchain (cargo) +# - Java 24+ (for jextract / FFI support) +# - Git +# +# Usage: +# ./native/build-slatedb-java.sh # clone + build + publish +# ./native/build-slatedb-java.sh --clean # remove clone and rebuild +# +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +SLATEDB_DIR="${SCRIPT_DIR}/slatedb" +SLATEDB_JAVA_DIR="${SLATEDB_DIR}/slatedb-java" +REPO_URL="https://github.com/slatedb/slatedb.git" +BRANCH="java-include-libs" + +# --- Parse arguments --- + +if [[ "${1:-}" == "--clean" ]]; then + echo "Cleaning previous clone..." + rm -rf "${SLATEDB_DIR}" +fi + +# --- Prerequisites --- + +check_command() { + if ! command -v "$1" &>/dev/null; then + echo "Error: '$1' is required but not found in PATH." + exit 1 + fi +} + +check_command git +check_command cargo +check_command java + +JAVA_VERSION=$(java -version 2>&1 | head -1 | sed -E 's/.*"([0-9]+).*/\1/') +if [[ "${JAVA_VERSION}" -lt 24 ]]; then + echo "Error: Java 24+ is required for slatedb-java (found Java ${JAVA_VERSION})." + echo "Set JAVA_HOME to a Java 24+ installation." + exit 1 +fi + +echo "Prerequisites OK: git, cargo, java ${JAVA_VERSION}" + +# --- Clone or update --- + +if [[ -d "${SLATEDB_DIR}" ]]; then + echo "Using existing clone at ${SLATEDB_DIR}" + cd "${SLATEDB_DIR}" + git fetch origin "${BRANCH}" + git checkout "${BRANCH}" + git reset --hard "origin/${BRANCH}" +else + echo "Cloning ${REPO_URL} (branch: ${BRANCH})..." + git clone --branch "${BRANCH}" --depth 1 "${REPO_URL}" "${SLATEDB_DIR}" +fi + +# --- Build --- + +echo "Building slatedb-java..." +cd "${SLATEDB_JAVA_DIR}" +./gradlew build -x test + +# --- Publish to Maven local --- +# +# The upstream build.gradle does not include the maven-publish plugin, +# so we inject it via a Gradle init script. + +INIT_SCRIPT=$(mktemp) +trap 'rm -f "${INIT_SCRIPT}"' EXIT + +cat > "${INIT_SCRIPT}" << 'INIT' +allprojects { + apply plugin: 'maven-publish' + + afterEvaluate { + publishing { + publications { + maven(MavenPublication) { + from components.java + } + } + } + } + + // Suppress Gradle module metadata so consumers don't see the + // org.gradle.jvm.version=24 attribute, which would block resolution + // from projects using a lower toolchain version. + tasks.withType(GenerateModuleMetadata) { + enabled = false + } +} +INIT + +echo "Publishing to Maven local..." +./gradlew publishToMavenLocal --init-script "${INIT_SCRIPT}" + +# --- Verify --- + +GAV="io.slatedb:slatedb:0.1.0-SNAPSHOT" +LOCAL_REPO="${HOME}/.m2/repository" +ARTIFACT_DIR="${LOCAL_REPO}/io/slatedb/slatedb/0.1.0-SNAPSHOT" + +if [[ -d "${ARTIFACT_DIR}" ]]; then + echo "" + echo "Published ${GAV} to Maven local:" + ls -la "${ARTIFACT_DIR}"/*.jar 2>/dev/null || true + echo "" + echo "Done. You can now run: ./gradlew :engine:build" +else + echo "Error: Expected artifact not found at ${ARTIFACT_DIR}" + exit 1 +fi diff --git a/server/src/main/resources/application-slatedb.yaml b/server/src/main/resources/application-slatedb.yaml new file mode 100644 index 00000000..9c2537c6 --- /dev/null +++ b/server/src/main/resources/application-slatedb.yaml @@ -0,0 +1,15 @@ +# SlateDB storage profile +# Usage: --spring.profiles.active=slatedb +# +# Native library is bundled in the slatedb JAR (loaded automatically from classpath). + +actionbase: + tenant: ab-slatedb + +kc: + graph: + defaultStorage: + type: SLATEDB + conf: + path: data + url: file://${SLATEDB_STORAGE_PATH:/tmp/actionbase-slatedb}