Name the system table ctk cfr sys-export is reading

CHANGES.md

@@ -1,6 +1,8 @@
 # Changelog
 
 ## Unreleased
+- `ctk cfr sys-export` names the table it is reading in its progress bar, and logs
+  each read's row count, size and duration at debug level.
 - Fixed `ctk cfr sys-import` reporting success while the cluster rejected rows.
   Thanks, @hammerhead.
 - Breaking change: `ctk cfr sys-import` exits non-zero when a table did not restore in

cratedb_toolkit/cfr/systable.py

@@ -22,10 +22,12 @@
 import re
 import tarfile
 import tempfile
+import time
 import typing as t
 from pathlib import Path
 
 import orjsonl
+from boltons.strutils import bytes2human
 from tqdm.contrib.logging import logging_redirect_tqdm
 
 if t.TYPE_CHECKING:
@@ -305,16 +307,26 @@ def cratedb_version(self) -> str:
         return "unknown"
 
     def read_table(self, tablename: str, schema: t.Optional[str] = None) -> "pl.DataFrame":
+        """
+        Read one system table in full, logging its row count, in-memory size and duration.
+        """
         import polars as pl
 
         schema = schema or SystemTableKnowledge.SYS_SCHEMA
         sql = f'SELECT * FROM "{schema}"."{tablename}"'  # noqa: S608
         logger.debug(f"Running SQL: {sql}")
-        return pl.read_database(
+        started = time.monotonic()
+        frame = pl.read_database(
             query=sql,  # noqa: S608
             connection=self.adapter.connection,
             infer_schema_length=100_000,
         )
+        duration = time.monotonic() - started
+        logger.debug(
+            f"Read {schema}.{tablename}: {frame.height} rows, "
+            f"~{bytes2human(frame.estimated_size(), ndigits=1)} in memory, {duration:.3f}s"
+        )
+        return frame
 
     def redact(self, frame: "pl.DataFrame", schema: str, tablename: str) -> "pl.DataFrame":
         """
@@ -428,15 +440,17 @@ def _save_schema(self, bundle_path: Path, schema: str) -> None:
             self.data_failures.append({"schema": schema, "table": "*", "reason": f"{type(ex).__name__}: {ex}"})
             return
 
-        for tablename in tqdm(tablenames, desc=f"Exporting {schema}", disable=None):
-            logger.debug(f"Exporting table: {schema}.{tablename}")
-            self._save_table(
-                schema=schema,
-                tablename=tablename,
-                path_schema=path_schema,
-                path_data=path_data,
-                prefix=prefix,
-            )
+        with tqdm(tablenames, desc=f"Exporting {schema}", disable=None) as progress:
+            for tablename in progress:
+                progress.set_postfix_str(tablename)
+                logger.debug(f"Exporting table: {schema}.{tablename}")
+                self._save_table(
+                    schema=schema,
+                    tablename=tablename,
+                    path_schema=path_schema,
+                    path_data=path_data,
+                    prefix=prefix,
+                )
 
     def _save_table(self, schema: str, tablename: str, path_schema: Path, path_data: Path, prefix: str) -> None:
         """
@@ -492,22 +506,24 @@ def _save_definitions(self, bundle_path: Path) -> t.Dict[str, int]:
             self.definition_failures.append({"kind": "relations", "reason": f"{type(ex).__name__}: {ex}"})
             return counts
 
-        for relation in tqdm(relations, desc="Capturing definitions", disable=None):
-            schema = relation["table_schema"]
-            name = relation["table_name"]
-            # `SHOW CREATE TABLE` only works on regular tables. Views come from
-            # `information_schema.views` below; anything else is left alone.
-            if relation.get("table_type") != SystemTableKnowledge.BASE_TABLE_TYPE:
-                continue
-            try:
-                ddl = self.schema_capture.table_ddl(schema=schema, table=name)
-                (path_tables / f"{schema}.{name}.sql").write_text(ddl.rstrip() + "\n")
-                counts["tables"] += 1
-            except Exception as ex:
-                logger.warning(f"Could not capture definition of {schema}.{name}: {ex}")
-                self.definition_failures.append(
-                    {"kind": "table", "schema": schema, "name": name, "reason": f"{type(ex).__name__}: {ex}"}
-                )
+        with tqdm(relations, desc="Capturing definitions", disable=None) as progress:
+            for relation in progress:
+                schema = relation["table_schema"]
+                name = relation["table_name"]
+                progress.set_postfix_str(f"{schema}.{name}")
+                # `SHOW CREATE TABLE` only works on regular tables. Views come from
+                # `information_schema.views` below; anything else is left alone.
+                if relation.get("table_type") != SystemTableKnowledge.BASE_TABLE_TYPE:
+                    continue
+                try:
+                    ddl = self.schema_capture.table_ddl(schema=schema, table=name)
+                    (path_tables / f"{schema}.{name}.sql").write_text(ddl.rstrip() + "\n")
+                    counts["tables"] += 1
+                except Exception as ex:
+                    logger.warning(f"Could not capture definition of {schema}.{name}: {ex}")
+                    self.definition_failures.append(
+                        {"kind": "table", "schema": schema, "name": name, "reason": f"{type(ex).__name__}: {ex}"}
+                    )
 
         try:
             views = self.schema_capture.views()

tests/cfr/test_systable.py

@@ -1,8 +1,10 @@
 # ruff: noqa: E402
+import io
 import json
 import os.path
 import re
 import shutil
+import sys
 import tarfile
 from importlib.resources import files
 from pathlib import Path
@@ -65,6 +67,62 @@ def test_cfr_sys_export_success(cratedb, click_kwargs, tmp_path, caplog):
     assert len(data_files) >= 10
 
 
+def test_cfr_sys_export_reports_the_row_count_it_read(cratedb, click_kwargs, tmp_path, caplog):
+    """
+    Verify `ctk cfr sys-export` logs the row count of every table it reads.
+    """
+
+    runner = CliRunner(env={"CRATEDB_CLUSTER_URL": cratedb.database.dburi, "CFR_TARGET": str(tmp_path)}, **click_kwargs)
+    result = runner.invoke(cli, args="--debug sys-export", catch_exceptions=False)
+    assert result.exit_code == 0, result.output
+
+    reads = {
+        table: int(rows) for table, rows in re.findall(r"Read (\S+): (\d+) rows, ~\S+ in memory, [\d.]+s", caplog.text)
+    }
+
+    assert reads, f"No read was reported at all: {caplog.text[-2000:]}"
+    for tablename in ["sys.jobs_log", "sys.shards", "sys.allocations", "sys.nodes"]:
+        assert tablename in reads, f"Read of {tablename} not accounted for: {sorted(reads)}"
+
+    # Creating one relation moves `information_schema.tables` by exactly one.
+    assert reads["sys.nodes"] >= 1
+    tables_before = reads["information_schema.tables"]
+    cratedb.database.run_sql(f'CREATE TABLE "{TESTDRIVE_DATA_SCHEMA}".volume (id INT)')
+    caplog.clear()
+    result = runner.invoke(cli, args="--debug sys-export", catch_exceptions=False)
+    assert result.exit_code == 0, result.output
+    second = re.search(r"Read information_schema\.tables: (\d+) rows", caplog.text)
+    assert second, "The second export reported no read of information_schema.tables"
+    assert int(second.group(1)) == tables_before + 1
+
+
+def test_cfr_sys_export_progress_bar_names_the_table_in_flight(cratedb, click_kwargs, tmp_path):
+    """
+    Verify the export's progress bar names the table it is reading.
+    """
+    from cratedb_toolkit.cfr.systable import SystemTableExporter
+
+    class TerminalLike(io.StringIO):
+        """Keep tqdm enabled: it silences itself when its stream is not a terminal."""
+
+        def isatty(self):
+            return True
+
+    recorder = TerminalLike()
+    stderr, sys.stderr = sys.stderr, recorder
+    try:
+        SystemTableExporter(dburi=cratedb.database.dburi, target=tmp_path).save()
+    finally:
+        sys.stderr = stderr
+
+    # Only the bar's own frames; the redirected log lines share the stream.
+    frames = [line for line in recorder.getvalue().split("\r") if line.startswith("Exporting sys:")]
+    assert frames, "The progress bar produced no output"
+    labelled = {line.rsplit(", ", 1)[-1].rstrip("] ") for line in frames if line.endswith("]")}
+    for tablename in ["allocations", "jobs_log", "shards"]:
+        assert tablename in labelled, f"Bar never named sys.{tablename}; it named {sorted(labelled)}"
+
+
 def test_cfr_sys_export_to_archive_file(cratedb, click_kwargs, tmp_path, caplog):
     """
     Verify `ctk cfr sys-export some-file.tgz` works.