diff --git a/cli/internal/client/actionbase_client.go b/cli/internal/client/actionbase_client.go
index 02399c6b9..76c96a2ae 100644
--- a/cli/internal/client/actionbase_client.go
+++ b/cli/internal/client/actionbase_client.go
@@ -2,6 +2,7 @@ package client
 
 import (
 	"net/url"
+	"strconv"
 
 	"github.com/kakao/actionbase/internal/client/model"
 )
@@ -121,3 +122,22 @@ func (a *ActionbaseClient) PostRaw(path string, body any) (int, string) {
 func (a *ActionbaseClient) GetHost() string {
 	return a.client.baseUrl
 }
+
+// MetastoreDumpPath is served by 0.4.x and was removed on later servers, so a caller has to be
+// ready for a 404 rather than assume it is there.
+const MetastoreDumpPath = "/graph/v2/metastore/global"
+
+// DumpMetastore reads one page of the metastore, ordered by primary key.
+//
+// Ordered by id rather than k: it is the primary key, so the server's OFFSET pagination walks an
+// index it already has. Sorting by k would order by a base64 string for no benefit here.
+func (a *ActionbaseClient) DumpMetastore(page, size int) *Response[model.MetastoreDumpPage] {
+	u := &url.URL{Path: MetastoreDumpPath}
+	q := u.Query()
+	q.Set("page", strconv.Itoa(page))
+	q.Set("size", strconv.Itoa(size))
+	q.Set("sort", "id,asc")
+	u.RawQuery = q.Encode()
+
+	return GetStream[model.MetastoreDumpPage](a.client, u.String())
+}
diff --git a/cli/internal/client/http_client.go b/cli/internal/client/http_client.go
index 494de21be..0ac1da4b6 100644
--- a/cli/internal/client/http_client.go
+++ b/cli/internal/client/http_client.go
@@ -69,6 +69,52 @@ func Get[T any](c *HTTPClient, uri string) *Response[T] {
 	return call[T](c, request, nil)
 }
 
+// GetStream is Get with two differences that a paged, version-dependent endpoint needs.
+//
+// It keeps the real status code when the body does not decode. Get replaces it with -1, and an
+// unauthorized response carries no body at all, so a caller cannot otherwise tell "this server does
+// not serve that path" from "this token may not". And it decodes straight off the connection rather
+// than reading the whole body first, so a page is not held twice.
+func GetStream[T any](c *HTTPClient, uri string) *Response[T] {
+	url := fmt.Sprintf("%s%s", c.baseUrl, uri)
+	request, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		var empty T
+		return NewResponse[T](-1, &empty, fmt.Errorf("failed to create request: %w", err))
+	}
+
+	request.Header.Set("Content-Type", "application/json")
+	if c.authKey != nil {
+		request.Header.Set("Authorization", *c.authKey)
+	}
+
+	if c.context.IsDebugEnabled {
+		slog.Debug(fmt.Sprintf("→ %s %s", request.Method, request.URL.RequestURI()))
+	}
+
+	response, err := c.client.Do(request)
+	if err != nil {
+		var empty T
+		return NewResponse[T](-1, &empty, fmt.Errorf("failed to execute request: %w", err))
+	}
+	defer func(body io.ReadCloser) {
+		_, _ = io.Copy(io.Discard, body)
+		_ = body.Close()
+	}(response.Body)
+
+	if c.context.IsDebugEnabled {
+		slog.Debug(fmt.Sprintf("← %s", response.Status))
+	}
+
+	var responseBody T
+	if err := json.NewDecoder(response.Body).Decode(&responseBody); err != nil {
+		var empty T
+		return NewResponse[T](response.StatusCode, &empty, fmt.Errorf("failed to read response Body: %w", err))
+	}
+
+	return NewResponse(response.StatusCode, &responseBody, nil)
+}
+
 func Post[T any, R any](c *HTTPClient, uri string, requestBody T) *Response[R] {
 	url := fmt.Sprintf("%s%s", c.baseUrl, uri)
 	requestBodyJson, err := json.Marshal(requestBody)
diff --git a/cli/internal/client/model/metastore.go b/cli/internal/client/model/metastore.go
index 99b2e6afb..69f55a05e 100644
--- a/cli/internal/client/model/metastore.go
+++ b/cli/internal/client/model/metastore.go
@@ -101,6 +101,36 @@ type DdlPage[T any] struct {
 	Content []T   `json:"content"`
 }
 
+// MetastoreDumpPage is one page of the metastore dump, which the server returns as a Spring Page.
+type MetastoreDumpPage struct {
+	Content          []MetastoreDumpRow `json:"content"`
+	TotalElements    int64              `json:"totalElements"`
+	NumberOfElements int                `json:"numberOfElements"`
+	Last             bool               `json:"last"`
+}
+
+// MetastoreDumpRow deliberately omits the fields the census does not read.
+//
+// The server sends the whole encoded value in "v", plus a decoded property map, a timestamp and a
+// direction. None of them are declared here, so the decoder walks past them instead of allocating a
+// string per row for data that would be thrown away - the whole point is to count a table that is
+// too large to hold.
+type MetastoreDumpRow struct {
+	ID      int64                 `json:"id"`
+	K       string                `json:"k"`
+	Decoded *MetastoreDecodedEdge `json:"decoded"`
+}
+
+// MetastoreDecodedEdge is the server's decode of one row. Src and Tgt are typed as any because the
+// server declares them as Object; metadata schemas make them strings, and anything else is still
+// printable.
+type MetastoreDecodedEdge struct {
+	Active  bool  `json:"active"`
+	Src     any   `json:"src"`
+	Tgt     any   `json:"tgt"`
+	LabelID int64 `json:"labelId"`
+}
+
 type StorageCreateRequest struct {
 	Desc string                 `json:"desc"`
 	Type string                 `json:"type"`
diff --git a/cli/internal/command/command_type.go b/cli/internal/command/command_type.go
index 9f005ef6d..67bd42b1b 100644
--- a/cli/internal/command/command_type.go
+++ b/cli/internal/command/command_type.go
@@ -85,6 +85,11 @@ var (
 		name:    "migrate",
 		command: "migrate <plan [-o <file>]|apply [-i <file>]>",
 	}
+
+	TypeJdbcMetastore = Type{
+		name:    "jdbc-metastore",
+		command: "jdbc-metastore <remaining> --capacity <n> [--page-size <n>] [--max-pages <n>] [--max-rows <n>]",
+	}
 )
 
 func (t Type) GetName() string {
diff --git a/cli/internal/command/jdbc_metastore_remaining.go b/cli/internal/command/jdbc_metastore_remaining.go
new file mode 100644
index 000000000..98bcb8c8a
--- /dev/null
+++ b/cli/internal/command/jdbc_metastore_remaining.go
@@ -0,0 +1,442 @@
+package command
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"sort"
+	"strconv"
+	"strings"
+	"syscall"
+
+	"github.com/kakao/actionbase/internal/client"
+	clientModel "github.com/kakao/actionbase/internal/client/model"
+	"github.com/kakao/actionbase/internal/command/model"
+	"github.com/kakao/actionbase/internal/util"
+)
+
+const (
+	// The OSS default for the server's metadataFetchLimit. A metadata scan asks the metastore for that
+	// many rows and filters inactive ones out afterwards, so tombstones spend the budget a live scan
+	// needs. Past it, a page is truncated and nothing says so.
+	//
+	// Named here only to say so in the error when --limit is missing. It is deliberately not a default
+	// for --limit: a deployment overrides it with graph.metadata-fetch-limit and the server does not
+	// expose the value - actuator masks it - so guessing produces confident, wrong verdicts against
+	// exactly the servers worth asking about.
+	ossMetadataFetchLimit = 1000
+
+	// Large on purpose, and the knob that decides what the server pays. Every page request runs a
+	// full COUNT(*) on top of its own LIMIT/OFFSET query, and the OFFSET cost grows with depth, so a
+	// table is far cheaper in few large pages than in many small ones. The census retains nothing per
+	// row, so the page only has to decode comfortably.
+	defaultRemainingPageSize = 500
+	maxRemainingPageSize     = 1000
+
+	// Two caps, because they bound different things. Rows cover a table larger than expected; pages
+	// cover what rows cannot - a server answering with short pages that never say they are last would
+	// advance the row count a row at a time, with a COUNT(*) behind every request.
+	defaultRemainingMaxRows  = 500000
+	defaultRemainingMaxPages = 2000
+
+	// The census holds one counter per scan window. More distinct windows than this is not a table
+	// this command can summarise, so it stops growing rather than grow without bound.
+	maxRemainingWindows = 10000
+)
+
+// metadataLabels says which metadata a scan of each label reads, in the order the report prints
+// them. Declaration order is the report order, so there is one list rather than a map and a parallel
+// slice that could disagree.
+//
+// A label's id is ValueUtils.stringHash(fullQualifiedName) - xxhash32 with seed 0 over a fixed name
+// - so these are constants rather than something to recompute here. Printed from the engine, and
+// they hold on 0.4.x too, which shares the names, the seed and the derivation.
+var metadataLabels = []struct {
+	id   int64
+	name string
+}{
+	{1773822659, "service"},
+	{-1834823921, "storage"},
+	{300431312, "label"},
+	{-641671585, "alias"},
+	{-1531313765, "info"},
+	{1397921084, "online_meta"},
+	{-1296613462, "nil"},
+}
+
+// window is one metadata scan prefix: everything a single getAll(name) reads. getAll builds its scan
+// key from encodeHashEdgeKeyPrefix(src, labelId) and the limit applies per key, so (src, labelId) is
+// exactly the unit the limit bounds. Both come off the server's own decode of each row.
+type window struct {
+	kind       string
+	namespace  string
+	src        string
+	total      int
+	tombstoned int
+}
+
+func (w window) live() int { return w.total - w.tombstoned }
+
+type windowKey struct {
+	src     string
+	labelID int64
+}
+
+// census is fixed-size state. Rows are folded in and dropped; none are retained.
+type census struct {
+	windows     map[windowKey]*window
+	rows        int
+	undecodable int
+	badKey      string
+	dropped     int
+}
+
+func newCensus() *census {
+	return &census{windows: make(map[windowKey]*window)}
+}
+
+func (c *census) add(rows []clientModel.MetastoreDumpRow) {
+	for i := range rows {
+		c.rows++
+		decoded := rows[i].Decoded
+		if decoded == nil {
+			// The server could not read it. It still occupies a scan window, but there is no way to
+			// say which one.
+			c.undecodable++
+			if c.badKey == "" {
+				c.badKey = rows[i].K
+			}
+			continue
+		}
+
+		key := windowKey{src: fmt.Sprintf("%v", decoded.Src), labelID: decoded.LabelID}
+		w := c.windows[key]
+		if w == nil {
+			if len(c.windows) >= maxRemainingWindows {
+				c.dropped++
+				continue
+			}
+			w = &window{kind: kindOf(decoded.LabelID), namespace: namespaceOf(key.src), src: key.src}
+			c.windows[key] = w
+		}
+		w.total++
+		if !decoded.Active {
+			w.tombstoned++
+		}
+	}
+}
+
+// namespaceOf splits the owner off the front of a scan source. Sources are `namespace:service`, and
+// the namespace is the tenant that owns the metadata - which matters because a metastore table can
+// be shared: several deployments point at one table and each sees the others' rows. Reporting
+// without it hands an operator another tenant's window as if it were theirs.
+//
+// A source with no separator is its own namespace rather than an error. The report is a diagnostic;
+// an unfamiliar source shape should still be counted and shown.
+func namespaceOf(src string) string {
+	namespace, _, _ := strings.Cut(src, ":")
+	return namespace
+}
+
+// namespaceTotal is what a namespace's heading says, which has to be known before its first window
+// is printed - and is also what orders the namespaces.
+type namespaceTotal struct{ rows, windows int }
+
+// ordered is the report, in print order: namespaces biggest first, known kinds in their declared
+// order and the rest after by name, fullest window first within a kind.
+//
+// One sorted slice rather than a tree of groups. Every window prints, so the report is a single walk
+// down this list that starts a heading whenever the namespace changes.
+func (c *census) ordered() ([]*window, map[string]namespaceTotal) {
+	totals := make(map[string]namespaceTotal)
+	out := make([]*window, 0, len(c.windows))
+	for _, w := range c.windows {
+		t := totals[w.namespace]
+		t.rows += w.total
+		t.windows++
+		totals[w.namespace] = t
+		out = append(out, w)
+	}
+
+	sort.Slice(out, func(i, j int) bool {
+		a, b := out[i], out[j]
+		switch {
+		case a.namespace != b.namespace && totals[a.namespace].rows != totals[b.namespace].rows:
+			return totals[a.namespace].rows > totals[b.namespace].rows
+		case a.namespace != b.namespace:
+			return a.namespace < b.namespace
+		case kindRank(a.kind) != kindRank(b.kind):
+			return kindRank(a.kind) < kindRank(b.kind)
+		case a.kind != b.kind:
+			return a.kind < b.kind
+		case a.total != b.total:
+			return a.total > b.total
+		default:
+			return a.src < b.src
+		}
+	})
+	return out, totals
+}
+
+// kindRank orders the known kinds ahead of anything unrecognised, which sorts after them by name.
+func kindRank(kind string) int {
+	for i, known := range metadataLabels {
+		if known.name == kind {
+			return i
+		}
+	}
+	return len(metadataLabels)
+}
+
+func kindOf(labelID int64) string {
+	for _, known := range metadataLabels {
+		if known.id == labelID {
+			return known.name
+		}
+	}
+	return fmt.Sprintf("labelId=%d", labelID)
+}
+
+type JdbcMetastoreRemaining struct {
+	client *client.ActionbaseClient
+}
+
+func NewJdbcMetastoreRemaining(c *client.ActionbaseClient) *JdbcMetastoreRemaining {
+	return &JdbcMetastoreRemaining{client: c}
+}
+
+type remainingOptions struct {
+	capacity int
+	pageSize int
+	maxRows  int
+	maxPages int
+}
+
+func (m *JdbcMetastoreRemaining) Execute(args []string) *model.Response {
+	if len(args) < 1 || args[0] != "remaining" {
+		return model.Fail(fmt.Sprintf("Usage: %s", m.GetType().GetCommand()))
+	}
+
+	opts, failure := parseRemainingOptions(args[1:])
+	if failure != nil {
+		return failure
+	}
+
+	// One cheap page first, both to learn the table size and to find out whether this server serves
+	// the endpoint at all. It was removed after 0.4.x, so a 404 is an answer, not a failure.
+	probe := m.client.DumpMetastore(0, 1)
+	switch {
+	case probe.StatusCode == http.StatusNotFound:
+		return model.Fail(fmt.Sprintf(
+			"No %s on this server (0.4.x only), so this cannot be measured here.",
+			client.MetastoreDumpPath))
+	case probe.StatusCode == http.StatusUnauthorized:
+		return model.Fail("Unauthorized. Set --authKey or ACT_API_KEY.")
+	case probe.IsError():
+		return model.Fail(fmt.Sprintf("Failed to read the JDBC metastore: %s", describeError(probe)))
+	}
+
+	totalRows := probe.Body.TotalElements
+	// One line, because an operator reading it already knows what a metastore is. The page count is
+	// here because each page costs the server a COUNT(*).
+	util.Println(fmt.Sprintf("%d rows, capacity %d/window, pages of %d (max %d). Ctrl-C reports partial.",
+		totalRows, opts.capacity, opts.pageSize, opts.maxPages))
+
+	stop, stopped := interruptible()
+	defer stopped()
+
+	c, stoppedBy, err := walk(m.fetch, opts, totalRows, stop)
+	if err != nil {
+		return model.Fail(fmt.Sprintf("Failed to read the JDBC metastore: %s", err))
+	}
+
+	return report(c, opts, totalRows, stoppedBy)
+}
+
+func (m *JdbcMetastoreRemaining) fetch(page, size int) (*clientModel.MetastoreDumpPage, error) {
+	resp := m.client.DumpMetastore(page, size)
+	if resp.IsError() {
+		return nil, fmt.Errorf("page %d: %s", page, describeError(resp))
+	}
+	return resp.Body, nil
+}
+
+func parseRemainingOptions(args []string) (remainingOptions, *model.Response) {
+	parser := util.ParseArgs(args)
+	opts := remainingOptions{
+		pageSize: defaultRemainingPageSize,
+		maxRows:  defaultRemainingMaxRows,
+		maxPages: defaultRemainingMaxPages,
+	}
+
+	// Required rather than defaulted. Every verdict here is a comparison against this number, and the
+	// server will not say what its own is, so a default would decide the answer while looking like a
+	// detail.
+	if _, ok := parser.Get("capacity"); !ok {
+		return opts, model.Fail(fmt.Sprintf(
+			"--capacity is required: the server's graph.metadata-fetch-limit, which it does not expose"+
+				" (actuator masks it). OSS default %d.", ossMetadataFetchLimit))
+	}
+
+	for _, flag := range []struct {
+		name string
+		into *int
+		max  int
+	}{
+		{"capacity", &opts.capacity, 0},
+		{"page-size", &opts.pageSize, maxRemainingPageSize},
+		{"max-rows", &opts.maxRows, 0},
+		{"max-pages", &opts.maxPages, 0},
+	} {
+		v, ok := parser.Get(flag.name)
+		if !ok {
+			continue
+		}
+		parsed, err := strconv.Atoi(v)
+		if err != nil || parsed < 1 || (flag.max > 0 && parsed > flag.max) {
+			bound := "a positive integer"
+			if flag.max > 0 {
+				bound = fmt.Sprintf("between 1 and %d", flag.max)
+			}
+			return opts, model.Fail(fmt.Sprintf("--%s must be %s", flag.name, bound))
+		}
+		*flag.into = parsed
+	}
+
+	return opts, nil
+}
+
+// interruptible reports whether Ctrl-C has been pressed, and stops listening when done. The walk
+// checks it once per page and returns on the first true, so there is nothing to remember - it stops
+// and reports what it has, which beats losing the census to a killed process. Nothing here writes,
+// so there is nothing to unwind.
+func interruptible() (aborted func() bool, stop func()) {
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
+	return func() bool {
+			select {
+			case <-signals:
+				return true
+			default:
+				return false
+			}
+		}, func() {
+			signal.Stop(signals)
+		}
+}
+
+type fetchPage func(page, size int) (*clientModel.MetastoreDumpPage, error)
+
+// walk pages to the end of the table, folding each page into the census and dropping it.
+//
+// Five ways to stop, and it needs all five. A last page or an empty one are the ordinary two. The
+// row and page caps are the two ceilings, and the page cap is the one that matters against a server
+// handing back short pages that never say they are last. Reading as many rows as the table claims to
+// hold ends it without trusting the server's own `last`. Whichever cap fired is named, so an
+// incomplete census never reads as a complete one.
+func walk(fetch fetchPage, opts remainingOptions, totalRows int64, aborted func() bool) (*census, string, error) {
+	c := newCensus()
+	for page := 0; ; page++ {
+		switch {
+		case aborted != nil && aborted():
+			return c, "Ctrl-C", nil
+		case page >= opts.maxPages:
+			return c, fmt.Sprintf("--max-pages %d", opts.maxPages), nil
+		case c.rows >= opts.maxRows:
+			return c, fmt.Sprintf("--max-rows %d", opts.maxRows), nil
+		}
+
+		body, err := fetch(page, opts.pageSize)
+		if err != nil {
+			return c, "", err
+		}
+		c.add(body.Content)
+
+		if body.Last || len(body.Content) == 0 || (totalRows > 0 && int64(c.rows) >= totalRows) {
+			return c, "", nil
+		}
+	}
+}
+
+func report(c *census, opts remainingOptions, totalRows int64, stoppedBy string) *model.Response {
+	capacity := opts.capacity
+	windows, totals := c.ordered()
+	over, tight := 0, 0
+
+	// Every window, always. Printing only the fullest of each kind plus whatever crossed a threshold
+	// meant the report decided what mattered, and it decided wrong: a window holding half the
+	// capacity sat behind one holding a little more and never appeared. The list is short - a table
+	// here has tens of windows, not thousands - and the cap on the census bounds it.
+	namespace, kind := "", ""
+	for _, w := range windows {
+		if w.total > capacity {
+			over++
+		} else if w.total*10 >= capacity*9 {
+			tight++ // within a tenth of capacity
+		}
+
+		if w.namespace != namespace {
+			namespace, kind = w.namespace, ""
+			t := totals[namespace]
+			util.Println(fmt.Sprintf("%-10s %5d rows  %s", namespace, t.rows, plural(t.windows, "window")))
+		}
+		label := ""
+		if w.kind != kind {
+			kind, label = w.kind, w.kind
+		}
+		util.Println(fmt.Sprintf("  %-12s %-11s %-22s %5d/%-5d live %-6d tomb %d",
+			label, room(w, capacity), w.src, w.total, capacity, w.live(), w.tombstoned))
+	}
+
+	if c.undecodable > 0 {
+		// Flush left rather than indented with a namespace's windows: an undecodable row has no
+		// namespace to belong to, so showing it inside one would read as a claim about that namespace.
+		util.Println(fmt.Sprintf("%d undecodable (window unknown, e.g. k=%q)", c.undecodable, c.badKey))
+	}
+	if c.dropped > 0 {
+		util.Println(fmt.Sprintf("%d rows uncounted, over %d windows", c.dropped, maxRemainingWindows))
+	}
+
+	summary := fmt.Sprintf("%d/%d rows, %s, %s.", c.rows, totalRows,
+		plural(len(c.windows), "window"), plural(len(totals), "namespace"))
+	switch {
+	case stoppedBy != "":
+		return model.Fail(fmt.Sprintf("%s Stopped at %s - incomplete, %d over is a lower bound.",
+			summary, stoppedBy, over))
+	case over > 0:
+		return model.Fail(fmt.Sprintf("%s %d over, %d near. An over-capacity window is already"+
+			" truncated and does not say so.", summary, over, tight))
+	default:
+		return model.SuccessWithResult(fmt.Sprintf("%s 0 over, %d near.", summary, tight))
+	}
+}
+
+func plural(n int, noun string) string {
+	if n == 1 {
+		return fmt.Sprintf("1 %s", noun)
+	}
+	return fmt.Sprintf("%d %ss", n, noun)
+}
+
+func room(w *window, capacity int) string {
+	if w.total > capacity {
+		return fmt.Sprintf("OVER by %d", w.total-capacity)
+	}
+	return fmt.Sprintf("%d left", capacity-w.total)
+}
+
+func describeError[T any](r *client.Response[T]) string {
+	if r.Error != nil {
+		return r.Error.Error()
+	}
+	return fmt.Sprintf("HTTP %d", r.StatusCode)
+}
+
+func (m *JdbcMetastoreRemaining) GetDescription() string {
+	return "Report how much room each metadata scan window has left in the JDBC metastore"
+}
+
+func (m *JdbcMetastoreRemaining) GetType() Type {
+	return TypeJdbcMetastore
+}
diff --git a/cli/internal/command/jdbc_metastore_remaining_test.go b/cli/internal/command/jdbc_metastore_remaining_test.go
new file mode 100644
index 000000000..1fc436b6a
--- /dev/null
+++ b/cli/internal/command/jdbc_metastore_remaining_test.go
@@ -0,0 +1,435 @@
+package command
+
+import (
+	"errors"
+	"strconv"
+	"strings"
+	"testing"
+
+	clientModel "github.com/kakao/actionbase/internal/client/model"
+)
+
+const (
+	serviceLabel = 1773822659
+	storageLabel = -1834823921
+	tableLabel   = 300431312
+	aliasLabel   = -641671585
+)
+
+func row(src, tgt string, labelID int64, active bool) clientModel.MetastoreDumpRow {
+	return clientModel.MetastoreDumpRow{
+		K: src + ":" + tgt,
+		Decoded: &clientModel.MetastoreDecodedEdge{
+			Active: active, Src: src, Tgt: tgt, LabelID: labelID,
+		},
+	}
+}
+
+// page builds n rows in one window. Only the walk's bounds care about the contents.
+func page(n int, total int64, last bool) *clientModel.MetastoreDumpPage {
+	rows := make([]clientModel.MetastoreDumpRow, 0, n)
+	for i := 0; i < n; i++ {
+		rows = append(rows, row("prod:wish", "t", tableLabel, true))
+	}
+	return &clientModel.MetastoreDumpPage{
+		Content: rows, TotalElements: total, NumberOfElements: n, Last: last,
+	}
+}
+
+func onlyWindow(t *testing.T, c *census) *window {
+	t.Helper()
+	if len(c.windows) != 1 {
+		t.Fatalf("expected 1 window, got %d", len(c.windows))
+	}
+	for _, w := range c.windows {
+		return w
+	}
+	return nil
+}
+
+func TestCensusCountsTombstonesTowardTheWindow(t *testing.T) {
+	c := newCensus()
+	c.add([]clientModel.MetastoreDumpRow{
+		row("prod:wish", "likes", tableLabel, true),
+		row("prod:wish", "gone", tableLabel, false),
+		row("prod:wish", "also_gone", tableLabel, false),
+	})
+
+	w := onlyWindow(t, c)
+	if w.total != 3 || w.tombstoned != 2 || w.live() != 1 {
+		t.Errorf("total/tombstoned/live = %d/%d/%d, want 3/2/1", w.total, w.tombstoned, w.live())
+	}
+}
+
+func TestCensusSeparatesWindowsBySourceAndLabel(t *testing.T) {
+	c := newCensus()
+	c.add([]clientModel.MetastoreDumpRow{
+		row("prod:wish", "likes", tableLabel, true),
+		row("prod:talk", "likes", tableLabel, true), // same label, other service
+		row("prod:wish", "likes", aliasLabel, true), // same service, other label
+	})
+
+	if len(c.windows) != 3 {
+		t.Fatalf("expected 3 windows, got %d", len(c.windows))
+	}
+}
+
+func TestCensusFoldsAcrossPagesWithoutRetainingRows(t *testing.T) {
+	c := newCensus()
+	for i := 0; i < 5; i++ {
+		rows := []clientModel.MetastoreDumpRow{
+			row("prod:wish", "a", tableLabel, true),
+			row("prod:wish", "b", tableLabel, false),
+		}
+		c.add(rows)
+		// Overwriting the slice the caller handed in must not disturb the counters.
+		for j := range rows {
+			rows[j] = clientModel.MetastoreDumpRow{}
+		}
+	}
+
+	w := onlyWindow(t, c)
+	if c.rows != 10 || w.total != 10 || w.tombstoned != 5 {
+		t.Errorf("rows/total/tombstoned = %d/%d/%d, want 10/10/5", c.rows, w.total, w.tombstoned)
+	}
+}
+
+func TestCensusCountsUndecodableRowsSeparately(t *testing.T) {
+	c := newCensus()
+	c.add([]clientModel.MetastoreDumpRow{
+		row("prod:wish", "likes", tableLabel, true),
+		{K: "not-an-encoded-key", Decoded: nil},
+	})
+
+	if c.undecodable != 1 || c.badKey != "not-an-encoded-key" {
+		t.Errorf("undecodable/badKey = %d/%q, want 1/not-an-encoded-key", c.undecodable, c.badKey)
+	}
+	if len(c.windows) != 1 {
+		t.Errorf("windows = %d, want 1: an undecodable row joins none", len(c.windows))
+	}
+	if c.rows != 2 {
+		t.Errorf("rows = %d, want 2: it was still read", c.rows)
+	}
+}
+
+func TestCensusStopsGrowingAtTheWindowCap(t *testing.T) {
+	c := newCensus()
+	rows := make([]clientModel.MetastoreDumpRow, 0, maxRemainingWindows+10)
+	for i := 0; i < maxRemainingWindows+10; i++ {
+		rows = append(rows, row("prod:svc", "t", int64(i), true))
+	}
+	c.add(rows)
+
+	if len(c.windows) != maxRemainingWindows {
+		t.Errorf("windows = %d, want the cap %d", len(c.windows), maxRemainingWindows)
+	}
+	if c.dropped != 10 {
+		t.Errorf("dropped = %d, want 10", c.dropped)
+	}
+	if c.rows != maxRemainingWindows+10 {
+		t.Errorf("rows = %d, want every row read", c.rows)
+	}
+}
+
+// printOrder is the report as a list of "kind src", which is the whole of what ordered() decides.
+func printOrder(c *census) []string {
+	var out []string
+	windows, _ := c.ordered()
+	for _, w := range windows {
+		out = append(out, w.kind+" "+w.src)
+	}
+	return out
+}
+
+func TestCensusOrdersKnownKindsFirstThenFullestWindow(t *testing.T) {
+	c := newCensus()
+	c.add([]clientModel.MetastoreDumpRow{
+		row("prod:wish", "a", aliasLabel, true),
+		row("prod:small", "b", tableLabel, true),
+		row("prod:big", "c", tableLabel, true),
+		row("prod:big", "d", tableLabel, false),
+		row("prod:origin", "e", storageLabel, true),
+		row("prod:origin", "f", serviceLabel, true),
+		row("prod:wish", "g", 424242, true),
+	})
+
+	want := []string{
+		"service prod:origin",
+		"storage prod:origin",
+		"label prod:big", // two rows, so it leads the one-row window
+		"label prod:small",
+		"alias prod:wish",
+		"labelId=424242 prod:wish", // unknown kinds sort after the known ones
+	}
+	if got := printOrder(c); strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Errorf("order =\n  %s\nwant\n  %s", strings.Join(got, "\n  "), strings.Join(want, "\n  "))
+	}
+}
+
+func TestKindOfNamesTheKnownMetadataLabels(t *testing.T) {
+	for labelID, want := range map[int64]string{
+		serviceLabel: "service",
+		storageLabel: "storage",
+		tableLabel:   "label",
+		aliasLabel:   "alias",
+		12345:        "labelId=12345",
+	} {
+		if got := kindOf(labelID); got != want {
+			t.Errorf("kindOf(%d) = %q, want %q", labelID, got, want)
+		}
+	}
+}
+
+func TestCensusKeepsANonStringSourcePrintable(t *testing.T) {
+	c := newCensus()
+	c.add([]clientModel.MetastoreDumpRow{{
+		K:       "k",
+		Decoded: &clientModel.MetastoreDecodedEdge{Active: true, Src: float64(42), Tgt: float64(7), LabelID: 3},
+	}})
+
+	if w := onlyWindow(t, c); w.src != "42" {
+		t.Errorf("src = %q, want %q", w.src, "42")
+	}
+}
+
+func TestWalkStopsWhenTheServerSaysThePageIsLast(t *testing.T) {
+	requests := 0
+	fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) {
+		requests++
+		return page(2, 4, p == 1), nil
+	}
+
+	c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 1000, maxPages: 100}, 4, nil)
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if stoppedBy != "" {
+		t.Errorf("stoppedBy = %q, want empty: a complete walk hit no cap", stoppedBy)
+	}
+	if requests != 2 || c.rows != 4 {
+		t.Errorf("requests/rows = %d/%d, want 2/4", requests, c.rows)
+	}
+}
+
+func TestWalkStopsOnAnEmptyPageEvenIfLastNeverArrives(t *testing.T) {
+	fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) {
+		if p == 0 {
+			return page(2, 100, false), nil
+		}
+		return page(0, 100, false), nil
+	}
+
+	c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 1000, maxPages: 100}, 100, nil)
+
+	if err != nil || stoppedBy != "" || c.rows != 2 {
+		t.Errorf("err/stoppedBy/rows = %v/%q/%d, want nil/empty/2", err, stoppedBy, c.rows)
+	}
+}
+
+// The case the row cap alone does not cover: a server answering every request with one row and never
+// admitting to being last. Without a page cap this is a request storm, and each of those requests
+// costs the server a COUNT(*).
+func TestWalkStopsAtThePageCapWhenPagesStayShort(t *testing.T) {
+	requests := 0
+	fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) {
+		requests++
+		return page(1, 1_000_000_000, false), nil
+	}
+
+	c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 500, maxRows: 500000, maxPages: 20}, 1_000_000_000, nil)
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if requests != 20 || c.rows != 20 {
+		t.Errorf("requests/rows = %d/%d, want 20/20: the page cap, not the row cap", requests, c.rows)
+	}
+	if stoppedBy != "--max-pages 20" {
+		t.Errorf("stoppedBy = %q, want the page cap named", stoppedBy)
+	}
+}
+
+func TestWalkStopsAtTheRowCapWhenPagesAreFull(t *testing.T) {
+	fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) {
+		return page(size, 1_000_000_000, false), nil
+	}
+
+	c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 100, maxRows: 250, maxPages: 1000}, 1_000_000_000, nil)
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if c.rows != 300 {
+		t.Errorf("rows = %d, want 300: the cap is checked between pages", c.rows)
+	}
+	if stoppedBy != "--max-rows 250" {
+		t.Errorf("stoppedBy = %q, want the row cap named", stoppedBy)
+	}
+}
+
+// The table's own row count bounds the walk, so a `last` that never comes cannot extend it past the
+// end of the table.
+func TestWalkStopsOnceItHasReadTheWholeTable(t *testing.T) {
+	requests := 0
+	fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) {
+		requests++
+		return page(size, 10, false), nil
+	}
+
+	c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 5, maxRows: 100000, maxPages: 1000}, 10, nil)
+
+	if err != nil || stoppedBy != "" {
+		t.Errorf("err/stoppedBy = %v/%q, want nil/empty", err, stoppedBy)
+	}
+	if requests != 2 || c.rows != 10 {
+		t.Errorf("requests/rows = %d/%d, want 2/10", requests, c.rows)
+	}
+}
+
+func TestWalkStopsOnInterruptAndKeepsWhatItRead(t *testing.T) {
+	requests := 0
+	fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) {
+		requests++
+		return page(2, 1_000_000, false), nil
+	}
+
+	c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 500000, maxPages: 1000}, 1_000_000, func() bool { return requests >= 3 })
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if stoppedBy != "Ctrl-C" {
+		t.Errorf("stoppedBy = %q, want Ctrl-C", stoppedBy)
+	}
+	if c.rows != 6 {
+		t.Errorf("rows = %d, want the 6 rows read before the interrupt", c.rows)
+	}
+}
+
+func TestWalkReturnsWhatItReadWhenAPageFails(t *testing.T) {
+	fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) {
+		if p == 0 {
+			return page(2, 100, false), nil
+		}
+		return nil, errors.New("page 1: HTTP 500")
+	}
+
+	c, _, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 1000, maxPages: 100}, 100, nil)
+
+	if err == nil {
+		t.Fatal("expected the failure to surface")
+	}
+	if c.rows != 2 {
+		t.Errorf("rows = %d, want the 2 rows already read", c.rows)
+	}
+}
+
+func TestParseRemainingOptionsRequiresTheCapacity(t *testing.T) {
+	_, failure := parseRemainingOptions(nil)
+	if failure == nil {
+		t.Fatal("a census with no --capacity should have been rejected")
+	}
+	if !strings.Contains(*failure.ErrorMessage, "graph.metadata-fetch-limit") {
+		t.Errorf("error = %q, want it to name the server property", *failure.ErrorMessage)
+	}
+}
+
+func TestParseRemainingOptionsDefaultsAndRejects(t *testing.T) {
+	opts, failure := parseRemainingOptions([]string{"--capacity", "2000"})
+	if failure != nil {
+		t.Fatalf("defaults should parse: %v", *failure.ErrorMessage)
+	}
+	if opts.capacity != 2000 {
+		t.Errorf("opts = %+v, want capacity 2000", opts)
+	}
+	if opts.pageSize != defaultRemainingPageSize || opts.maxPages != defaultRemainingMaxPages {
+		t.Errorf("defaults = %+v, want page size %d and max pages %d",
+			opts, defaultRemainingPageSize, defaultRemainingMaxPages)
+	}
+
+	for _, args := range [][]string{
+		{"--capacity", "2000", "--page-size", "0"},
+		{"--capacity", "2000", "--page-size", strconv.Itoa(maxRemainingPageSize + 1)},
+		{"--capacity", "2000", "--max-pages", "0"},
+		{"--capacity", "2000", "--max-rows", "-1"},
+		{"--capacity", "abc"},
+	} {
+		if _, failure := parseRemainingOptions(args); failure == nil {
+			t.Errorf("%v should have been rejected", args)
+		}
+	}
+}
+
+func TestNamespaceOfSplitsTheOwnerOffTheSource(t *testing.T) {
+	for src, want := range map[string]string{
+		"stg:origin":     "stg",
+		"crm:aircrm":     "crm",
+		"a:b:c":          "a",
+		"no-separator":   "no-separator",
+		":leading-colon": "",
+	} {
+		if got := namespaceOf(src); got != want {
+			t.Errorf("namespaceOf(%q) = %q, want %q", src, got, want)
+		}
+	}
+}
+
+// A shared metastore table holds every namespace pointed at it, so the census has to keep them
+// apart. This is the shape of `ab_default`, where several deployments read one table.
+func sharedTable() *census {
+	c := newCensus()
+	rows := []clientModel.MetastoreDumpRow{}
+	for i := 0; i < 30; i++ {
+		rows = append(rows, row("stg:origin", "s"+strconv.Itoa(i), storageLabel, false))
+	}
+	for i := 0; i < 8; i++ {
+		rows = append(rows, row("crm:aircrm", "t"+strconv.Itoa(i), tableLabel, true))
+	}
+	for i := 0; i < 2; i++ {
+		rows = append(rows, row("bmt:origin", "s"+strconv.Itoa(i), storageLabel, true))
+	}
+	c.add(rows)
+	return c
+}
+
+func TestCensusOrdersBiggestNamespaceFirst(t *testing.T) {
+	c := sharedTable()
+
+	windows, totals := c.ordered()
+	if len(totals) != 3 || totals["stg"].rows != 30 || totals["stg"].windows != 1 {
+		t.Fatalf("totals = %+v, want 3 namespaces and stg holding 30 rows in 1 window", totals)
+	}
+
+	var got []string
+	for _, w := range windows {
+		got = append(got, w.namespace)
+	}
+	if want := "stg,crm,bmt"; strings.Join(got, ",") != want {
+		t.Errorf("namespaces = %v, want %s", got, want)
+	}
+}
+
+// Every window prints, not just the fullest of each kind. The report used to decide what mattered
+// and hid a window holding half the capacity behind one holding a little more.
+func TestReportPrintsEveryWindow(t *testing.T) {
+	c := sharedTable()
+
+	got := report(c, remainingOptions{capacity: 2000}, 45, "")
+	if !got.IsSuccess {
+		t.Fatalf("nothing is over 2000: %v", *got.ErrorMessage)
+	}
+	if !strings.Contains(*got.Result, "3 windows, 3 namespaces") {
+		t.Errorf("summary = %q, want every window and namespace counted", *got.Result)
+	}
+}
+
+func TestReportFailsOnAnOverCapacityWindow(t *testing.T) {
+	got := report(sharedTable(), remainingOptions{capacity: 10}, 45, "")
+	if got.IsSuccess {
+		t.Fatal("a 30-row window against a capacity of 10 should fail")
+	}
+	if !strings.Contains(*got.ErrorMessage, "over-capacity") {
+		t.Errorf("error = %q, want it to name the truncation", *got.ErrorMessage)
+	}
+}
diff --git a/cli/internal/runner/actionbase_runner.go b/cli/internal/runner/actionbase_runner.go
index 9b6dffcdf..15f109943 100644
--- a/cli/internal/runner/actionbase_runner.go
+++ b/cli/internal/runner/actionbase_runner.go
@@ -74,6 +74,7 @@ func NewActionbaseCommandLineRunner(version, host string, authKey *string, curre
 	runner.RegisterCommand(command.TypeGuide.GetName(), command.NewGuide(runner, actionbaseClient))
 	runner.RegisterCommand(command.TypeDoctor.GetName(), command.NewDoctor(actionbaseClient))
 	runner.RegisterCommand(command.TypeMigrate.GetName(), command.NewMigrate(actionbaseClient))
+	runner.RegisterCommand(command.TypeJdbcMetastore.GetName(), command.NewJdbcMetastoreRemaining(actionbaseClient))
 
 	return runner
 }