feat: add explicit --collectdatabase option

CHANGELOG.md

@@ -15,6 +15,21 @@ expose.
 
 ## [Unreleased]
 
+### Added
+
+- Added `--collectdatabase`/`S3GC_COLLECTDATABASE` to qualify the auxiliary
+  and run-log tables' database explicitly, instead of only as an embedded
+  `db.prefix_` string in `COLLECTTABLEPREFIX`. A production collect Job once
+  created its auxiliary table in the ClickHouse user's current database
+  because that embedded qualification was easy to omit. The two settings must
+  agree when both are given: a conflicting pair now fails before any S3 or
+  ClickHouse call, instead of silently picking one. `COLLECTTABLEPREFIX`
+  alone still works unchanged for existing deployments. Also fixed
+  `check_samples_match_partitioning()`, which always checked
+  `currentDatabase()` and so silently skipped its `--samples`-mismatch
+  warning whenever the auxiliary table lived in a different database — a
+  second, independent bug the same incident exposed.
+
 ### Documentation
 
 - Reworked the Kubernetes guidance into a phase-based production runbook and a

README.md

@@ -65,6 +65,7 @@ export S3GC_S3SECURE_FLAG=true
 export S3GC_S3DISKNAME=s3
 export S3GC_CLUSTERNAME='<clickhouse-cluster>'
 export S3GC_EXPECTED_REPLICAS=2
+export S3GC_COLLECTDATABASE='s3gc'
 export S3GC_COLLECTTABLEPREFIX='s3gc_example_'
 export S3GC_USEAGE=24
 ```

deploy/kubernetes/README.md

@@ -50,7 +50,7 @@ Set the values in `s3gc.env`, then check these five items before rendering:
 1. Set `CHHOST` to one per-replica ClickHouse Service and use the same host for every phase. The inventory table is node-local; a load-balanced Service can send a later phase to a replica without that table.
 2. Set the exact bucket, prefix, and underlying object-disk name. `S3PATH` may be empty. GCS commonly uses `gcs`; never use a `*_cache` disk.
 3. Set the actual `CLUSTERNAME` and `EXPECTED_REPLICAS` for clustered cleanup. Delete fails closed if this preflight does not match.
-4. Choose a unique, database-qualified `COLLECTTABLEPREFIX`, such as `s3gc.s3gc_<run>_`. Keep it after a partial delete so the replacement Job can use the deletion checkpoints. A bare prefix uses the ClickHouse user's current database, which can differ from `default` and cause a grant failure.
+4. Set `COLLECTDATABASE` to the auxiliary database, and choose a unique `COLLECTTABLEPREFIX` such as `s3gc_<run>_`. Keep the prefix after a partial delete so the replacement Job can use the deletion checkpoints. Without `COLLECTDATABASE`, the auxiliary table lands in the ClickHouse user's current database, which can differ from `default` and cause a grant failure; `COLLECTTABLEPREFIX` still accepts an embedded `db.prefix_` form for compatibility, but a value there that disagrees with `COLLECTDATABASE` is rejected at startup.
 5. Set `USEAGE_HOURS` to 24 or more. Raise it for slow merges or long mutations; production phases cannot lower it.
 
 For an installation-wide cleanup, set `S3PATH` to the parent prefix that
@@ -94,7 +94,7 @@ are suitable when they meet that cluster policy.
 
 Create a dedicated auxiliary database and ClickHouse user before the first run.
 Replace `<cluster-name>`, `<auxiliary-database>`, and the password placeholder.
-Use the same `<auxiliary-database>.` prefix in `COLLECTTABLEPREFIX`.
+Set the same `<auxiliary-database>` as `COLLECTDATABASE`.
 
 ```sql
 CREATE DATABASE IF NOT EXISTS <auxiliary-database>
@@ -211,7 +211,7 @@ windows. Start with [example.env](example.env).
 | Group | Required values | Notes |
 | --- | --- | --- |
 | Job | `JOB_NAME`, `NAMESPACE`, `IMAGE`, `IMAGE_PULL_SECRET`, `SERVICE_ACCOUNT`, `CREDENTIALS_SECRET` | `JOB_NAME` is a DNS label and becomes the durable run ID. Leave `IMAGE_PULL_SECRET` empty for the public image. |
-| ClickHouse | `CHHOST`, `CHPORT`, `CLUSTERNAME`, `EXPECTED_REPLICAS`, `COLLECTTABLEPREFIX` | Pin `CHHOST` to one replica for the entire cleanup. |
+| ClickHouse | `CHHOST`, `CHPORT`, `CLUSTERNAME`, `EXPECTED_REPLICAS`, `COLLECTTABLEPREFIX` | Pin `CHHOST` to one replica for the entire cleanup. `COLLECTDATABASE` (optional) qualifies the auxiliary/run-log tables' database. |
 | Object store | `S3IP`, `S3PORT`, `S3BUCKET`, `S3PATH`, `S3REGION`, `S3SECURE_FLAG`, `S3DISKNAME`, `S3AUTH`, `S3PROFILE` | `S3PROFILE` requires `S3AUTH=aws`. |
 | Limits | `SAMPLES`, `DELETE_BATCH_SIZE`, `USEAGE_HOURS`, `ACTIVE_DEADLINE_SECONDS`, `TTL_SECONDS_AFTER_FINISHED`, `MEMORY_REQUEST`, `MEMORY_LIMIT` | Keep `SAMPLES` unchanged after collect for partition pruning. |
 | Behavior | `PHASE`, `DELETE_CONFIRMATION`, `ORDER_BY_OBJPATH`, `VERBOSE` | The renderer requires the confirmation token for delete and development automation. |

deploy/kubernetes/example.env

@@ -22,9 +22,11 @@ CHHOST=clickhouse.example.svc.cluster.local
 CHPORT=8123
 CLUSTERNAME=example-cluster
 EXPECTED_REPLICAS=2
-# Include the dedicated auxiliary database. A bare prefix instead uses the
-# ClickHouse user's current database, which may not be the one you granted.
-COLLECTTABLEPREFIX=s3gc.s3gc_example_
+# COLLECTDATABASE names the dedicated auxiliary database. Without it, the
+# auxiliary table lands in the ClickHouse user's current database, which may
+# not be the one you granted.
+COLLECTDATABASE=s3gc
+COLLECTTABLEPREFIX=s3gc_example_
 
 # Object-store scope. S3PATH may be empty for a bucket-root layout.
 S3IP=s3.example.com

deploy/kubernetes/job.yaml.tmpl

@@ -64,6 +64,8 @@ spec:
               value: "${EXPECTED_REPLICAS}"
             - name: S3GC_COLLECTTABLEPREFIX
               value: "${COLLECTTABLEPREFIX}"
+            - name: S3GC_COLLECTDATABASE
+              value: "${COLLECTDATABASE}"
             - name: S3GC_S3IP
               value: "${S3IP}"
             - name: S3GC_S3PORT

deploy/kubernetes/render.py

@@ -50,9 +50,9 @@
 # reject an empty string.
 # RUNLOG defaults on: the durable run-log table is the only record that
 # outlives the Job, whose pod and logs ttlSecondsAfterFinished deletes.
-OPTIONAL = {"USETOTAL": "", "RUNLOG": "true"}
+OPTIONAL = {"USETOTAL": "", "RUNLOG": "true", "COLLECTDATABASE": ""}
 # Environment variables dropped from the manifest when they render empty.
-OPTIONAL_ENV = ("S3GC_USETOTAL",)
+OPTIONAL_ENV = ("S3GC_USETOTAL", "S3GC_COLLECTDATABASE")
 JOB_NAME_RE = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$")
 
 

s3gc.py

@@ -249,6 +249,14 @@ def coerce_bool(value):
     default="s3objects_for_",
     help="prefix for table name to keep data about objects (database is allowed, if not exists, specify --create-database)",
 )
+parser.add_argument(
+    "--collectdatabase",
+    "--collect-database",
+    dest="collectdatabase",
+    default="",
+    help="database for the auxiliary/run-log tables; must match the database "
+    "embedded in --collecttableprefix if both are given",
+)
 parser.add_argument(
     "--collectbatchsize",
     "--collect-batch-size",
@@ -669,23 +677,42 @@ def graceful_exit():
 tname = ""
 dbname = None
 
+# The database can be given two ways: embedded in COLLECTTABLEPREFIX as
+# "db.prefix_", or explicitly via COLLECTDATABASE. Embedding it in one string
+# is easy to omit by accident -- that is what let a production collect Job
+# create its auxiliary table in the wrong database. COLLECTDATABASE makes the
+# qualification an explicit, separate setting; a conflicting pair fails fast
+# instead of silently picking one.
 dbparts = args.collecttableprefix.split(".")
 if len(dbparts) > 2:
     raise ValueError("invalid collecttableprefix")
-elif len(dbparts) == 2:
-    dbname = f"`{dbparts[0]}`"
-    tname = f"{dbname}.`{dbparts[1]}{args.s3diskname}`"
+if len(dbparts) == 2:
+    embedded_db, table_prefix = dbparts
+    if args.collectdatabase and args.collectdatabase != embedded_db:
+        raise ValueError(
+            f"--collectdatabase {args.collectdatabase!r} conflicts with the "
+            f"database embedded in --collecttableprefix {embedded_db!r}"
+        )
+    dbname = f"`{embedded_db}`"
 else:
-    tname = f"`{dbparts[0]}{args.s3diskname}`"
+    table_prefix = dbparts[0]
+    dbname = f"`{args.collectdatabase}`" if args.collectdatabase else None
+
+# Unquoted table name, for system.tables lookups; tname is the quoted SQL form.
+aux_table_name = f"{table_prefix}{args.s3diskname}"
+if dbname:
+    tname = f"{dbname}.`{aux_table_name}`"
+else:
+    tname = f"`{aux_table_name}`"
 
 # The run-log table lives beside the auxiliary table and follows the same naming
 # convention, so one COLLECTTABLEPREFIX still identifies one cleanup. Unlike the
 # auxiliary table it is NEVER truncated: it is the durable record of what the
 # cleanup did after the pod and its logs are gone.
 if dbname:
-    log_tname = f"{dbname}.`{dbparts[1]}{args.s3diskname}_log`"
+    log_tname = f"{dbname}.`{table_prefix}{args.s3diskname}_log`"
 else:
-    log_tname = f"`{dbparts[0]}{args.s3diskname}_log`"
+    log_tname = f"`{table_prefix}{args.s3diskname}_log`"
 
 minio_client = None
 ch_client = None
@@ -1111,7 +1138,6 @@ def do_collect():
     )
 
     if args.createdatabase_flag:
-        parts = args.collecttableprefix.split(".")
         if dbname:
             logger.info(f"creating database {dbname}")
             ch_client.command(f"CREATE DATABASE IF NOT EXISTS {dbname}")
@@ -1199,10 +1225,13 @@ def check_samples_match_partitioning():
     partition pruning: on one production cluster the matching case scanned a
     sample in ~2 min where the mismatching case took ~26 min.
     """
+    database_filter = (
+        f"database = '{dbname.strip('`')}'" if dbname else "database = currentDatabase()"
+    )
     try:
         rows = ch_client.query(
             "SELECT partition_key FROM system.tables "
-            f"WHERE database = currentDatabase() AND name = '{tname.strip('`').split('.')[-1]}'"
+            f"WHERE {database_filter} AND name = '{aux_table_name}'"
         ).result_rows
     except Exception as exc:
         logger.debug(f"could not read partition_key for {tname}: {exc}")

tests/test_s3gc.py

@@ -1701,6 +1701,171 @@ def refuse():
     s3gc_module["do_use"]()
 
 
+# ---------------------------------------------------------------------------
+# Regression tests: explicit --collectdatabase (Magellan HANDOFF.md item #1).
+#
+# A production collect Job failed safely when an unqualified
+# COLLECTTABLEPREFIX created its auxiliary table in the ClickHouse user's
+# current database instead of the intended one. Database qualification was
+# implicit in a single dotted string and easy to omit.
+# ---------------------------------------------------------------------------
+
+
+def test_collectdatabase_qualifies_an_unqualified_prefix(monkeypatch):
+    """--collectdatabase must work without embedding a dot in the prefix."""
+    module = _load_with_env(
+        monkeypatch,
+        S3GC_COLLECTDATABASE="aux_db",
+        S3GC_COLLECTTABLEPREFIX="s3gc_run_",
+    )
+    assert module["dbname"] == "`aux_db`"
+    assert module["tname"] == "`aux_db`.`s3gc_run_s3`"
+    assert module["log_tname"] == "`aux_db`.`s3gc_run_s3_log`"
+
+
+def test_collectdatabase_matching_embedded_database_is_accepted(monkeypatch):
+    """The same database given both ways is not a conflict."""
+    module = _load_with_env(
+        monkeypatch,
+        S3GC_COLLECTDATABASE="aux_db",
+        S3GC_COLLECTTABLEPREFIX="aux_db.s3gc_run_",
+    )
+    assert module["tname"] == "`aux_db`.`s3gc_run_s3`"
+
+
+def test_collectdatabase_conflicting_with_embedded_database_is_rejected(monkeypatch):
+    """A mismatched pair must fail closed rather than silently pick one."""
+    with pytest.raises(ValueError, match="conflicts"):
+        _load_with_env(
+            monkeypatch,
+            S3GC_COLLECTDATABASE="aux_db",
+            S3GC_COLLECTTABLEPREFIX="other_db.s3gc_run_",
+        )
+
+
+def test_embedded_database_alone_still_works(monkeypatch):
+    """Existing deployments that only dot-qualify the prefix are unaffected."""
+    module = _load_with_env(
+        monkeypatch, S3GC_COLLECTTABLEPREFIX="aux_db.s3gc_run_"
+    )
+    assert module["dbname"] == "`aux_db`"
+    assert module["tname"] == "`aux_db`.`s3gc_run_s3`"
+
+
+def test_unqualified_prefix_without_collectdatabase_still_works(monkeypatch):
+    """No database anywhere keeps using the ClickHouse session's current one."""
+    module = _load_with_env(
+        monkeypatch, S3GC_COLLECTTABLEPREFIX="s3gc_run_"
+    )
+    assert module["dbname"] is None
+    assert module["tname"] == "`s3gc_run_s3`"
+
+
+def test_renderer_omits_unset_collectdatabase(tmp_path):
+    """COLLECTDATABASE is optional: an env file without it must keep rendering."""
+    source = (ROOT / "deploy/kubernetes/example.env").read_text()
+    without = "\n".join(
+        line for line in source.splitlines() if not line.startswith("COLLECTDATABASE=")
+    )
+    config_path = tmp_path / "legacy.env"
+    config_path.write_text(without)
+
+    result = subprocess.run(
+        [sys.executable, str(ROOT / "deploy/kubernetes/render.py"), config_path],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert "S3GC_COLLECTDATABASE" not in result.stdout
+
+
+def test_renderer_keeps_configured_collectdatabase(tmp_path):
+    """An explicit auxiliary database must reach the container."""
+    source = (ROOT / "deploy/kubernetes/example.env").read_text()
+    config_path = tmp_path / "qualified.env"
+    config_path.write_text(source.replace("COLLECTDATABASE=s3gc", "COLLECTDATABASE=aux_db"))
+
+    result = subprocess.run(
+        [sys.executable, str(ROOT / "deploy/kubernetes/render.py"), config_path],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert "S3GC_COLLECTDATABASE" in result.stdout
+    assert '"aux_db"' in result.stdout
+
+
+def test_samples_mismatch_check_uses_the_qualified_database(
+    s3gc_module, args_factory, monkeypatch
+):
+    """The partition-key lookup must use dbname, not always currentDatabase().
+
+    check_samples_match_partitioning() previously always queried
+    `WHERE database = currentDatabase()`, so it silently found nothing and
+    skipped the mismatch warning whenever the auxiliary table lived in a
+    database other than the ClickHouse user's current one.
+    """
+    namespace = s3gc_module["check_samples_match_partitioning"].__globals__
+
+    class RecordingCH:
+        def __init__(self):
+            self.queries = []
+
+        def query(self, query):
+            self.queries.append(query)
+            return QueryResult("CRC32(objpath) % 4")
+
+    client = RecordingCH()
+    monkeypatch.setitem(namespace, "args", args_factory(samples=3))
+    monkeypatch.setitem(namespace, "ch_client", client)
+    monkeypatch.setitem(namespace, "tname", "`aux_db`.`aux`")
+    monkeypatch.setitem(namespace, "dbname", "`aux_db`")
+
+    s3gc_module["check_samples_match_partitioning"]()
+
+    assert "currentDatabase()" not in client.queries[0]
+    assert "database = 'aux_db'" in client.queries[0]
+
+
+def test_samples_mismatch_check_matches_a_qualified_table_name(
+    s3gc_module, args_factory, monkeypatch, caplog
+):
+    """The partition-key lookup must use the bare table name.
+
+    Splitting a qualified tname such as `aux_db`.`aux` left a stray backtick
+    on the table component, so system.tables returned no row and the
+    --samples mismatch warning was skipped silently.
+    """
+    namespace = s3gc_module["check_samples_match_partitioning"].__globals__
+
+    class SystemTablesCH:
+        def __init__(self):
+            self.queries = []
+
+        def query(self, query):
+            self.queries.append(query)
+            if "database = 'aux_db' AND name = 'aux'" in query:
+                return QueryResult("CRC32(objpath) % 4")
+            return QueryResult(None)
+
+    client = SystemTablesCH()
+    monkeypatch.setitem(namespace, "args", args_factory(samples=3))
+    monkeypatch.setitem(namespace, "ch_client", client)
+    monkeypatch.setitem(namespace, "tname", "`aux_db`.`aux`")
+    monkeypatch.setitem(namespace, "dbname", "`aux_db`")
+    monkeypatch.setitem(namespace, "aux_table_name", "aux")
+
+    with caplog.at_level("WARNING"):
+        s3gc_module["check_samples_match_partitioning"]()
+
+    assert "name = 'aux'" in client.queries[0]
+    assert "does not match" in caplog.text
+
+
 def test_image_disables_stdout_buffering():
     """A Job killed at activeDeadlineSeconds used to lose its buffered tail.