From 0902e1d71d5ec045dafebe800a98ddf2044f6401 Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Mon, 13 Apr 2026 18:36:02 +0900
Subject: [PATCH 01/11] Add Engine skeleton with metadata reload loop (#247)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Phase 1 of the V3-native engine: a composition root with a single child,
a MetadataLoader bound at construction time. PeriodicMetadataLoader
mirrors Graph.startMetastoreReload but the reload body is empty until
phase 2 fills it in.

Engine.create() is the canonical entry point — Spring config, CLI, and
tests all use it, and the engine.runtime package has zero Spring import
and zero v2 import. The Engine class itself only delegates: new
responsibilities arrive as new children rather than as new methods on
Engine. This is the structural defense against the Graph.kt god-class
pattern.

Highlights:

- MetadataLoader is a small interface (bind + AutoCloseable). The bind
  contract is documented: called once by Engine's ctor, must not call
  back into the engine synchronously, idempotent close.
- PeriodicMetadataLoader runs a single Reactor pipeline. Periodic mode
  uses Flux.interval(initialDelay, interval); disabled mode uses
  Mono.delay(initialDelay).flux() for a one-shot. Both branches share
  the same operators, error handler, and disposable so the engine gets
  exactly one initial reload regardless of configuration.
- bind / close are synchronized; the bound flag prevents double-binding
  and unbound-close noise.
- ServerProperties.MetastoreProperties groups the two new settings under
  actionbase.metastore.*; application.yaml mirrors the existing
  kc.graph.metastoreReloadInterval value and notes the two configs run
  side by side until phase 2 completes the cutover.
- Tests cover construction binding, lifecycle delegation, periodic
  scheduling, bind-triggers-exactly-one-reload, the disabled-but-still-
  initial path, and that the configured initial delay actually defers
  the first reload.
---
 .../kakao/actionbase/engine/runtime/Engine.kt |  31 +++++
 .../engine/runtime/MetadataLoader.kt          |  23 ++++
 .../engine/runtime/PeriodicMetadataLoader.kt  | 114 ++++++++++++++++++
 .../actionbase/engine/runtime/EngineTest.kt   |  51 ++++++++
 .../runtime/PeriodicMetadataLoaderTest.kt     |  93 ++++++++++++++
 .../configuration/EngineConfiguration.kt      |  16 +++
 .../server/configuration/ServerProperties.kt  |   8 ++
 server/src/main/resources/application.yaml    |   5 +
 8 files changed, 341 insertions(+)
 create mode 100644 engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
 create mode 100644 engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
 create mode 100644 engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
 create mode 100644 engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
 create mode 100644 engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
 create mode 100644 server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
new file mode 100644
index 00000000..724abab7
--- /dev/null
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
@@ -0,0 +1,31 @@
+package com.kakao.actionbase.engine.runtime
+
+import java.time.Duration
+
+/**
+ * V3-native engine. Composition root and lifecycle handle for the runtime.
+ * See #247.
+ */
+class Engine(
+    private val loader: MetadataLoader,
+) : AutoCloseable {
+    init {
+        loader.bind(this)
+    }
+
+    override fun close() {
+        loader.close()
+    }
+
+    companion object {
+        fun create(
+            metastoreReloadInitialDelay: Duration = Duration.ZERO,
+            metastoreReloadInterval: Duration? = null,
+        ): Engine = Engine(
+            PeriodicMetadataLoader(
+                metastoreReloadInitialDelay,
+                metastoreReloadInterval,
+            ),
+        )
+    }
+}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
new file mode 100644
index 00000000..4a666cd2
--- /dev/null
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
@@ -0,0 +1,23 @@
+package com.kakao.actionbase.engine.runtime
+
+/**
+ * Loads the engine's view of the metastore.
+ *
+ * A `MetadataLoader` is bound to its [Engine] at construction time so it
+ * can read metadata that is itself stored as Actionbase data — the same
+ * self-hosted pattern as the v2 `Graph`. [Engine] depends on this
+ * interface only and never on a concrete implementation.
+ *
+ * ## Lifecycle
+ * - [bind] is called exactly once by [Engine]'s constructor and must not
+ *   call back into the engine synchronously, since the engine is not yet
+ *   fully constructed at that point.
+ * - [close] is called by the engine on shutdown. Implementations must be
+ *   idempotent — `close` may be called multiple times and must not throw.
+ * - Both methods may assume single-threaded invocation by the engine.
+ *
+ * Phase 1 (#247) exposes only lifecycle. Data accessors land in phase 2.
+ */
+interface MetadataLoader : AutoCloseable {
+    fun bind(engine: Engine)
+}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
new file mode 100644
index 00000000..e480d09f
--- /dev/null
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
@@ -0,0 +1,114 @@
+package com.kakao.actionbase.engine.runtime
+
+import java.time.Duration
+import java.time.Instant
+
+import org.slf4j.LoggerFactory
+
+import reactor.core.Disposable
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+
+/**
+ * Pull-based [MetadataLoader] that periodically reloads the metastore.
+ *
+ * `bind` schedules an initial reload to fire after `metastoreReloadInitialDelay`.
+ * If a periodic interval is configured, that initial reload is the first
+ * tick of `Flux.interval(initialDelay, interval)` and the loop continues
+ * every interval afterwards. If the interval is `null`, only the one-shot
+ * initial reload runs.
+ *
+ * Phase 1 (#247) is a bare loop — [reload] does no real loading yet.
+ * Phase 2 will fill it in by reading the metadata labels through the
+ * bound [Engine] (Actionbase metadata is itself stored as Actionbase
+ * data).
+ *
+ * Mirrors `Graph.startMetastoreReload` so the v2 god-class can be replaced
+ * one consumer at a time.
+ *
+ * `close` cancels the subscription but does not await an in-flight reload
+ * already inside `doOnNext`. Phase 1 reloads are pure counter bumps so
+ * the race is harmless; phase 2 will need to revisit this once `reload`
+ * does real I/O.
+ */
+class PeriodicMetadataLoader(
+    private val metastoreReloadInitialDelay: Duration,
+    private val metastoreReloadInterval: Duration?,
+) : MetadataLoader {
+    @Volatile private var reloadCount: Long = 0
+    @Volatile private var lastReloadAt: Instant? = null
+    private var engine: Engine? = null
+    private var bound = false
+    private var disposable: Disposable? = null
+
+    @Synchronized
+    override fun bind(engine: Engine) {
+        if (bound) {
+            log.warn("PeriodicMetadataLoader already bound")
+            return
+        }
+        this.engine = engine
+        bound = true
+
+        val interval = metastoreReloadInterval
+        if (interval == null) {
+            log.info(
+                "metastore periodic reload disabled; one-shot reload after {} ms.",
+                metastoreReloadInitialDelay.toMillis(),
+            )
+        } else {
+            log.info(
+                "Starting Flux.interval for reloading metastore every {} ms after {} ms delay.",
+                interval.toMillis(),
+                metastoreReloadInitialDelay.toMillis(),
+            )
+        }
+
+        val source: Flux<Long> = if (interval == null) {
+            Mono.delay(metastoreReloadInitialDelay).flux()
+        } else {
+            Flux.interval(metastoreReloadInitialDelay, interval)
+        }
+
+        disposable = source
+            .onBackpressureDrop { log.warn("backpressure drop {}", it) }
+            .doOnNext { reload() }
+            .subscribeOn(Schedulers.boundedElastic())
+            .onErrorContinue { error, _ ->
+                log.error(
+                    "Error occurred during metastore reload or unexpected error: {}. Continuing with next interval.",
+                    error.message,
+                    error,
+                )
+            }
+            .subscribe()
+    }
+
+    @Synchronized
+    override fun close() {
+        if (!bound) return
+        log.info("PeriodicMetadataLoader closing after {} reloads", reloadCount)
+        disposable?.dispose()
+        disposable = null
+        engine = null
+    }
+
+    fun reloadCount(): Long = reloadCount
+
+    fun lastReloadAt(): Instant? = lastReloadAt
+
+    private fun reload() {
+        // Guards against an in-flight tick that fires after `close()`
+        // nulled out `engine`. Phase 2 will read metadata through `engine`
+        // here, which makes this null check load-bearing.
+        if (engine == null) return
+        log.debug("reloading metastore")
+        reloadCount++
+        lastReloadAt = Instant.now()
+    }
+
+    companion object {
+        private val log = LoggerFactory.getLogger(PeriodicMetadataLoader::class.java)
+    }
+}
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
new file mode 100644
index 00000000..e5141323
--- /dev/null
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
@@ -0,0 +1,51 @@
+package com.kakao.actionbase.engine.runtime
+
+import kotlin.test.assertEquals
+import kotlin.test.assertSame
+
+import org.junit.jupiter.api.Test
+
+class EngineTest {
+    @Test
+    fun `construction binds loader to self exactly once`() {
+        val loader = FakeLoader()
+        val engine = Engine(loader)
+        assertEquals(1, loader.bindCount)
+        assertSame(engine, loader.bound)
+    }
+
+    @Test
+    fun `close delegates to loader`() {
+        val loader = FakeLoader()
+        val engine = Engine(loader)
+        engine.close()
+        assertEquals(1, loader.closeCount)
+    }
+
+    @Test
+    fun `try-with-resources closes the loader`() {
+        val loader = FakeLoader()
+        Engine(loader).use { /* no-op */ }
+        assertEquals(1, loader.closeCount)
+    }
+
+    @Test
+    fun `create wires defaults without throwing`() {
+        Engine.create().close()
+    }
+
+    private class FakeLoader : MetadataLoader {
+        var bound: Engine? = null
+        var bindCount = 0
+        var closeCount = 0
+
+        override fun bind(engine: Engine) {
+            bound = engine
+            bindCount++
+        }
+
+        override fun close() {
+            closeCount++
+        }
+    }
+}
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
new file mode 100644
index 00000000..619b1ada
--- /dev/null
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
@@ -0,0 +1,93 @@
+package com.kakao.actionbase.engine.runtime
+
+import java.time.Duration
+
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+import org.junit.jupiter.api.Test
+
+class PeriodicMetadataLoaderTest {
+    @Test
+    fun `periodic loop fires on schedule`() {
+        val loader = PeriodicMetadataLoader(
+            metastoreReloadInitialDelay = Duration.ZERO,
+            metastoreReloadInterval = Duration.ofMillis(20),
+        )
+        Engine(loader).use {
+            waitUntil(Duration.ofSeconds(2)) { loader.reloadCount() >= 3 }
+            assertTrue(loader.reloadCount() >= 3)
+            assertNotNull(loader.lastReloadAt())
+        }
+    }
+
+    @Test
+    fun `close halts the loop`() {
+        val loader = PeriodicMetadataLoader(
+            metastoreReloadInitialDelay = Duration.ZERO,
+            metastoreReloadInterval = Duration.ofMillis(20),
+        )
+        val engine = Engine(loader)
+        waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() >= 1 }
+        engine.close()
+        Thread.sleep(50)
+        val snapshot = loader.reloadCount()
+        Thread.sleep(100)
+        assertTrue(loader.reloadCount() == snapshot, "reloadCount should not advance after close")
+    }
+
+    @Test
+    fun `bind triggers exactly one reload before the periodic loop kicks in`() {
+        // A long interval lets us observe the initial reload alone.
+        val loader = PeriodicMetadataLoader(
+            metastoreReloadInitialDelay = Duration.ZERO,
+            metastoreReloadInterval = Duration.ofMinutes(10),
+        )
+        Engine(loader).use {
+            waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
+            // Window between the initial reload and the next periodic tick
+            // should stay at exactly 1.
+            Thread.sleep(50)
+            assertEquals(1, loader.reloadCount())
+        }
+    }
+
+    @Test
+    fun `null interval still runs the initial reload`() {
+        val loader = PeriodicMetadataLoader(
+            metastoreReloadInitialDelay = Duration.ZERO,
+            metastoreReloadInterval = null,
+        )
+        Engine(loader).use {
+            waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
+            // No periodic loop, so the count must stay at 1 forever.
+            Thread.sleep(50)
+            assertEquals(1, loader.reloadCount())
+        }
+    }
+
+    @Test
+    fun `initial delay defers the first reload`() {
+        val loader = PeriodicMetadataLoader(
+            metastoreReloadInitialDelay = Duration.ofMillis(300),
+            metastoreReloadInterval = null,
+        )
+        Engine(loader).use {
+            // Before the delay expires, the reload has not fired yet.
+            Thread.sleep(100)
+            assertEquals(0, loader.reloadCount())
+            // After the delay, the one-shot reload arrives.
+            waitUntil(Duration.ofSeconds(2)) { loader.reloadCount() == 1L }
+        }
+    }
+
+    private fun waitUntil(timeout: Duration, condition: () -> Boolean) {
+        val deadline = System.currentTimeMillis() + timeout.toMillis()
+        while (System.currentTimeMillis() < deadline) {
+            if (condition()) return
+            Thread.sleep(10)
+        }
+        throw AssertionError("condition not met within $timeout")
+    }
+}
diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
new file mode 100644
index 00000000..856b82aa
--- /dev/null
+++ b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
@@ -0,0 +1,16 @@
+package com.kakao.actionbase.server.configuration
+
+import com.kakao.actionbase.engine.runtime.Engine
+
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+
+@Configuration
+class EngineConfiguration {
+    @Bean(destroyMethod = "close")
+    fun engine(properties: ServerProperties): Engine =
+        Engine.create(
+            metastoreReloadInitialDelay = properties.metastore.reloadInitialDelay,
+            metastoreReloadInterval = properties.metastore.reloadInterval,
+        )
+}
diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt
index 77fab174..3a457341 100644
--- a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt
+++ b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt
@@ -3,6 +3,8 @@ package com.kakao.actionbase.server.configuration
 import com.kakao.actionbase.core.metadata.DatastoreDescriptor
 import com.kakao.actionbase.core.metadata.common.DatastoreType
 
+import java.time.Duration
+
 import org.springframework.boot.context.properties.ConfigurationProperties
 
 // NOTE: If DatastoreProperties is placed in a submodule, IntelliJ's application.yaml -> code navigation does not work.
@@ -12,7 +14,13 @@ data class ServerProperties(
     val tenant: String,
     val datastore: DatastoreProperties,
     val readOnly: Boolean = false,
+    val metastore: MetastoreProperties = MetastoreProperties(),
 ) {
+    data class MetastoreProperties(
+        val reloadInitialDelay: Duration = Duration.ZERO,
+        val reloadInterval: Duration? = null,
+    )
+
     data class DatastoreProperties(
         val type: DatastoreType,
         val configuration: Map<String, String> = emptyMap(),
diff --git a/server/src/main/resources/application.yaml b/server/src/main/resources/application.yaml
index 5e8de487..873d3b5f 100644
--- a/server/src/main/resources/application.yaml
+++ b/server/src/main/resources/application.yaml
@@ -16,6 +16,11 @@ actionbase:
   tenant: ${AB_TENANT:ab-none}
   datastore:
     type: memory
+  metastore:
+    # Phase 1 of #247 — runs alongside kc.graph.metastoreReloadInterval
+    # until the v3 engine takes over the query path.
+    reload-initial-delay: 0s
+    reload-interval: 1m
 
 kc:
   graph:

From 15ab77f8c8636a72c1edfc97522a60a94966344c Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Mon, 13 Apr 2026 18:39:52 +0900
Subject: [PATCH 02/11] Trim Engine runtime KDoc and rename loader field

- Drop KDoc blocks from Engine, MetadataLoader, and PeriodicMetadataLoader.
- Rename Engine's ctor parameter from `loader` to `metadataLoader` for symmetry with the property in `create`.
- Reformat `Engine.create` as a block body with a named local for clarity.
---
 .../kakao/actionbase/engine/runtime/Engine.kt | 19 ++++++-------
 .../engine/runtime/MetadataLoader.kt          | 18 ------------
 .../engine/runtime/PeriodicMetadataLoader.kt  | 28 ++-----------------
 3 files changed, 10 insertions(+), 55 deletions(-)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
index 724abab7..99f59fda 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
@@ -2,30 +2,27 @@ package com.kakao.actionbase.engine.runtime
 
 import java.time.Duration
 
-/**
- * V3-native engine. Composition root and lifecycle handle for the runtime.
- * See #247.
- */
 class Engine(
-    private val loader: MetadataLoader,
+    private val metadataLoader: MetadataLoader,
 ) : AutoCloseable {
     init {
-        loader.bind(this)
+        metadataLoader.bind(this)
     }
 
     override fun close() {
-        loader.close()
+        metadataLoader.close()
     }
 
     companion object {
         fun create(
             metastoreReloadInitialDelay: Duration = Duration.ZERO,
             metastoreReloadInterval: Duration? = null,
-        ): Engine = Engine(
-            PeriodicMetadataLoader(
+        ): Engine {
+            val metadataLoader = PeriodicMetadataLoader(
                 metastoreReloadInitialDelay,
                 metastoreReloadInterval,
-            ),
-        )
+            )
+            return Engine(metadataLoader)
+        }
     }
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
index 4a666cd2..e1fa9077 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
@@ -1,23 +1,5 @@
 package com.kakao.actionbase.engine.runtime
 
-/**
- * Loads the engine's view of the metastore.
- *
- * A `MetadataLoader` is bound to its [Engine] at construction time so it
- * can read metadata that is itself stored as Actionbase data — the same
- * self-hosted pattern as the v2 `Graph`. [Engine] depends on this
- * interface only and never on a concrete implementation.
- *
- * ## Lifecycle
- * - [bind] is called exactly once by [Engine]'s constructor and must not
- *   call back into the engine synchronously, since the engine is not yet
- *   fully constructed at that point.
- * - [close] is called by the engine on shutdown. Implementations must be
- *   idempotent — `close` may be called multiple times and must not throw.
- * - Both methods may assume single-threaded invocation by the engine.
- *
- * Phase 1 (#247) exposes only lifecycle. Data accessors land in phase 2.
- */
 interface MetadataLoader : AutoCloseable {
     fun bind(engine: Engine)
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
index e480d09f..4e8c1c80 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
@@ -1,37 +1,13 @@
 package com.kakao.actionbase.engine.runtime
 
-import java.time.Duration
-import java.time.Instant
-
 import org.slf4j.LoggerFactory
-
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
+import java.time.Duration
+import java.time.Instant
 
-/**
- * Pull-based [MetadataLoader] that periodically reloads the metastore.
- *
- * `bind` schedules an initial reload to fire after `metastoreReloadInitialDelay`.
- * If a periodic interval is configured, that initial reload is the first
- * tick of `Flux.interval(initialDelay, interval)` and the loop continues
- * every interval afterwards. If the interval is `null`, only the one-shot
- * initial reload runs.
- *
- * Phase 1 (#247) is a bare loop — [reload] does no real loading yet.
- * Phase 2 will fill it in by reading the metadata labels through the
- * bound [Engine] (Actionbase metadata is itself stored as Actionbase
- * data).
- *
- * Mirrors `Graph.startMetastoreReload` so the v2 god-class can be replaced
- * one consumer at a time.
- *
- * `close` cancels the subscription but does not await an in-flight reload
- * already inside `doOnNext`. Phase 1 reloads are pure counter bumps so
- * the race is harmless; phase 2 will need to revisit this once `reload`
- * does real I/O.
- */
 class PeriodicMetadataLoader(
     private val metastoreReloadInitialDelay: Duration,
     private val metastoreReloadInterval: Duration?,

From 15d64f74ebfa5621f880df9d6acf78fb12dbe9a6 Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Mon, 13 Apr 2026 18:40:54 +0900
Subject: [PATCH 03/11] Apply spotless to engine runtime files

---
 .../kakao/actionbase/engine/runtime/Engine.kt |  9 ++--
 .../engine/runtime/PeriodicMetadataLoader.kt  | 42 +++++++++-------
 .../runtime/PeriodicMetadataLoaderTest.kt     | 50 +++++++++++--------
 3 files changed, 57 insertions(+), 44 deletions(-)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
index 99f59fda..e2faa858 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
@@ -18,10 +18,11 @@ class Engine(
             metastoreReloadInitialDelay: Duration = Duration.ZERO,
             metastoreReloadInterval: Duration? = null,
         ): Engine {
-            val metadataLoader = PeriodicMetadataLoader(
-                metastoreReloadInitialDelay,
-                metastoreReloadInterval,
-            )
+            val metadataLoader =
+                PeriodicMetadataLoader(
+                    metastoreReloadInitialDelay,
+                    metastoreReloadInterval,
+                )
             return Engine(metadataLoader)
         }
     }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
index 4e8c1c80..a900ddb2 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
@@ -1,18 +1,21 @@
 package com.kakao.actionbase.engine.runtime
 
+import java.time.Duration
+import java.time.Instant
+
 import org.slf4j.LoggerFactory
+
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
-import java.time.Duration
-import java.time.Instant
 
 class PeriodicMetadataLoader(
     private val metastoreReloadInitialDelay: Duration,
     private val metastoreReloadInterval: Duration?,
 ) : MetadataLoader {
     @Volatile private var reloadCount: Long = 0
+
     @Volatile private var lastReloadAt: Instant? = null
     private var engine: Engine? = null
     private var bound = false
@@ -41,24 +44,25 @@ class PeriodicMetadataLoader(
             )
         }
 
-        val source: Flux<Long> = if (interval == null) {
-            Mono.delay(metastoreReloadInitialDelay).flux()
-        } else {
-            Flux.interval(metastoreReloadInitialDelay, interval)
-        }
-
-        disposable = source
-            .onBackpressureDrop { log.warn("backpressure drop {}", it) }
-            .doOnNext { reload() }
-            .subscribeOn(Schedulers.boundedElastic())
-            .onErrorContinue { error, _ ->
-                log.error(
-                    "Error occurred during metastore reload or unexpected error: {}. Continuing with next interval.",
-                    error.message,
-                    error,
-                )
+        val source: Flux<Long> =
+            if (interval == null) {
+                Mono.delay(metastoreReloadInitialDelay).flux()
+            } else {
+                Flux.interval(metastoreReloadInitialDelay, interval)
             }
-            .subscribe()
+
+        disposable =
+            source
+                .onBackpressureDrop { log.warn("backpressure drop {}", it) }
+                .doOnNext { reload() }
+                .subscribeOn(Schedulers.boundedElastic())
+                .onErrorContinue { error, _ ->
+                    log.error(
+                        "Error occurred during metastore reload or unexpected error: {}. Continuing with next interval.",
+                        error.message,
+                        error,
+                    )
+                }.subscribe()
     }
 
     @Synchronized
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
index 619b1ada..2b73ba12 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
@@ -11,10 +11,11 @@ import org.junit.jupiter.api.Test
 class PeriodicMetadataLoaderTest {
     @Test
     fun `periodic loop fires on schedule`() {
-        val loader = PeriodicMetadataLoader(
-            metastoreReloadInitialDelay = Duration.ZERO,
-            metastoreReloadInterval = Duration.ofMillis(20),
-        )
+        val loader =
+            PeriodicMetadataLoader(
+                metastoreReloadInitialDelay = Duration.ZERO,
+                metastoreReloadInterval = Duration.ofMillis(20),
+            )
         Engine(loader).use {
             waitUntil(Duration.ofSeconds(2)) { loader.reloadCount() >= 3 }
             assertTrue(loader.reloadCount() >= 3)
@@ -24,10 +25,11 @@ class PeriodicMetadataLoaderTest {
 
     @Test
     fun `close halts the loop`() {
-        val loader = PeriodicMetadataLoader(
-            metastoreReloadInitialDelay = Duration.ZERO,
-            metastoreReloadInterval = Duration.ofMillis(20),
-        )
+        val loader =
+            PeriodicMetadataLoader(
+                metastoreReloadInitialDelay = Duration.ZERO,
+                metastoreReloadInterval = Duration.ofMillis(20),
+            )
         val engine = Engine(loader)
         waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() >= 1 }
         engine.close()
@@ -40,10 +42,11 @@ class PeriodicMetadataLoaderTest {
     @Test
     fun `bind triggers exactly one reload before the periodic loop kicks in`() {
         // A long interval lets us observe the initial reload alone.
-        val loader = PeriodicMetadataLoader(
-            metastoreReloadInitialDelay = Duration.ZERO,
-            metastoreReloadInterval = Duration.ofMinutes(10),
-        )
+        val loader =
+            PeriodicMetadataLoader(
+                metastoreReloadInitialDelay = Duration.ZERO,
+                metastoreReloadInterval = Duration.ofMinutes(10),
+            )
         Engine(loader).use {
             waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
             // Window between the initial reload and the next periodic tick
@@ -55,10 +58,11 @@ class PeriodicMetadataLoaderTest {
 
     @Test
     fun `null interval still runs the initial reload`() {
-        val loader = PeriodicMetadataLoader(
-            metastoreReloadInitialDelay = Duration.ZERO,
-            metastoreReloadInterval = null,
-        )
+        val loader =
+            PeriodicMetadataLoader(
+                metastoreReloadInitialDelay = Duration.ZERO,
+                metastoreReloadInterval = null,
+            )
         Engine(loader).use {
             waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
             // No periodic loop, so the count must stay at 1 forever.
@@ -69,10 +73,11 @@ class PeriodicMetadataLoaderTest {
 
     @Test
     fun `initial delay defers the first reload`() {
-        val loader = PeriodicMetadataLoader(
-            metastoreReloadInitialDelay = Duration.ofMillis(300),
-            metastoreReloadInterval = null,
-        )
+        val loader =
+            PeriodicMetadataLoader(
+                metastoreReloadInitialDelay = Duration.ofMillis(300),
+                metastoreReloadInterval = null,
+            )
         Engine(loader).use {
             // Before the delay expires, the reload has not fired yet.
             Thread.sleep(100)
@@ -82,7 +87,10 @@ class PeriodicMetadataLoaderTest {
         }
     }
 
-    private fun waitUntil(timeout: Duration, condition: () -> Boolean) {
+    private fun waitUntil(
+        timeout: Duration,
+        condition: () -> Boolean,
+    ) {
         val deadline = System.currentTimeMillis() + timeout.toMillis()
         while (System.currentTimeMillis() < deadline) {
             if (condition()) return

From fcae7b4c5fd0d6eee0d19a520ed40c3a2fcb6aa4 Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Mon, 13 Apr 2026 18:48:30 +0900
Subject: [PATCH 04/11] Rename Metadata to Catalog in engine runtime
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The engine runtime tracks Database, Table, and Alias definitions —
exactly what the industry calls a catalog (Spark/Trino/Iceberg/Glue
all use this term). Metadata was too broad; Catalog names what we
actually load.

Renames:
- MetadataLoader -> CatalogLoader
- PeriodicMetadataLoader -> PeriodicCatalogLoader
- Engine.metadataLoader -> catalogLoader
- metastoreReloadInitialDelay/Interval -> catalogReloadInitialDelay/Interval
- ServerProperties.metastore -> catalog (CatalogProperties)
- actionbase.metastore.* -> actionbase.catalog.*
- log strings: 'reloading metastore' -> 'reloading catalog' etc.

Metastore (the JDBC backing store) keeps its name; the catalog is
what lives in memory and is loaded *from* the metastore.
---
 .../{MetadataLoader.kt => CatalogLoader.kt}   |  2 +-
 .../kakao/actionbase/engine/runtime/Engine.kt | 20 +++++------
 ...dataLoader.kt => PeriodicCatalogLoader.kt} | 36 +++++++++----------
 .../actionbase/engine/runtime/EngineTest.kt   |  2 +-
 ...erTest.kt => PeriodicCatalogLoaderTest.kt} | 32 ++++++++---------
 .../configuration/EngineConfiguration.kt      |  4 +--
 .../server/configuration/ServerProperties.kt  |  4 +--
 server/src/main/resources/application.yaml    |  2 +-
 8 files changed, 51 insertions(+), 51 deletions(-)
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/{MetadataLoader.kt => CatalogLoader.kt} (63%)
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/{PeriodicMetadataLoader.kt => PeriodicCatalogLoader.kt} (61%)
 rename engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/{PeriodicMetadataLoaderTest.kt => PeriodicCatalogLoaderTest.kt} (76%)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/CatalogLoader.kt
similarity index 63%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/CatalogLoader.kt
index e1fa9077..0fa74869 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/MetadataLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/CatalogLoader.kt
@@ -1,5 +1,5 @@
 package com.kakao.actionbase.engine.runtime
 
-interface MetadataLoader : AutoCloseable {
+interface CatalogLoader : AutoCloseable {
     fun bind(engine: Engine)
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
index e2faa858..9151340e 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
@@ -3,27 +3,27 @@ package com.kakao.actionbase.engine.runtime
 import java.time.Duration
 
 class Engine(
-    private val metadataLoader: MetadataLoader,
+    private val catalogLoader: CatalogLoader,
 ) : AutoCloseable {
     init {
-        metadataLoader.bind(this)
+        catalogLoader.bind(this)
     }
 
     override fun close() {
-        metadataLoader.close()
+        catalogLoader.close()
     }
 
     companion object {
         fun create(
-            metastoreReloadInitialDelay: Duration = Duration.ZERO,
-            metastoreReloadInterval: Duration? = null,
+            catalogReloadInitialDelay: Duration = Duration.ZERO,
+            catalogReloadInterval: Duration? = null,
         ): Engine {
-            val metadataLoader =
-                PeriodicMetadataLoader(
-                    metastoreReloadInitialDelay,
-                    metastoreReloadInterval,
+            val catalogLoader =
+                PeriodicCatalogLoader(
+                    catalogReloadInitialDelay,
+                    catalogReloadInterval,
                 )
-            return Engine(metadataLoader)
+            return Engine(catalogLoader)
         }
     }
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoader.kt
similarity index 61%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoader.kt
index a900ddb2..ea9cf5be 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoader.kt
@@ -10,10 +10,10 @@ import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 
-class PeriodicMetadataLoader(
-    private val metastoreReloadInitialDelay: Duration,
-    private val metastoreReloadInterval: Duration?,
-) : MetadataLoader {
+class PeriodicCatalogLoader(
+    private val catalogReloadInitialDelay: Duration,
+    private val catalogReloadInterval: Duration?,
+) : CatalogLoader {
     @Volatile private var reloadCount: Long = 0
 
     @Volatile private var lastReloadAt: Instant? = null
@@ -24,31 +24,31 @@ class PeriodicMetadataLoader(
     @Synchronized
     override fun bind(engine: Engine) {
         if (bound) {
-            log.warn("PeriodicMetadataLoader already bound")
+            log.warn("PeriodicCatalogLoader already bound")
             return
         }
         this.engine = engine
         bound = true
 
-        val interval = metastoreReloadInterval
+        val interval = catalogReloadInterval
         if (interval == null) {
             log.info(
-                "metastore periodic reload disabled; one-shot reload after {} ms.",
-                metastoreReloadInitialDelay.toMillis(),
+                "catalog periodic reload disabled; one-shot reload after {} ms.",
+                catalogReloadInitialDelay.toMillis(),
             )
         } else {
             log.info(
-                "Starting Flux.interval for reloading metastore every {} ms after {} ms delay.",
+                "Starting Flux.interval for reloading catalog every {} ms after {} ms delay.",
                 interval.toMillis(),
-                metastoreReloadInitialDelay.toMillis(),
+                catalogReloadInitialDelay.toMillis(),
             )
         }
 
         val source: Flux<Long> =
             if (interval == null) {
-                Mono.delay(metastoreReloadInitialDelay).flux()
+                Mono.delay(catalogReloadInitialDelay).flux()
             } else {
-                Flux.interval(metastoreReloadInitialDelay, interval)
+                Flux.interval(catalogReloadInitialDelay, interval)
             }
 
         disposable =
@@ -58,7 +58,7 @@ class PeriodicMetadataLoader(
                 .subscribeOn(Schedulers.boundedElastic())
                 .onErrorContinue { error, _ ->
                     log.error(
-                        "Error occurred during metastore reload or unexpected error: {}. Continuing with next interval.",
+                        "Error occurred during catalog reload or unexpected error: {}. Continuing with next interval.",
                         error.message,
                         error,
                     )
@@ -68,7 +68,7 @@ class PeriodicMetadataLoader(
     @Synchronized
     override fun close() {
         if (!bound) return
-        log.info("PeriodicMetadataLoader closing after {} reloads", reloadCount)
+        log.info("PeriodicCatalogLoader closing after {} reloads", reloadCount)
         disposable?.dispose()
         disposable = null
         engine = null
@@ -80,15 +80,15 @@ class PeriodicMetadataLoader(
 
     private fun reload() {
         // Guards against an in-flight tick that fires after `close()`
-        // nulled out `engine`. Phase 2 will read metadata through `engine`
-        // here, which makes this null check load-bearing.
+        // nulled out `engine`. Phase 2 will read the catalog through
+        // `engine` here, which makes this null check load-bearing.
         if (engine == null) return
-        log.debug("reloading metastore")
+        log.debug("reloading catalog")
         reloadCount++
         lastReloadAt = Instant.now()
     }
 
     companion object {
-        private val log = LoggerFactory.getLogger(PeriodicMetadataLoader::class.java)
+        private val log = LoggerFactory.getLogger(PeriodicCatalogLoader::class.java)
     }
 }
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
index e5141323..5ccdb69f 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
@@ -34,7 +34,7 @@ class EngineTest {
         Engine.create().close()
     }
 
-    private class FakeLoader : MetadataLoader {
+    private class FakeLoader : CatalogLoader {
         var bound: Engine? = null
         var bindCount = 0
         var closeCount = 0
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoaderTest.kt
similarity index 76%
rename from engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
rename to engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoaderTest.kt
index 2b73ba12..40092420 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicMetadataLoaderTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoaderTest.kt
@@ -8,13 +8,13 @@ import kotlin.test.assertTrue
 
 import org.junit.jupiter.api.Test
 
-class PeriodicMetadataLoaderTest {
+class PeriodicCatalogLoaderTest {
     @Test
     fun `periodic loop fires on schedule`() {
         val loader =
-            PeriodicMetadataLoader(
-                metastoreReloadInitialDelay = Duration.ZERO,
-                metastoreReloadInterval = Duration.ofMillis(20),
+            PeriodicCatalogLoader(
+                catalogReloadInitialDelay = Duration.ZERO,
+                catalogReloadInterval = Duration.ofMillis(20),
             )
         Engine(loader).use {
             waitUntil(Duration.ofSeconds(2)) { loader.reloadCount() >= 3 }
@@ -26,9 +26,9 @@ class PeriodicMetadataLoaderTest {
     @Test
     fun `close halts the loop`() {
         val loader =
-            PeriodicMetadataLoader(
-                metastoreReloadInitialDelay = Duration.ZERO,
-                metastoreReloadInterval = Duration.ofMillis(20),
+            PeriodicCatalogLoader(
+                catalogReloadInitialDelay = Duration.ZERO,
+                catalogReloadInterval = Duration.ofMillis(20),
             )
         val engine = Engine(loader)
         waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() >= 1 }
@@ -43,9 +43,9 @@ class PeriodicMetadataLoaderTest {
     fun `bind triggers exactly one reload before the periodic loop kicks in`() {
         // A long interval lets us observe the initial reload alone.
         val loader =
-            PeriodicMetadataLoader(
-                metastoreReloadInitialDelay = Duration.ZERO,
-                metastoreReloadInterval = Duration.ofMinutes(10),
+            PeriodicCatalogLoader(
+                catalogReloadInitialDelay = Duration.ZERO,
+                catalogReloadInterval = Duration.ofMinutes(10),
             )
         Engine(loader).use {
             waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
@@ -59,9 +59,9 @@ class PeriodicMetadataLoaderTest {
     @Test
     fun `null interval still runs the initial reload`() {
         val loader =
-            PeriodicMetadataLoader(
-                metastoreReloadInitialDelay = Duration.ZERO,
-                metastoreReloadInterval = null,
+            PeriodicCatalogLoader(
+                catalogReloadInitialDelay = Duration.ZERO,
+                catalogReloadInterval = null,
             )
         Engine(loader).use {
             waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
@@ -74,9 +74,9 @@ class PeriodicMetadataLoaderTest {
     @Test
     fun `initial delay defers the first reload`() {
         val loader =
-            PeriodicMetadataLoader(
-                metastoreReloadInitialDelay = Duration.ofMillis(300),
-                metastoreReloadInterval = null,
+            PeriodicCatalogLoader(
+                catalogReloadInitialDelay = Duration.ofMillis(300),
+                catalogReloadInterval = null,
             )
         Engine(loader).use {
             // Before the delay expires, the reload has not fired yet.
diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
index 856b82aa..704ebc29 100644
--- a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
+++ b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
@@ -10,7 +10,7 @@ class EngineConfiguration {
     @Bean(destroyMethod = "close")
     fun engine(properties: ServerProperties): Engine =
         Engine.create(
-            metastoreReloadInitialDelay = properties.metastore.reloadInitialDelay,
-            metastoreReloadInterval = properties.metastore.reloadInterval,
+            catalogReloadInitialDelay = properties.catalog.reloadInitialDelay,
+            catalogReloadInterval = properties.catalog.reloadInterval,
         )
 }
diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt
index 3a457341..61853ae0 100644
--- a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt
+++ b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/ServerProperties.kt
@@ -14,9 +14,9 @@ data class ServerProperties(
     val tenant: String,
     val datastore: DatastoreProperties,
     val readOnly: Boolean = false,
-    val metastore: MetastoreProperties = MetastoreProperties(),
+    val catalog: CatalogProperties = CatalogProperties(),
 ) {
-    data class MetastoreProperties(
+    data class CatalogProperties(
         val reloadInitialDelay: Duration = Duration.ZERO,
         val reloadInterval: Duration? = null,
     )
diff --git a/server/src/main/resources/application.yaml b/server/src/main/resources/application.yaml
index 873d3b5f..34b7e93a 100644
--- a/server/src/main/resources/application.yaml
+++ b/server/src/main/resources/application.yaml
@@ -16,7 +16,7 @@ actionbase:
   tenant: ${AB_TENANT:ab-none}
   datastore:
     type: memory
-  metastore:
+  catalog:
     # Phase 1 of #247 — runs alongside kc.graph.metastoreReloadInterval
     # until the v3 engine takes over the query path.
     reload-initial-delay: 0s

From 5f214a367a2a1ad669760d11ea7e07d8adda4665 Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Mon, 13 Apr 2026 18:55:10 +0900
Subject: [PATCH 05/11] Move Engine to engine root and catalog into a
 sub-package

- com.kakao.actionbase.engine.Engine (was engine.runtime.Engine).
  Engine now sits next to QueryEngine, MutationEngine, and the
  Actionbase stub it will eventually replace.
- com.kakao.actionbase.engine.catalog.{CatalogLoader, PeriodicCatalogLoader}
  (was engine.runtime.*). Future catalog types (CatalogSnapshot,
  TableId, push-based loaders) get a natural home.
- The empty engine.runtime package is removed.
---
 .../com/kakao/actionbase/engine/{runtime => }/Engine.kt    | 5 ++++-
 .../com/kakao/actionbase/engine/catalog/CatalogLoader.kt   | 7 +++++++
 .../engine/{runtime => catalog}/PeriodicCatalogLoader.kt   | 4 +++-
 .../com/kakao/actionbase/engine/runtime/CatalogLoader.kt   | 5 -----
 .../kakao/actionbase/engine/{runtime => }/EngineTest.kt    | 4 +++-
 .../{runtime => catalog}/PeriodicCatalogLoaderTest.kt      | 4 +++-
 .../actionbase/server/configuration/EngineConfiguration.kt | 2 +-
 7 files changed, 21 insertions(+), 10 deletions(-)
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/{runtime => }/Engine.kt (80%)
 create mode 100644 engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/CatalogLoader.kt
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/{runtime => catalog}/PeriodicCatalogLoader.kt (97%)
 delete mode 100644 engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/CatalogLoader.kt
 rename engine/src/test/kotlin/com/kakao/actionbase/engine/{runtime => }/EngineTest.kt (92%)
 rename engine/src/test/kotlin/com/kakao/actionbase/engine/{runtime => catalog}/PeriodicCatalogLoaderTest.kt (97%)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/Engine.kt
similarity index 80%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/Engine.kt
index 9151340e..8e89d66b 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/Engine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/Engine.kt
@@ -1,4 +1,7 @@
-package com.kakao.actionbase.engine.runtime
+package com.kakao.actionbase.engine
+
+import com.kakao.actionbase.engine.catalog.CatalogLoader
+import com.kakao.actionbase.engine.catalog.PeriodicCatalogLoader
 
 import java.time.Duration
 
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/CatalogLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/CatalogLoader.kt
new file mode 100644
index 00000000..f47e6a04
--- /dev/null
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/CatalogLoader.kt
@@ -0,0 +1,7 @@
+package com.kakao.actionbase.engine.catalog
+
+import com.kakao.actionbase.engine.Engine
+
+interface CatalogLoader : AutoCloseable {
+    fun bind(engine: Engine)
+}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoader.kt
similarity index 97%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoader.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoader.kt
index ea9cf5be..e291d823 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoader.kt
@@ -1,4 +1,6 @@
-package com.kakao.actionbase.engine.runtime
+package com.kakao.actionbase.engine.catalog
+
+import com.kakao.actionbase.engine.Engine
 
 import java.time.Duration
 import java.time.Instant
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/CatalogLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/CatalogLoader.kt
deleted file mode 100644
index 0fa74869..00000000
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/runtime/CatalogLoader.kt
+++ /dev/null
@@ -1,5 +0,0 @@
-package com.kakao.actionbase.engine.runtime
-
-interface CatalogLoader : AutoCloseable {
-    fun bind(engine: Engine)
-}
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
similarity index 92%
rename from engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
rename to engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
index 5ccdb69f..7b25b9ad 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/EngineTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
@@ -1,4 +1,6 @@
-package com.kakao.actionbase.engine.runtime
+package com.kakao.actionbase.engine
+
+import com.kakao.actionbase.engine.catalog.CatalogLoader
 
 import kotlin.test.assertEquals
 import kotlin.test.assertSame
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoaderTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoaderTest.kt
similarity index 97%
rename from engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoaderTest.kt
rename to engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoaderTest.kt
index 40092420..fbf2b405 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/runtime/PeriodicCatalogLoaderTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoaderTest.kt
@@ -1,4 +1,6 @@
-package com.kakao.actionbase.engine.runtime
+package com.kakao.actionbase.engine.catalog
+
+import com.kakao.actionbase.engine.Engine
 
 import java.time.Duration
 
diff --git a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
index 704ebc29..131fd220 100644
--- a/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
+++ b/server/src/main/kotlin/com/kakao/actionbase/server/configuration/EngineConfiguration.kt
@@ -1,6 +1,6 @@
 package com.kakao.actionbase.server.configuration
 
-import com.kakao.actionbase.engine.runtime.Engine
+import com.kakao.actionbase.engine.Engine
 
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration

From 40417e5611a31999cc07cbff26b3430cdd282386 Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Mon, 13 Apr 2026 18:58:19 +0900
Subject: [PATCH 06/11] Rename CatalogLoader to Catalog

The interface IS the catalog (the registry of databases / tables /
aliases), matching how Spark, Iceberg, Trino, and Glue use the term.
`bind` is part of the catalog's lifecycle, the same way Spark's
ExternalCatalog has `init`/`close` next to its query methods.

- CatalogLoader -> Catalog
- PeriodicCatalogLoader -> PeriodicCatalog
- log strings and field names follow the rename.
---
 .../com/kakao/actionbase/engine/Engine.kt     | 16 +++---
 .../catalog/{CatalogLoader.kt => Catalog.kt}  |  2 +-
 ...dicCatalogLoader.kt => PeriodicCatalog.kt} | 10 ++--
 .../com/kakao/actionbase/engine/EngineTest.kt |  4 +-
 ...ogLoaderTest.kt => PeriodicCatalogTest.kt} | 56 +++++++++----------
 5 files changed, 44 insertions(+), 44 deletions(-)
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/{CatalogLoader.kt => Catalog.kt} (73%)
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/{PeriodicCatalogLoader.kt => PeriodicCatalog.kt} (93%)
 rename engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/{PeriodicCatalogLoaderTest.kt => PeriodicCatalogTest.kt} (64%)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/Engine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/Engine.kt
index 8e89d66b..ec78fdba 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/Engine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/Engine.kt
@@ -1,19 +1,19 @@
 package com.kakao.actionbase.engine
 
-import com.kakao.actionbase.engine.catalog.CatalogLoader
-import com.kakao.actionbase.engine.catalog.PeriodicCatalogLoader
+import com.kakao.actionbase.engine.catalog.Catalog
+import com.kakao.actionbase.engine.catalog.PeriodicCatalog
 
 import java.time.Duration
 
 class Engine(
-    private val catalogLoader: CatalogLoader,
+    private val catalog: Catalog,
 ) : AutoCloseable {
     init {
-        catalogLoader.bind(this)
+        catalog.bind(this)
     }
 
     override fun close() {
-        catalogLoader.close()
+        catalog.close()
     }
 
     companion object {
@@ -21,12 +21,12 @@ class Engine(
             catalogReloadInitialDelay: Duration = Duration.ZERO,
             catalogReloadInterval: Duration? = null,
         ): Engine {
-            val catalogLoader =
-                PeriodicCatalogLoader(
+            val catalog =
+                PeriodicCatalog(
                     catalogReloadInitialDelay,
                     catalogReloadInterval,
                 )
-            return Engine(catalogLoader)
+            return Engine(catalog)
         }
     }
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/CatalogLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
similarity index 73%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/CatalogLoader.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
index f47e6a04..53e8d84a 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/CatalogLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
@@ -2,6 +2,6 @@ package com.kakao.actionbase.engine.catalog
 
 import com.kakao.actionbase.engine.Engine
 
-interface CatalogLoader : AutoCloseable {
+interface Catalog : AutoCloseable {
     fun bind(engine: Engine)
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoader.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
similarity index 93%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoader.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
index e291d823..01526664 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoader.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
@@ -12,10 +12,10 @@ import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 
-class PeriodicCatalogLoader(
+class PeriodicCatalog(
     private val catalogReloadInitialDelay: Duration,
     private val catalogReloadInterval: Duration?,
-) : CatalogLoader {
+) : Catalog {
     @Volatile private var reloadCount: Long = 0
 
     @Volatile private var lastReloadAt: Instant? = null
@@ -26,7 +26,7 @@ class PeriodicCatalogLoader(
     @Synchronized
     override fun bind(engine: Engine) {
         if (bound) {
-            log.warn("PeriodicCatalogLoader already bound")
+            log.warn("PeriodicCatalog already bound")
             return
         }
         this.engine = engine
@@ -70,7 +70,7 @@ class PeriodicCatalogLoader(
     @Synchronized
     override fun close() {
         if (!bound) return
-        log.info("PeriodicCatalogLoader closing after {} reloads", reloadCount)
+        log.info("PeriodicCatalog closing after {} reloads", reloadCount)
         disposable?.dispose()
         disposable = null
         engine = null
@@ -91,6 +91,6 @@ class PeriodicCatalogLoader(
     }
 
     companion object {
-        private val log = LoggerFactory.getLogger(PeriodicCatalogLoader::class.java)
+        private val log = LoggerFactory.getLogger(PeriodicCatalog::class.java)
     }
 }
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
index 7b25b9ad..63a62f9d 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
@@ -1,6 +1,6 @@
 package com.kakao.actionbase.engine
 
-import com.kakao.actionbase.engine.catalog.CatalogLoader
+import com.kakao.actionbase.engine.catalog.Catalog
 
 import kotlin.test.assertEquals
 import kotlin.test.assertSame
@@ -36,7 +36,7 @@ class EngineTest {
         Engine.create().close()
     }
 
-    private class FakeLoader : CatalogLoader {
+    private class FakeLoader : Catalog {
         var bound: Engine? = null
         var bindCount = 0
         var closeCount = 0
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoaderTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogTest.kt
similarity index 64%
rename from engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoaderTest.kt
rename to engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogTest.kt
index fbf2b405..9a198087 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogLoaderTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalogTest.kt
@@ -10,82 +10,82 @@ import kotlin.test.assertTrue
 
 import org.junit.jupiter.api.Test
 
-class PeriodicCatalogLoaderTest {
+class PeriodicCatalogTest {
     @Test
     fun `periodic loop fires on schedule`() {
-        val loader =
-            PeriodicCatalogLoader(
+        val catalog =
+            PeriodicCatalog(
                 catalogReloadInitialDelay = Duration.ZERO,
                 catalogReloadInterval = Duration.ofMillis(20),
             )
-        Engine(loader).use {
-            waitUntil(Duration.ofSeconds(2)) { loader.reloadCount() >= 3 }
-            assertTrue(loader.reloadCount() >= 3)
-            assertNotNull(loader.lastReloadAt())
+        Engine(catalog).use {
+            waitUntil(Duration.ofSeconds(2)) { catalog.reloadCount() >= 3 }
+            assertTrue(catalog.reloadCount() >= 3)
+            assertNotNull(catalog.lastReloadAt())
         }
     }
 
     @Test
     fun `close halts the loop`() {
-        val loader =
-            PeriodicCatalogLoader(
+        val catalog =
+            PeriodicCatalog(
                 catalogReloadInitialDelay = Duration.ZERO,
                 catalogReloadInterval = Duration.ofMillis(20),
             )
-        val engine = Engine(loader)
-        waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() >= 1 }
+        val engine = Engine(catalog)
+        waitUntil(Duration.ofSeconds(1)) { catalog.reloadCount() >= 1 }
         engine.close()
         Thread.sleep(50)
-        val snapshot = loader.reloadCount()
+        val snapshot = catalog.reloadCount()
         Thread.sleep(100)
-        assertTrue(loader.reloadCount() == snapshot, "reloadCount should not advance after close")
+        assertTrue(catalog.reloadCount() == snapshot, "reloadCount should not advance after close")
     }
 
     @Test
     fun `bind triggers exactly one reload before the periodic loop kicks in`() {
         // A long interval lets us observe the initial reload alone.
-        val loader =
-            PeriodicCatalogLoader(
+        val catalog =
+            PeriodicCatalog(
                 catalogReloadInitialDelay = Duration.ZERO,
                 catalogReloadInterval = Duration.ofMinutes(10),
             )
-        Engine(loader).use {
-            waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
+        Engine(catalog).use {
+            waitUntil(Duration.ofSeconds(1)) { catalog.reloadCount() == 1L }
             // Window between the initial reload and the next periodic tick
             // should stay at exactly 1.
             Thread.sleep(50)
-            assertEquals(1, loader.reloadCount())
+            assertEquals(1, catalog.reloadCount())
         }
     }
 
     @Test
     fun `null interval still runs the initial reload`() {
-        val loader =
-            PeriodicCatalogLoader(
+        val catalog =
+            PeriodicCatalog(
                 catalogReloadInitialDelay = Duration.ZERO,
                 catalogReloadInterval = null,
             )
-        Engine(loader).use {
-            waitUntil(Duration.ofSeconds(1)) { loader.reloadCount() == 1L }
+        Engine(catalog).use {
+            waitUntil(Duration.ofSeconds(1)) { catalog.reloadCount() == 1L }
             // No periodic loop, so the count must stay at 1 forever.
             Thread.sleep(50)
-            assertEquals(1, loader.reloadCount())
+            assertEquals(1, catalog.reloadCount())
         }
     }
 
     @Test
     fun `initial delay defers the first reload`() {
-        val loader =
-            PeriodicCatalogLoader(
+        val catalog =
+            PeriodicCatalog(
                 catalogReloadInitialDelay = Duration.ofMillis(300),
                 catalogReloadInterval = null,
             )
-        Engine(loader).use {
+        Engine(catalog).use {
             // Before the delay expires, the reload has not fired yet.
             Thread.sleep(100)
-            assertEquals(0, loader.reloadCount())
+            assertEquals(0, catalog.reloadCount())
             // After the delay, the one-shot reload arrives.
-            waitUntil(Duration.ofSeconds(2)) { loader.reloadCount() == 1L }
+            waitUntil(Duration.ofSeconds(2)) { catalog.reloadCount() == 1L }
         }
     }
 

From c98b632d4c6f21aed7cd196e4c33fecf06db168f Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Tue, 14 Apr 2026 10:57:52 +0900
Subject: [PATCH 07/11] Expose databases / tables / aliases on Catalog

- Catalog interface exposes three maps directly (DatabaseDescriptor,
  TableDescriptor, AliasDescriptor) so callers can write catalog.tables[id]
  instead of catalog.snapshot.tables[id].
- PeriodicCatalog keeps a private nested Snapshot data class holding
  all three maps in one `@Volatile` reference, so reload swaps them
  atomically (a single reference write). Each getter delegates to the
  current snapshot, giving consistent per-call reads without putting
  the snapshot type on the public API.
- EngineTest's FakeLoader implements the three maps directly (empty).
---
 .../actionbase/engine/catalog/Catalog.kt      | 11 +++++++++
 .../engine/catalog/PeriodicCatalog.kt         | 24 ++++++++++++++++++-
 .../com/kakao/actionbase/engine/EngineTest.kt |  9 +++++++
 3 files changed, 43 insertions(+), 1 deletion(-)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
index 53e8d84a..f1460afe 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
@@ -1,7 +1,18 @@
 package com.kakao.actionbase.engine.catalog
 
+import com.kakao.actionbase.core.metadata.AliasDescriptor
+import com.kakao.actionbase.core.metadata.DatabaseDescriptor
+import com.kakao.actionbase.core.metadata.DatabaseId
+import com.kakao.actionbase.core.metadata.TableDescriptor
+import com.kakao.actionbase.core.metadata.TableId
 import com.kakao.actionbase.engine.Engine
 
 interface Catalog : AutoCloseable {
     fun bind(engine: Engine)
+
+    val databases: Map<DatabaseId, DatabaseDescriptor>
+
+    val tables: Map<TableId, TableDescriptor<*>>
+
+    val aliases: Map<TableId, AliasDescriptor>
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
index 01526664..60570908 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
@@ -1,5 +1,10 @@
 package com.kakao.actionbase.engine.catalog
 
+import com.kakao.actionbase.core.metadata.AliasDescriptor
+import com.kakao.actionbase.core.metadata.DatabaseDescriptor
+import com.kakao.actionbase.core.metadata.DatabaseId
+import com.kakao.actionbase.core.metadata.TableDescriptor
+import com.kakao.actionbase.core.metadata.TableId
 import com.kakao.actionbase.engine.Engine
 
 import java.time.Duration
@@ -16,6 +21,12 @@ class PeriodicCatalog(
     private val catalogReloadInitialDelay: Duration,
     private val catalogReloadInterval: Duration?,
 ) : Catalog {
+    @Volatile private var snapshot: Snapshot = Snapshot.EMPTY
+
+    override val databases: Map<DatabaseId, DatabaseDescriptor> get() = snapshot.databases
+    override val tables: Map<TableId, TableDescriptor<*>> get() = snapshot.tables
+    override val aliases: Map<TableId, AliasDescriptor> get() = snapshot.aliases
+
     @Volatile private var reloadCount: Long = 0
 
     @Volatile private var lastReloadAt: Instant? = null
@@ -83,13 +94,24 @@ class PeriodicCatalog(
     private fun reload() {
         // Guards against an in-flight tick that fires after `close()`
         // nulled out `engine`. Phase 2 will read the catalog through
-        // `engine` here, which makes this null check load-bearing.
+        // `engine` here and swap `snapshot` atomically.
         if (engine == null) return
         log.debug("reloading catalog")
+        // Phase 2: snapshot = Snapshot(loadedDbs, loadedTables, loadedAliases)
         reloadCount++
         lastReloadAt = Instant.now()
     }
 
+    private data class Snapshot(
+        val databases: Map<DatabaseId, DatabaseDescriptor>,
+        val tables: Map<TableId, TableDescriptor<*>>,
+        val aliases: Map<TableId, AliasDescriptor>,
+    ) {
+        companion object {
+            val EMPTY = Snapshot(emptyMap(), emptyMap(), emptyMap())
+        }
+    }
+
     companion object {
         private val log = LoggerFactory.getLogger(PeriodicCatalog::class.java)
     }
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
index 63a62f9d..95119467 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
@@ -1,5 +1,10 @@
 package com.kakao.actionbase.engine
 
+import com.kakao.actionbase.core.metadata.AliasDescriptor
+import com.kakao.actionbase.core.metadata.DatabaseDescriptor
+import com.kakao.actionbase.core.metadata.DatabaseId
+import com.kakao.actionbase.core.metadata.TableDescriptor
+import com.kakao.actionbase.core.metadata.TableId
 import com.kakao.actionbase.engine.catalog.Catalog
 
 import kotlin.test.assertEquals
@@ -41,6 +46,10 @@ class EngineTest {
         var bindCount = 0
         var closeCount = 0
 
+        override val databases: Map<DatabaseId, DatabaseDescriptor> = emptyMap()
+        override val tables: Map<TableId, TableDescriptor<*>> = emptyMap()
+        override val aliases: Map<TableId, AliasDescriptor> = emptyMap()
+
         override fun bind(engine: Engine) {
             bound = engine
             bindCount++

From 917dd88d3f348f04e7c8ef096ce678d134f4957e Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Tue, 14 Apr 2026 11:00:11 +0900
Subject: [PATCH 08/11] Polish Catalog review nits
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Catalog: restore KDoc covering the bind/close contract, single-
  threaded lifecycle assumption, and the reader contract for databases/
  tables/aliases (immutable or snapshot-at-read; single-getter
  consistency guaranteed, cross-map best-effort).
- PeriodicCatalog: mark `engine` as @Volatile so the null-check in
  reload() has a happens-before with close()'s write — the code now
  actually matches what the guard comment claims.
- PeriodicCatalog: regroup fields into state / public views so readers
  see a single ordering.
---
 .../actionbase/engine/catalog/Catalog.kt      | 25 +++++++++++++++++++
 .../engine/catalog/PeriodicCatalog.kt         | 18 +++++++------
 2 files changed, 36 insertions(+), 7 deletions(-)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
index f1460afe..12ccfe35 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
@@ -7,6 +7,31 @@ import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.TableId
 import com.kakao.actionbase.engine.Engine
 
+/**
+ * The engine's registry of databases, tables, and aliases — the thing
+ * Spark, Iceberg, Trino, and Glue all call a "catalog".
+ *
+ * A `Catalog` is bound to its [Engine] at construction time so it can
+ * read metadata that is itself stored as Actionbase data — the same
+ * self-hosted pattern as the v2 `Graph`. [Engine] depends on this
+ * interface only and never on a concrete implementation.
+ *
+ * ## Lifecycle
+ * - [bind] is called exactly once, from [Engine]'s constructor. The
+ *   engine is not yet fully constructed at that point, so implementations
+ *   must not call back into it synchronously.
+ * - [close] is called on shutdown. Implementations must be idempotent —
+ *   `close` may be called multiple times and must not throw.
+ * - Both methods may assume single-threaded invocation by the engine.
+ *
+ * ## Reading
+ * - [databases], [tables], and [aliases] may be read from any thread at
+ *   any time after construction. Implementations must return maps that
+ *   do not change from the caller's perspective — either immutable maps
+ *   or snapshot-at-read views. A single getter call returns a consistent
+ *   map; cross-map consistency (reading two getters back to back) is not
+ *   guaranteed across reloads but is effectively atomic in practice.
+ */
 interface Catalog : AutoCloseable {
     fun bind(engine: Engine)
 
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
index 60570908..8d7e35df 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
@@ -21,19 +21,22 @@ class PeriodicCatalog(
     private val catalogReloadInitialDelay: Duration,
     private val catalogReloadInterval: Duration?,
 ) : Catalog {
+    // --- state ---
     @Volatile private var snapshot: Snapshot = Snapshot.EMPTY
 
-    override val databases: Map<DatabaseId, DatabaseDescriptor> get() = snapshot.databases
-    override val tables: Map<TableId, TableDescriptor<*>> get() = snapshot.tables
-    override val aliases: Map<TableId, AliasDescriptor> get() = snapshot.aliases
-
     @Volatile private var reloadCount: Long = 0
 
     @Volatile private var lastReloadAt: Instant? = null
-    private var engine: Engine? = null
+
+    @Volatile private var engine: Engine? = null
     private var bound = false
     private var disposable: Disposable? = null
 
+    // --- public views ---
+    override val databases: Map<DatabaseId, DatabaseDescriptor> get() = snapshot.databases
+    override val tables: Map<TableId, TableDescriptor<*>> get() = snapshot.tables
+    override val aliases: Map<TableId, AliasDescriptor> get() = snapshot.aliases
+
     @Synchronized
     override fun bind(engine: Engine) {
         if (bound) {
@@ -93,8 +96,9 @@ class PeriodicCatalog(
 
     private fun reload() {
         // Guards against an in-flight tick that fires after `close()`
-        // nulled out `engine`. Phase 2 will read the catalog through
-        // `engine` here and swap `snapshot` atomically.
+        // nulled out `engine`. `engine` is @Volatile so this read has a
+        // happens-before with close()'s write. Phase 2 will read the
+        // catalog through `engine` here and swap `snapshot` atomically.
         if (engine == null) return
         log.debug("reloading catalog")
         // Phase 2: snapshot = Snapshot(loadedDbs, loadedTables, loadedAliases)

From 48f7fa1e37306bbebb235f618bb80dff14b783bc Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Tue, 14 Apr 2026 13:49:56 +0900
Subject: [PATCH 09/11] Trim Catalog KDoc and space field groups

---
 .../actionbase/engine/catalog/Catalog.kt      | 25 -------------------
 .../engine/catalog/PeriodicCatalog.kt         |  2 ++
 2 files changed, 2 insertions(+), 25 deletions(-)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
index 12ccfe35..f1460afe 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
@@ -7,31 +7,6 @@ import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.TableId
 import com.kakao.actionbase.engine.Engine
 
-/**
- * The engine's registry of databases, tables, and aliases — the thing
- * Spark, Iceberg, Trino, and Glue all call a "catalog".
- *
- * A `Catalog` is bound to its [Engine] at construction time so it can
- * read metadata that is itself stored as Actionbase data — the same
- * self-hosted pattern as the v2 `Graph`. [Engine] depends on this
- * interface only and never on a concrete implementation.
- *
- * ## Lifecycle
- * - [bind] is called exactly once, from [Engine]'s constructor. The
- *   engine is not yet fully constructed at that point, so implementations
- *   must not call back into it synchronously.
- * - [close] is called on shutdown. Implementations must be idempotent —
- *   `close` may be called multiple times and must not throw.
- * - Both methods may assume single-threaded invocation by the engine.
- *
- * ## Reading
- * - [databases], [tables], and [aliases] may be read from any thread at
- *   any time after construction. Implementations must return maps that
- *   do not change from the caller's perspective — either immutable maps
- *   or snapshot-at-read views. A single getter call returns a consistent
- *   map; cross-map consistency (reading two getters back to back) is not
- *   guaranteed across reloads but is effectively atomic in practice.
- */
 interface Catalog : AutoCloseable {
     fun bind(engine: Engine)
 
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
index 8d7e35df..08778a2e 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
@@ -29,7 +29,9 @@ class PeriodicCatalog(
     @Volatile private var lastReloadAt: Instant? = null
 
     @Volatile private var engine: Engine? = null
+
     private var bound = false
+
     private var disposable: Disposable? = null
 
     // --- public views ---

From 96948df343e707bc029b42cbfcb427c197021bbe Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Tue, 14 Apr 2026 14:04:20 +0900
Subject: [PATCH 10/11] Rename TableBinding to Table (Iceberg-style)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The runtime object for a table was called TableBinding, which is an
unusual term. The industry-standard pair for what we have is
TableDescriptor (declarative spec) and Table (live runtime, cached)
— see Spark, Iceberg, Trino, Glue.

Renames (interface + concrete + test + callers):
- engine.binding.TableBinding -> engine.binding.Table
- v2.engine.v3.V2BackedTableBinding -> V2BackedTable
- v2.engine.v3.NilTableBinding -> NilTable
- V2BackedTableBindingTest -> V2BackedTableTest
- QueryEngine.getTableBinding -> getTable
- MutationEngine.getTableBinding -> getTable
- HBaseIndexedLabel.tableBinding -> table

The package `engine.binding` is kept as-is for now — moving it to
`engine.table` or `engine.catalog` is a separate cleanup. Prepares
for the Catalog to cache Table instances in phase 3.
---
 .../kakao/actionbase/engine/MutationEngine.kt    |  8 ++++----
 .../com/kakao/actionbase/engine/QueryEngine.kt   |  6 +++---
 .../engine/binding/{TableBinding.kt => Table.kt} |  2 +-
 .../actionbase/engine/service/MutationService.kt |  6 +++---
 .../actionbase/engine/service/QueryService.kt    | 16 ++++++++--------
 .../v2/engine/label/hbase/HBaseIndexedLabel.kt   | 10 +++++-----
 .../v3/{NilTableBinding.kt => NilTable.kt}       | 12 ++++++------
 .../actionbase/v2/engine/v3/V2BackedEngine.kt    | 10 +++++-----
 ...{V2BackedTableBinding.kt => V2BackedTable.kt} |  8 ++++----
 ...dTableBindingTest.kt => V2BackedTableTest.kt} |  8 ++++----
 10 files changed, 43 insertions(+), 43 deletions(-)
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/binding/{TableBinding.kt => Table.kt} (98%)
 rename engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/{NilTableBinding.kt => NilTable.kt} (87%)
 rename engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/{V2BackedTableBinding.kt => V2BackedTable.kt} (99%)
 rename engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/{V2BackedTableBindingTest.kt => V2BackedTableTest.kt} (98%)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt
index f6635d2c..f626be29 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt
@@ -2,7 +2,7 @@ package com.kakao.actionbase.engine
 
 import com.kakao.actionbase.core.edge.MutationEvent
 import com.kakao.actionbase.core.state.State
-import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.binding.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 
 import reactor.core.publisher.Mono
@@ -13,12 +13,12 @@ import reactor.core.publisher.Mono
  */
 interface MutationEngine {
     /**
-     * Resolves a table binding from a database/alias pair.
+     * Resolves a table from a database/alias pair.
      */
-    fun getTableBinding(
+    fun getTable(
         database: String,
         alias: String,
-    ): TableBinding
+    ): Table
 
     fun writeWal(
         ctx: MutationContext,
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt
index c1946101..43cd8f34 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt
@@ -1,6 +1,6 @@
 package com.kakao.actionbase.engine
 
-import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.binding.Table
 import com.kakao.actionbase.engine.query.ActionbaseQuery
 import com.kakao.actionbase.v2.engine.sql.DataFrame
 
@@ -11,10 +11,10 @@ import reactor.core.publisher.Mono
  * Decouples QueryService from Graph by exposing only the operations it needs.
  */
 interface QueryEngine {
-    fun getTableBinding(
+    fun getTable(
         database: String,
         alias: String,
-    ): TableBinding
+    ): Table
 
     fun query(request: ActionbaseQuery): Mono<Map<String, DataFrame>>
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/binding/TableBinding.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/binding/Table.kt
similarity index 98%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/binding/TableBinding.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/binding/Table.kt
index 2da302e5..5c6f9fa4 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/binding/TableBinding.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/binding/Table.kt
@@ -11,7 +11,7 @@ import com.kakao.actionbase.v2.core.metadata.Direction
 
 import reactor.core.publisher.Mono
 
-interface TableBinding {
+interface Table {
     val table: String
     val schema: ModelSchema
     val mutationMode: MutationMode
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt
index e0b69720..a03b898f 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt
@@ -8,7 +8,7 @@ import com.kakao.actionbase.core.state.transit
 import com.kakao.actionbase.engine.Audit
 import com.kakao.actionbase.engine.MutationContext
 import com.kakao.actionbase.engine.MutationEngine
-import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.binding.Table
 import com.kakao.actionbase.engine.context.RequestContext
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.engine.metadata.MutationModeContext
@@ -35,7 +35,7 @@ class MutationService(
     ): Mono<List<MutationResult>> =
         Mono
             .fromCallable {
-                val tb = engine.getTableBinding(database, alias)
+                val tb = engine.getTable(database, alias)
                 val ctx =
                     MutationContext(
                         database = database,
@@ -73,7 +73,7 @@ class MutationService(
             }
 
     private fun readModifyWrite(
-        tb: TableBinding,
+        tb: Table,
         key: MutationKey,
         sorted: List<MutationEvent>,
         acquireLock: Boolean,
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt
index d46692b0..d4aa2747 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/QueryService.kt
@@ -48,7 +48,7 @@ class QueryService(
         require(filters == null) { "`filters` is not yet supported in count query." }
         require(features.isEmpty()) { "`features` ${features.joinToString(", ")} are not supported in get query." }
 
-        return engine.getTableBinding(database, table).count(start.toSet(), direction)
+        return engine.getTable(database, table).count(start.toSet(), direction)
     }
 
     @Suppress("UnusedParameter")
@@ -69,7 +69,7 @@ class QueryService(
                 target.distinct().map { t -> s to t }
             }
 
-        return engine.getTableBinding(database, table).gets(keys, filters)
+        return engine.getTable(database, table).gets(keys, filters)
     }
 
     @Suppress("UnusedParameter")
@@ -82,14 +82,14 @@ class QueryService(
     ): Mono<DataFrameEdgePayload> {
         require(features.isEmpty()) { "`features` ${features.joinToString(", ")} are not supported in get query." }
 
-        val tb = engine.getTableBinding(database, table)
+        val t = engine.getTable(database, table)
 
-        require(tb.schema is com.kakao.actionbase.core.metadata.common.ModelSchema.MultiEdge) {
+        require(t.schema is com.kakao.actionbase.core.metadata.common.ModelSchema.MultiEdge) {
             "get query with ids is only supported for multi-edge tables."
         }
 
         val keys = ids.distinct().map { id -> id to id }
-        return tb.gets(keys, filters)
+        return t.gets(keys, filters)
     }
 
     fun scan(
@@ -103,7 +103,7 @@ class QueryService(
         ranges: String? = null,
         filters: String? = null,
         features: List<String> = emptyList(),
-    ): Mono<DataFrameEdgePayload> = engine.getTableBinding(database, table).scan(index, start, direction, limit, offset, ranges, filters, features)
+    ): Mono<DataFrameEdgePayload> = engine.getTable(database, table).scan(index, start, direction, limit, offset, ranges, filters, features)
 
     fun seek(
         database: String,
@@ -113,7 +113,7 @@ class QueryService(
         direction: Direction,
         limit: Int = ScanFilter.defaultLimit,
         offset: String? = null,
-    ): Mono<DataFrameEdgePayload> = engine.getTableBinding(database, table).seek(cache, start, direction, limit, offset)
+    ): Mono<DataFrameEdgePayload> = engine.getTable(database, table).seek(cache, start, direction, limit, offset)
 
     fun agg(
         database: String,
@@ -125,7 +125,7 @@ class QueryService(
         filters: String? = null,
         features: List<String> = emptyList(),
         ttl: Long? = null,
-    ): Mono<DataFrameEdgeAggPayload> = engine.getTableBinding(database, table).agg(group, start, direction, ranges, filters, features, ttl)
+    ): Mono<DataFrameEdgeAggPayload> = engine.getTable(database, table).agg(group, start, direction, ranges, filters, features, ttl)
 
     fun query(request: ActionbaseQuery): Mono<Map<String, DataFrame>> = engine.query(request)
 
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt
index a45d1564..f0733200 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt
@@ -3,7 +3,7 @@ package com.kakao.actionbase.v2.engine.label.hbase
 import com.kakao.actionbase.core.edge.Edge
 import com.kakao.actionbase.core.edge.mapper.EdgeRecordMapper
 import com.kakao.actionbase.core.edge.record.EdgeCacheRecord
-import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.binding.Table
 import com.kakao.actionbase.v2.core.code.EdgeEncoder
 import com.kakao.actionbase.v2.core.code.IdEdgeEncoder
 import com.kakao.actionbase.v2.core.code.Index
@@ -24,8 +24,8 @@ import com.kakao.actionbase.v2.engine.sql.ScanFilter
 import com.kakao.actionbase.v2.engine.sql.StatKey
 import com.kakao.actionbase.v2.engine.storage.hbase.HBaseStorage
 import com.kakao.actionbase.v2.engine.storage.hbase.HBaseTables
-import com.kakao.actionbase.v2.engine.v3.V2BackedTableBinding
-import com.kakao.actionbase.v2.engine.v3.V2BackedTableBinding.Companion.toV3
+import com.kakao.actionbase.v2.engine.v3.V2BackedTable
+import com.kakao.actionbase.v2.engine.v3.V2BackedTable.Companion.toV3
 import com.kakao.actionbase.v2.engine.v3.V3TableDescriptor
 
 import reactor.core.publisher.Mono
@@ -47,8 +47,8 @@ open class HBaseIndexedLabel(
         tables = tables,
     ),
     IndexedLabelMixin<ByteArray> {
-    val tableBinding: TableBinding =
-        V2BackedTableBinding(
+    val table: Table =
+        V2BackedTable(
             descriptor = V3TableDescriptor.create(entity),
             label = this,
             mapper = edgeRecordMapper,
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTableBinding.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTable.kt
similarity index 87%
rename from engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTableBinding.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTable.kt
index b20e4470..862f89ff 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTableBinding.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTable.kt
@@ -7,15 +7,15 @@ import com.kakao.actionbase.core.edge.payload.DataFrameEdgePayload
 import com.kakao.actionbase.core.metadata.common.ModelSchema
 import com.kakao.actionbase.core.state.State
 import com.kakao.actionbase.engine.binding.MutationRecordsSummary
-import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.binding.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.v2.core.metadata.Direction
 
 import reactor.core.publisher.Mono
 
-class NilTableBinding(
+class NilTable(
     descriptor: V3TableDescriptor,
-) : TableBinding {
+) : Table {
     override val table: String = descriptor.table
     override val schema: ModelSchema = descriptor.schema
     override val mutationMode: MutationMode = MutationMode.SYNC
@@ -43,7 +43,7 @@ class NilTableBinding(
     override fun gets(
         keys: List<Pair<Any, Any>>,
         filters: String?,
-    ): Mono<DataFrameEdgePayload> = V2BackedTableBinding.EMPTY_EDGE_PAYLOAD
+    ): Mono<DataFrameEdgePayload> = V2BackedTable.EMPTY_EDGE_PAYLOAD
 
     override fun scan(
         index: String,
@@ -54,7 +54,7 @@ class NilTableBinding(
         ranges: String?,
         filters: String?,
         features: List<String>,
-    ): Mono<DataFrameEdgePayload> = V2BackedTableBinding.EMPTY_EDGE_PAYLOAD
+    ): Mono<DataFrameEdgePayload> = V2BackedTable.EMPTY_EDGE_PAYLOAD
 
     override fun seek(
         cache: String,
@@ -62,7 +62,7 @@ class NilTableBinding(
         direction: Direction,
         limit: Int,
         offset: String?,
-    ): Mono<DataFrameEdgePayload> = V2BackedTableBinding.EMPTY_EDGE_PAYLOAD
+    ): Mono<DataFrameEdgePayload> = V2BackedTable.EMPTY_EDGE_PAYLOAD
 
     override fun agg(
         group: String,
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt
index 4b4c2740..1ae8419e 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt
@@ -7,7 +7,7 @@ import com.kakao.actionbase.core.state.State
 import com.kakao.actionbase.engine.MutationContext
 import com.kakao.actionbase.engine.MutationEngine
 import com.kakao.actionbase.engine.QueryEngine
-import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.binding.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.engine.query.ActionbaseQuery
 import com.kakao.actionbase.v2.engine.Graph
@@ -26,20 +26,20 @@ class V2BackedEngine(
     private val graph: Graph,
 ) : MutationEngine,
     QueryEngine {
-    override fun getTableBinding(
+    override fun getTable(
         database: String,
         alias: String,
-    ): TableBinding {
+    ): Table {
         val label = graph.getLabel(EntityName(database, alias))
         if (label is NilLabel) {
-            return NilTableBinding(V3TableDescriptor.create(label.entity))
+            return NilTable(V3TableDescriptor.create(label.entity))
         }
         if (label !is HBaseIndexedLabel) {
             throw UnsupportedOperationException(
                 "This Label (${label.entity.fullName}, ${label.javaClass}) is not indexed or not supported for edge mutation",
             )
         }
-        return label.tableBinding
+        return label.table
     }
 
     override fun query(request: ActionbaseQuery): Mono<Map<String, DataFrame>> = graph.query(request)
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableBinding.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTable.kt
similarity index 99%
rename from engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableBinding.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTable.kt
index 91c11178..f320d208 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableBinding.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTable.kt
@@ -24,7 +24,7 @@ import com.kakao.actionbase.core.state.SpecialStateValue
 import com.kakao.actionbase.core.state.State
 import com.kakao.actionbase.core.storage.HBaseRecord
 import com.kakao.actionbase.engine.binding.MutationRecordsSummary
-import com.kakao.actionbase.engine.binding.TableBinding
+import com.kakao.actionbase.engine.binding.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.v2.core.code.CryptoUtils
 import com.kakao.actionbase.v2.core.edge.Edge
@@ -46,12 +46,12 @@ import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 
-class V2BackedTableBinding(
+class V2BackedTable(
     private val descriptor: V3TableDescriptor,
     private val label: HBaseIndexedLabel,
     private val mapper: EdgeRecordMapper,
     private val lockTimeout: Long,
-) : TableBinding {
+) : Table {
     override val table: String = descriptor.table
     override val schema: ModelSchema = descriptor.schema
     override val mutationMode: MutationMode = MutationMode.valueOf(label.entity.mode.name)
@@ -609,7 +609,7 @@ class V2BackedTableBinding(
     }
 
     companion object {
-        private val log = LoggerFactory.getLogger(V2BackedTableBinding::class.java)
+        private val log = LoggerFactory.getLogger(V2BackedTable::class.java)
 
         private const val SELECT_COUNT_FIELD = "COUNT(1)"
         private const val TS_FIELD = "ts"
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableBindingTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableTest.kt
similarity index 98%
rename from engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableBindingTest.kt
rename to engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableTest.kt
index 50112d76..78476d72 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableBindingTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTableTest.kt
@@ -9,8 +9,8 @@ import com.kakao.actionbase.core.state.SpecialStateValue
 import com.kakao.actionbase.core.state.State
 import com.kakao.actionbase.core.state.StateValue
 import com.kakao.actionbase.v2.core.code.hbase.Constants
-import com.kakao.actionbase.v2.engine.v3.V2BackedTableBinding.Companion.mergeQualifiers
-import com.kakao.actionbase.v2.engine.v3.V2BackedTableBinding.Companion.specialStateValueToNull
+import com.kakao.actionbase.v2.engine.v3.V2BackedTable.Companion.mergeQualifiers
+import com.kakao.actionbase.v2.engine.v3.V2BackedTable.Companion.specialStateValueToNull
 
 import kotlin.test.assertEquals
 import kotlin.test.assertNull
@@ -21,7 +21,7 @@ import org.apache.hadoop.hbase.client.Put
 import org.junit.jupiter.api.Nested
 import org.junit.jupiter.api.Test
 
-class V2BackedTableBindingTest {
+class V2BackedTableTest {
     @Nested
     inner class MergeQualifiers {
         private fun groupRecord(
@@ -99,7 +99,7 @@ class V2BackedTableBindingTest {
 
     /**
      * Verifies that EdgeCacheRecord encodes into HBase Put/Delete correctly,
-     * matching the pattern used in V2BackedTableBinding.buildHBaseMutations().
+     * matching the pattern used in V2BackedTable.buildHBaseMutations().
      */
     @Nested
     inner class CacheHBaseMutations {

From fbc29bd90c4047a3a54d73e739579433b068b449 Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Tue, 14 Apr 2026 14:18:21 +0900
Subject: [PATCH 11/11] Expose Database / Table / Alias runtime types on
 Catalog
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The catalog snapshot now holds fully-resolved runtime objects instead
of raw descriptors, matching Iceberg's Table / Catalog model.

- Add engine.catalog.Database  (thin wrapper over DatabaseDescriptor).
- Add engine.catalog.Alias     (AliasDescriptor + resolved Table).
- Move engine.binding.Table -> engine.catalog.Table and add
  `val descriptor: TableDescriptor<*>` so Table carries both its spec
  and its runtime ops — the core of the user's 'Table = Descriptor +
  runtime' model.
- V3TableDescriptor gains `toTableDescriptor(entity)` so the v2
  adapters can build the full core.metadata.TableDescriptor from a
  LabelEntity without pulling in V3MetadataConverter.
- V2BackedTable and NilTable implement the new `descriptor` field
  using the V3TableDescriptor conversion. NilTable now takes the
  LabelEntity too so it can build its descriptor.
- Catalog interface exposes Database / Table / Alias maps; Snapshot in
  PeriodicCatalog swaps runtime types atomically via a single reference.

Phase 2 will read the metastore through `engine` inside reload() and
build fresh Database / Table / Alias instances, reusing existing Table
instances when their descriptors haven't changed.
---
 .../kakao/actionbase/engine/MutationEngine.kt |  2 +-
 .../kakao/actionbase/engine/QueryEngine.kt    |  2 +-
 .../kakao/actionbase/engine/catalog/Alias.kt  | 10 ++++++
 .../actionbase/engine/catalog/Catalog.kt      |  9 ++---
 .../actionbase/engine/catalog/Database.kt     |  7 ++++
 .../engine/catalog/PeriodicCatalog.kt         | 23 ++++++-------
 .../engine/{binding => catalog}/Table.kt      |  4 ++-
 .../engine/service/MutationService.kt         |  2 +-
 .../engine/label/hbase/HBaseIndexedLabel.kt   |  4 +--
 .../kakao/actionbase/v2/engine/v3/NilTable.kt | 14 +++++---
 .../actionbase/v2/engine/v3/V2BackedEngine.kt |  4 +--
 .../actionbase/v2/engine/v3/V2BackedTable.kt  | 12 ++++---
 .../v2/engine/v3/V3TableDescriptor.kt         | 34 +++++++++++++++++--
 .../com/kakao/actionbase/engine/EngineTest.kt | 12 +++----
 14 files changed, 94 insertions(+), 45 deletions(-)
 create mode 100644 engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Alias.kt
 create mode 100644 engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Database.kt
 rename engine/src/main/kotlin/com/kakao/actionbase/engine/{binding => catalog}/Table.kt (93%)

diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt
index f626be29..d4ff3f1d 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/MutationEngine.kt
@@ -2,7 +2,7 @@ package com.kakao.actionbase.engine
 
 import com.kakao.actionbase.core.edge.MutationEvent
 import com.kakao.actionbase.core.state.State
-import com.kakao.actionbase.engine.binding.Table
+import com.kakao.actionbase.engine.catalog.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 
 import reactor.core.publisher.Mono
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt
index 43cd8f34..22603c84 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/QueryEngine.kt
@@ -1,6 +1,6 @@
 package com.kakao.actionbase.engine
 
-import com.kakao.actionbase.engine.binding.Table
+import com.kakao.actionbase.engine.catalog.Table
 import com.kakao.actionbase.engine.query.ActionbaseQuery
 import com.kakao.actionbase.v2.engine.sql.DataFrame
 
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Alias.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Alias.kt
new file mode 100644
index 00000000..6b314f15
--- /dev/null
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Alias.kt
@@ -0,0 +1,10 @@
+package com.kakao.actionbase.engine.catalog
+
+import com.kakao.actionbase.core.metadata.AliasDescriptor
+
+interface Alias {
+    val descriptor: AliasDescriptor
+
+    /** The table this alias resolves to, already materialized. */
+    val table: Table
+}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
index f1460afe..b1dd085f 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Catalog.kt
@@ -1,18 +1,15 @@
 package com.kakao.actionbase.engine.catalog
 
-import com.kakao.actionbase.core.metadata.AliasDescriptor
-import com.kakao.actionbase.core.metadata.DatabaseDescriptor
 import com.kakao.actionbase.core.metadata.DatabaseId
-import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.TableId
 import com.kakao.actionbase.engine.Engine
 
 interface Catalog : AutoCloseable {
     fun bind(engine: Engine)
 
-    val databases: Map<DatabaseId, DatabaseDescriptor>
+    val databases: Map<DatabaseId, Database>
 
-    val tables: Map<TableId, TableDescriptor<*>>
+    val tables: Map<TableId, Table>
 
-    val aliases: Map<TableId, AliasDescriptor>
+    val aliases: Map<TableId, Alias>
 }
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Database.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Database.kt
new file mode 100644
index 00000000..22c61352
--- /dev/null
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Database.kt
@@ -0,0 +1,7 @@
+package com.kakao.actionbase.engine.catalog
+
+import com.kakao.actionbase.core.metadata.DatabaseDescriptor
+
+interface Database {
+    val descriptor: DatabaseDescriptor
+}
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
index 08778a2e..e6fa664c 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/PeriodicCatalog.kt
@@ -1,9 +1,6 @@
 package com.kakao.actionbase.engine.catalog
 
-import com.kakao.actionbase.core.metadata.AliasDescriptor
-import com.kakao.actionbase.core.metadata.DatabaseDescriptor
 import com.kakao.actionbase.core.metadata.DatabaseId
-import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.TableId
 import com.kakao.actionbase.engine.Engine
 
@@ -35,9 +32,9 @@ class PeriodicCatalog(
     private var disposable: Disposable? = null
 
     // --- public views ---
-    override val databases: Map<DatabaseId, DatabaseDescriptor> get() = snapshot.databases
-    override val tables: Map<TableId, TableDescriptor<*>> get() = snapshot.tables
-    override val aliases: Map<TableId, AliasDescriptor> get() = snapshot.aliases
+    override val databases: Map<DatabaseId, Database> get() = snapshot.databases
+    override val tables: Map<TableId, Table> get() = snapshot.tables
+    override val aliases: Map<TableId, Alias> get() = snapshot.aliases
 
     @Synchronized
     override fun bind(engine: Engine) {
@@ -98,20 +95,20 @@ class PeriodicCatalog(
 
     private fun reload() {
         // Guards against an in-flight tick that fires after `close()`
-        // nulled out `engine`. `engine` is @Volatile so this read has a
-        // happens-before with close()'s write. Phase 2 will read the
-        // catalog through `engine` here and swap `snapshot` atomically.
+        // nulled out `engine`. Phase 2 will read the catalog through
+        // `engine` here and swap `snapshot` atomically, reusing existing
+        // Table instances when their descriptors haven't changed.
         if (engine == null) return
         log.debug("reloading catalog")
-        // Phase 2: snapshot = Snapshot(loadedDbs, loadedTables, loadedAliases)
+        // Phase 2: snapshot = Snapshot(freshDatabases, freshTables, freshAliases)
         reloadCount++
         lastReloadAt = Instant.now()
     }
 
     private data class Snapshot(
-        val databases: Map<DatabaseId, DatabaseDescriptor>,
-        val tables: Map<TableId, TableDescriptor<*>>,
-        val aliases: Map<TableId, AliasDescriptor>,
+        val databases: Map<DatabaseId, Database>,
+        val tables: Map<TableId, Table>,
+        val aliases: Map<TableId, Alias>,
     ) {
         companion object {
             val EMPTY = Snapshot(emptyMap(), emptyMap(), emptyMap())
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/binding/Table.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Table.kt
similarity index 93%
rename from engine/src/main/kotlin/com/kakao/actionbase/engine/binding/Table.kt
rename to engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Table.kt
index 5c6f9fa4..c917d745 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/binding/Table.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/catalog/Table.kt
@@ -1,9 +1,10 @@
-package com.kakao.actionbase.engine.binding
+package com.kakao.actionbase.engine.catalog
 
 import com.kakao.actionbase.core.edge.MutationKey
 import com.kakao.actionbase.core.edge.payload.DataFrameEdgeAggPayload
 import com.kakao.actionbase.core.edge.payload.DataFrameEdgeCountPayload
 import com.kakao.actionbase.core.edge.payload.DataFrameEdgePayload
+import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.common.ModelSchema
 import com.kakao.actionbase.core.state.State
 import com.kakao.actionbase.engine.metadata.MutationMode
@@ -12,6 +13,7 @@ import com.kakao.actionbase.v2.core.metadata.Direction
 import reactor.core.publisher.Mono
 
 interface Table {
+    val descriptor: TableDescriptor<*>
     val table: String
     val schema: ModelSchema
     val mutationMode: MutationMode
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt
index a03b898f..ba72123a 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/engine/service/MutationService.kt
@@ -8,7 +8,7 @@ import com.kakao.actionbase.core.state.transit
 import com.kakao.actionbase.engine.Audit
 import com.kakao.actionbase.engine.MutationContext
 import com.kakao.actionbase.engine.MutationEngine
-import com.kakao.actionbase.engine.binding.Table
+import com.kakao.actionbase.engine.catalog.Table
 import com.kakao.actionbase.engine.context.RequestContext
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.engine.metadata.MutationModeContext
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt
index f0733200..9ee0c599 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/label/hbase/HBaseIndexedLabel.kt
@@ -3,7 +3,7 @@ package com.kakao.actionbase.v2.engine.label.hbase
 import com.kakao.actionbase.core.edge.Edge
 import com.kakao.actionbase.core.edge.mapper.EdgeRecordMapper
 import com.kakao.actionbase.core.edge.record.EdgeCacheRecord
-import com.kakao.actionbase.engine.binding.Table
+import com.kakao.actionbase.engine.catalog.Table
 import com.kakao.actionbase.v2.core.code.EdgeEncoder
 import com.kakao.actionbase.v2.core.code.IdEdgeEncoder
 import com.kakao.actionbase.v2.core.code.Index
@@ -49,7 +49,7 @@ open class HBaseIndexedLabel(
     IndexedLabelMixin<ByteArray> {
     val table: Table =
         V2BackedTable(
-            descriptor = V3TableDescriptor.create(entity),
+            v3 = V3TableDescriptor.create(entity),
             label = this,
             mapper = edgeRecordMapper,
             lockTimeout = lockTimeout,
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTable.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTable.kt
index 862f89ff..579a2326 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTable.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/NilTable.kt
@@ -4,20 +4,24 @@ import com.kakao.actionbase.core.edge.MutationKey
 import com.kakao.actionbase.core.edge.payload.DataFrameEdgeAggPayload
 import com.kakao.actionbase.core.edge.payload.DataFrameEdgeCountPayload
 import com.kakao.actionbase.core.edge.payload.DataFrameEdgePayload
+import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.common.ModelSchema
 import com.kakao.actionbase.core.state.State
-import com.kakao.actionbase.engine.binding.MutationRecordsSummary
-import com.kakao.actionbase.engine.binding.Table
+import com.kakao.actionbase.engine.catalog.MutationRecordsSummary
+import com.kakao.actionbase.engine.catalog.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.v2.core.metadata.Direction
+import com.kakao.actionbase.v2.engine.entity.LabelEntity
 
 import reactor.core.publisher.Mono
 
 class NilTable(
-    descriptor: V3TableDescriptor,
+    v3: V3TableDescriptor,
+    entity: LabelEntity,
 ) : Table {
-    override val table: String = descriptor.table
-    override val schema: ModelSchema = descriptor.schema
+    override val descriptor: TableDescriptor<*> = v3.toTableDescriptor(entity)
+    override val table: String = v3.table
+    override val schema: ModelSchema = v3.schema
     override val mutationMode: MutationMode = MutationMode.SYNC
 
     override fun <T> withLock(
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt
index 1ae8419e..f3f18aa4 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedEngine.kt
@@ -7,7 +7,7 @@ import com.kakao.actionbase.core.state.State
 import com.kakao.actionbase.engine.MutationContext
 import com.kakao.actionbase.engine.MutationEngine
 import com.kakao.actionbase.engine.QueryEngine
-import com.kakao.actionbase.engine.binding.Table
+import com.kakao.actionbase.engine.catalog.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.engine.query.ActionbaseQuery
 import com.kakao.actionbase.v2.engine.Graph
@@ -32,7 +32,7 @@ class V2BackedEngine(
     ): Table {
         val label = graph.getLabel(EntityName(database, alias))
         if (label is NilLabel) {
-            return NilTable(V3TableDescriptor.create(label.entity))
+            return NilTable(V3TableDescriptor.create(label.entity), label.entity)
         }
         if (label !is HBaseIndexedLabel) {
             throw UnsupportedOperationException(
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTable.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTable.kt
index f320d208..dd0bb29e 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTable.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V2BackedTable.kt
@@ -18,13 +18,14 @@ import com.kakao.actionbase.core.edge.record.EdgeCacheRecord
 import com.kakao.actionbase.core.edge.record.EdgeGroupRecord
 import com.kakao.actionbase.core.edge.record.EdgeStateRecord
 import com.kakao.actionbase.core.java.codec.common.hbase.Order
+import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.common.Group
 import com.kakao.actionbase.core.metadata.common.ModelSchema
 import com.kakao.actionbase.core.state.SpecialStateValue
 import com.kakao.actionbase.core.state.State
 import com.kakao.actionbase.core.storage.HBaseRecord
-import com.kakao.actionbase.engine.binding.MutationRecordsSummary
-import com.kakao.actionbase.engine.binding.Table
+import com.kakao.actionbase.engine.catalog.MutationRecordsSummary
+import com.kakao.actionbase.engine.catalog.Table
 import com.kakao.actionbase.engine.metadata.MutationMode
 import com.kakao.actionbase.v2.core.code.CryptoUtils
 import com.kakao.actionbase.v2.core.edge.Edge
@@ -47,13 +48,14 @@ import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 
 class V2BackedTable(
-    private val descriptor: V3TableDescriptor,
+    v3: V3TableDescriptor,
     private val label: HBaseIndexedLabel,
     private val mapper: EdgeRecordMapper,
     private val lockTimeout: Long,
 ) : Table {
-    override val table: String = descriptor.table
-    override val schema: ModelSchema = descriptor.schema
+    override val descriptor: TableDescriptor<*> = v3.toTableDescriptor(label.entity)
+    override val table: String = v3.table
+    override val schema: ModelSchema = v3.schema
     override val mutationMode: MutationMode = MutationMode.valueOf(label.entity.mode.name)
 
     private val groupRecordMapper = mapper.group
diff --git a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V3TableDescriptor.kt b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V3TableDescriptor.kt
index 4ccce2fa..1d553707 100644
--- a/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V3TableDescriptor.kt
+++ b/engine/src/main/kotlin/com/kakao/actionbase/v2/engine/v3/V3TableDescriptor.kt
@@ -1,5 +1,8 @@
 package com.kakao.actionbase.v2.engine.v3
 
+import com.kakao.actionbase.core.metadata.common.MutationMode as V3MutationMode
+
+import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.common.Cache
 import com.kakao.actionbase.core.metadata.common.Field
 import com.kakao.actionbase.core.metadata.common.Group
@@ -15,6 +18,7 @@ import com.kakao.actionbase.v2.core.types.DataType
 import com.kakao.actionbase.v2.core.types.EdgeSchema
 import com.kakao.actionbase.v2.core.types.VertexField
 import com.kakao.actionbase.v2.core.types.VertexType
+import com.kakao.actionbase.v2.engine.entity.EntityName
 import com.kakao.actionbase.v2.engine.entity.LabelEntity
 
 sealed class V3TableDescriptor {
@@ -22,17 +26,43 @@ sealed class V3TableDescriptor {
     abstract val table: String
     abstract val schema: ModelSchema
 
+    abstract fun toTableDescriptor(entity: LabelEntity): TableDescriptor<*>
+
     data class Edge(
         override val database: String,
         override val table: String,
         override val schema: ModelSchema.Edge,
-    ) : V3TableDescriptor()
+    ) : V3TableDescriptor() {
+        override fun toTableDescriptor(entity: LabelEntity): TableDescriptor.Edge =
+            TableDescriptor.Edge(
+                tenant = EntityName.tenant,
+                database = database,
+                table = table,
+                schema = schema,
+                mode = V3MutationMode.valueOf(entity.mode.name),
+                storage = entity.storage,
+                active = entity.active,
+                comment = entity.desc,
+            )
+    }
 
     data class MultiEdge(
         override val database: String,
         override val table: String,
         override val schema: ModelSchema.MultiEdge,
-    ) : V3TableDescriptor()
+    ) : V3TableDescriptor() {
+        override fun toTableDescriptor(entity: LabelEntity): TableDescriptor.MultiEdge =
+            TableDescriptor.MultiEdge(
+                tenant = EntityName.tenant,
+                database = database,
+                table = table,
+                schema = schema,
+                mode = V3MutationMode.valueOf(entity.mode.name),
+                storage = entity.storage,
+                active = entity.active,
+                comment = entity.desc,
+            )
+    }
 
     companion object {
         fun create(entity: LabelEntity): V3TableDescriptor {
diff --git a/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
index 95119467..dcb64b25 100644
--- a/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
+++ b/engine/src/test/kotlin/com/kakao/actionbase/engine/EngineTest.kt
@@ -1,11 +1,11 @@
 package com.kakao.actionbase.engine
 
-import com.kakao.actionbase.core.metadata.AliasDescriptor
-import com.kakao.actionbase.core.metadata.DatabaseDescriptor
 import com.kakao.actionbase.core.metadata.DatabaseId
-import com.kakao.actionbase.core.metadata.TableDescriptor
 import com.kakao.actionbase.core.metadata.TableId
+import com.kakao.actionbase.engine.catalog.Alias
 import com.kakao.actionbase.engine.catalog.Catalog
+import com.kakao.actionbase.engine.catalog.Database
+import com.kakao.actionbase.engine.catalog.Table
 
 import kotlin.test.assertEquals
 import kotlin.test.assertSame
@@ -46,9 +46,9 @@ class EngineTest {
         var bindCount = 0
         var closeCount = 0
 
-        override val databases: Map<DatabaseId, DatabaseDescriptor> = emptyMap()
-        override val tables: Map<TableId, TableDescriptor<*>> = emptyMap()
-        override val aliases: Map<TableId, AliasDescriptor> = emptyMap()
+        override val databases: Map<DatabaseId, Database> = emptyMap()
+        override val tables: Map<TableId, Table> = emptyMap()
+        override val aliases: Map<TableId, Alias> = emptyMap()
 
         override fun bind(engine: Engine) {
             bound = engine