Skip to content

Commit 2b306fc

Browse files
em3sem3s
committed
refactor(engine): extract shared long encode/decode helpers and deduplicate merge operator
- Add toSlateBytes() and toLong() extension functions in SlateDbTable.kt - Define incrementMergeOperator once, reuse in connections and tests - Remove duplicate ByteBuffer encode/decode from SlateDbHashLabel, tests
1 parent 89ba08d commit 2b306fc

5 files changed

Lines changed: 32 additions & 91 deletions

File tree

‎engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/slatedb/SlateDbHashLabel.kt‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,9 @@ import com.kakao.actionbase.v2.engine.sql.Row
1919
import com.kakao.actionbase.v2.engine.sql.StatKey
2020
import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbStorage
2121
import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbTable
22+
import com.kakao.actionbase.v2.engine.storage.slatedb.toLong
23+
import com.kakao.actionbase.v2.engine.storage.slatedb.toSlateBytes
2224

23-
import java.nio.ByteBuffer
24-
import java.nio.ByteOrder
2525
import java.util.Arrays
2626

2727
import reactor.core.publisher.Mono
@@ -105,13 +105,7 @@ open class SlateDbHashLabel(
105105
acc: Long,
106106
): Mono<List<Any>> =
107107
table.flatMap { tbl ->
108-
val deltaBytes =
109-
ByteBuffer
110-
.allocate(Long.SIZE_BYTES)
111-
.order(ByteOrder.BIG_ENDIAN)
112-
.putLong(acc)
113-
.array()
114-
tbl.merge(key, deltaBytes).thenReturn(emptyList())
108+
tbl.merge(key, acc.toSlateBytes()).thenReturn(emptyList())
115109
}
116110

117111
override fun scanStorage(
@@ -262,7 +256,7 @@ open class SlateDbHashLabel(
262256
srcAndKeys.map { (src, key) ->
263257
tbl
264258
.get(key)
265-
.map { bytes -> ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getLong() }
259+
.map { bytes -> bytes.toLong() }
266260
.defaultIfEmpty(0L)
267261
.map { count -> Row(arrayOf(src, count, dir)) }
268262
},

‎engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbConnections.kt‎

Lines changed: 0 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -2,14 +2,11 @@ package com.kakao.actionbase.v2.engine.storage.slatedb
22

33
import com.kakao.actionbase.v2.engine.util.getLogger
44

5-
import java.nio.ByteBuffer
6-
import java.nio.ByteOrder
75
import java.util.concurrent.ConcurrentHashMap
86
import java.util.concurrent.atomic.AtomicBoolean
97

108
import io.slatedb.SlateDb
119
import io.slatedb.SlateDbConfig
12-
import io.slatedb.SlateDbMergeOperator
1310
import reactor.core.publisher.Mono
1411
import reactor.core.scheduler.Schedulers
1512

@@ -19,22 +16,6 @@ object SlateDbConnections {
1916
private val initialized = AtomicBoolean(false)
2017
private val connections: ConcurrentHashMap<String, Mono<SlateDbTable>> = ConcurrentHashMap()
2118

22-
private val incrementMergeOperator =
23-
SlateDbMergeOperator { _, existingValue, operand ->
24-
val current =
25-
if (existingValue != null) {
26-
ByteBuffer.wrap(existingValue).order(ByteOrder.BIG_ENDIAN).long
27-
} else {
28-
0L
29-
}
30-
val delta = ByteBuffer.wrap(operand).order(ByteOrder.BIG_ENDIAN).long
31-
ByteBuffer
32-
.allocate(Long.SIZE_BYTES)
33-
.order(ByteOrder.BIG_ENDIAN)
34-
.putLong(current + delta)
35-
.array()
36-
}
37-
3819
fun ensureInitialized() {
3920
if (initialized.compareAndSet(false, true)) {
4021
logger.info("Initializing SlateDB (native library loaded from JAR classpath)")

‎engine/src/main/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTable.kt‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,26 @@ import java.nio.ByteOrder
55

66
import io.slatedb.SlateDb
77
import io.slatedb.SlateDbKeyValue
8+
import io.slatedb.SlateDbMergeOperator
89
import reactor.core.publisher.Mono
910
import reactor.core.scheduler.Schedulers
1011

12+
fun Long.toSlateBytes(): ByteArray =
13+
ByteBuffer
14+
.allocate(Long.SIZE_BYTES)
15+
.order(ByteOrder.BIG_ENDIAN)
16+
.putLong(this)
17+
.array()
18+
19+
fun ByteArray.toLong(): Long = ByteBuffer.wrap(this).order(ByteOrder.BIG_ENDIAN).long
20+
21+
val incrementMergeOperator =
22+
SlateDbMergeOperator { _, existingValue, operand ->
23+
val current = existingValue?.toLong() ?: 0L
24+
val delta = operand.toLong()
25+
(current + delta).toSlateBytes()
26+
}
27+
1128
sealed class BatchOperation {
1229
data class Put(
1330
val key: ByteArray,
@@ -119,15 +136,7 @@ internal class SlateDbTableImpl(
119136
when (op) {
120137
is BatchOperation.Put -> batch.put(op.key, op.value)
121138
is BatchOperation.Delete -> batch.delete(op.key)
122-
is BatchOperation.Increment -> {
123-
val deltaBytes =
124-
ByteBuffer
125-
.allocate(Long.SIZE_BYTES)
126-
.order(ByteOrder.BIG_ENDIAN)
127-
.putLong(op.delta)
128-
.array()
129-
batch.merge(op.key, deltaBytes)
130-
}
139+
is BatchOperation.Increment -> batch.merge(op.key, op.delta.toSlateBytes())
131140
}
132141
}
133142
db.write(batch)

‎engine/src/test/kotlin/com/kakao/actionbase/engine/datastore/SlateDBDatastoreCompatibilityTest.kt‎

Lines changed: 6 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,10 @@ package com.kakao.actionbase.engine.datastore
22

33
import com.kakao.actionbase.v2.engine.storage.slatedb.BatchOperation
44
import com.kakao.actionbase.v2.engine.storage.slatedb.SlateDbTable
5+
import com.kakao.actionbase.v2.engine.storage.slatedb.incrementMergeOperator
6+
import com.kakao.actionbase.v2.engine.storage.slatedb.toLong
7+
import com.kakao.actionbase.v2.engine.storage.slatedb.toSlateBytes
58

6-
import java.nio.ByteBuffer
7-
import java.nio.ByteOrder
89
import java.nio.file.Path
910

1011
import org.junit.jupiter.api.AfterAll
@@ -15,7 +16,6 @@ import org.junit.jupiter.api.io.TempDir
1516

1617
import io.slatedb.SlateDb
1718
import io.slatedb.SlateDbConfig
18-
import io.slatedb.SlateDbMergeOperator
1919

2020
/**
2121
* SlateDB compatibility test.
@@ -39,24 +39,9 @@ class SlateDBDatastoreCompatibilityTest : DatastoreCompatibilityTest() {
3939
assumeTrue(enabled, "SLATEDB_TEST=true not set")
4040
tempDir = dir
4141
SlateDb.initLogging(SlateDbConfig.LogLevel.INFO)
42-
val incrementOp =
43-
SlateDbMergeOperator { _, existingValue, operand ->
44-
val current =
45-
if (existingValue != null) {
46-
ByteBuffer.wrap(existingValue).order(ByteOrder.BIG_ENDIAN).long
47-
} else {
48-
0L
49-
}
50-
val delta = ByteBuffer.wrap(operand).order(ByteOrder.BIG_ENDIAN).long
51-
ByteBuffer
52-
.allocate(Long.SIZE_BYTES)
53-
.order(ByteOrder.BIG_ENDIAN)
54-
.putLong(current + delta)
55-
.array()
56-
}
5742
val db =
5843
SlateDb.builder("data", "file://${tempDir.toAbsolutePath()}", null).use { builder ->
59-
builder.withMergeOperator(incrementOp)
44+
builder.withMergeOperator(incrementMergeOperator)
6045
builder.build()
6146
}
6247
table = SlateDbTable.create(db)
@@ -106,16 +91,8 @@ class SlateDBDatastoreCompatibilityTest : DatastoreCompatibilityTest() {
10691
key: ByteArray,
10792
delta: Long,
10893
): Long {
109-
val deltaBytes =
110-
ByteBuffer
111-
.allocate(Long.SIZE_BYTES)
112-
.order(ByteOrder.BIG_ENDIAN)
113-
.putLong(delta)
114-
.array()
115-
table.merge(key, deltaBytes).block()
116-
return table.get(key).block()?.let {
117-
ByteBuffer.wrap(it).order(ByteOrder.BIG_ENDIAN).long
118-
} ?: 0L
94+
table.merge(key, delta.toSlateBytes()).block()
95+
return table.get(key).block()?.toLong() ?: 0L
11996
}
12097

12198
override fun batch(mutations: List<Mutation>) {

‎engine/src/test/kotlin/com/kakao/actionbase/v2/engine/storage/slatedb/SlateDbTableTest.kt‎

Lines changed: 4 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,5 @@
11
package com.kakao.actionbase.v2.engine.storage.slatedb
22

3-
import java.nio.ByteBuffer
4-
import java.nio.ByteOrder
53
import java.nio.charset.StandardCharsets
64
import java.nio.file.Path
75

@@ -12,7 +10,6 @@ import org.junit.jupiter.api.io.TempDir
1210

1311
import io.slatedb.SlateDb
1412
import io.slatedb.SlateDbConfig
15-
import io.slatedb.SlateDbMergeOperator
1613
import reactor.test.StepVerifier
1714

1815
class SlateDbTableTest {
@@ -25,26 +22,9 @@ class SlateDbTableTest {
2522
fun setUp() {
2623
SlateDb.initLogging(SlateDbConfig.LogLevel.INFO)
2724

28-
val fileUrl = "file://${tempDir.toAbsolutePath()}"
29-
val dbPath = "data"
30-
val incrementOp =
31-
SlateDbMergeOperator { _, existingValue, operand ->
32-
val current =
33-
if (existingValue != null) {
34-
ByteBuffer.wrap(existingValue).order(ByteOrder.BIG_ENDIAN).long
35-
} else {
36-
0L
37-
}
38-
val delta = ByteBuffer.wrap(operand).order(ByteOrder.BIG_ENDIAN).long
39-
ByteBuffer
40-
.allocate(Long.SIZE_BYTES)
41-
.order(ByteOrder.BIG_ENDIAN)
42-
.putLong(current + delta)
43-
.array()
44-
}
4525
val db =
46-
SlateDb.builder(dbPath, fileUrl, null).use { builder ->
47-
builder.withMergeOperator(incrementOp)
26+
SlateDb.builder("data", "file://${tempDir.toAbsolutePath()}", null).use { builder ->
27+
builder.withMergeOperator(incrementMergeOperator)
4828
builder.build()
4929
}
5030
table = SlateDbTable.create(db)
@@ -178,7 +158,7 @@ class SlateDbTableTest {
178158
table
179159
.batch(listOf(BatchOperation.Increment(key, 5)))
180160
.then(table.get(key)),
181-
).expectNextMatches { ByteBuffer.wrap(it).order(ByteOrder.BIG_ENDIAN).long == 5L }
161+
).expectNextMatches { it.toLong() == 5L }
182162
.verifyComplete()
183163

184164
// Increment existing key
@@ -187,7 +167,7 @@ class SlateDbTableTest {
187167
table
188168
.batch(listOf(BatchOperation.Increment(key, 3)))
189169
.then(table.get(key)),
190-
).expectNextMatches { ByteBuffer.wrap(it).order(ByteOrder.BIG_ENDIAN).long == 8L }
170+
).expectNextMatches { it.toLong() == 8L }
191171
.verifyComplete()
192172
}
193173
}

0 commit comments

Comments
 (0)