diff --git a/docs/json-output.md b/docs/json-output.md index f552c529..cffda652 100644 --- a/docs/json-output.md +++ b/docs/json-output.md @@ -9,7 +9,7 @@ promises, and what a script may rely on. |---|---|---| | `version` | — (plain payload, no `status` field) | `version`, `git_sha`, `build_date`, `go_version`, `platform` | | `data ingest` | `succeeded` · `dry-run` · `detached` · `completed_with_failures` · `failed` · `unknown` · `auth_error` · `submit_error` · `watch_error` | result includes the ingest summary (row counts, success rate) when one was produced | -| `data list` | — (a listing, no `status` field) | `namespace`, `release`, `count`, `datasets` | +| `data list` | — (a listing, no `status` field) | `namespace`, `release`, `count`, `datasets` (names, unchanged), `details` (per-dataset objects: `name`, `task` (real ingest task; omitted for datasets ingested before task-persistence), `modality`, `intent`, `records`, `classes`, `format`, `size_bytes`, `ingested`) | | `data delete` | `deleted` · `dry-run` · `declined` | result includes `database`, `table` (the case-resolved spelling), `pvc_paths`, `removed_paths`. Never prompts — pass `--yes` (or `--dry-run`) | Not covered (yet): `doctor`, `resources`, `auth status` — extending @@ -34,7 +34,7 @@ epic's OQ5 decision. `auth status --check` is exit-code-only by design. that need "it actually happened" must check `status`, not just the exit code. 5. **Arrays are never `null`.** Empty lists marshal as `[]` - (`datasets`, `pvc_paths`, `removed_paths`), so indexing is safe. + (`datasets`, `details`, `pvc_paths`, `removed_paths`), so indexing is safe. 6. **`--output-json` implies non-interactive.** Commands never prompt in JSON mode: `data ingest` treats it as `--no-input`; `data delete` requires an explicit `--yes` (or `--dry-run`) and otherwise fails diff --git a/internal/cli/data_list.go b/internal/cli/data_list.go index a0e5953e..31d37bb6 100644 --- a/internal/cli/data_list.go +++ b/internal/cli/data_list.go @@ -6,6 +6,10 @@ import ( "errors" "fmt" "io" + "sort" + "strings" + "time" + "unicode/utf8" "github.com/spf13/cobra" @@ -14,41 +18,47 @@ import ( "github.com/tracebloc/cli/internal/ui" ) -// runDataListArgs is the resolved input to runDataList — same -// shape convention as the other data verbs, keeping the RunE a thin -// flag-to-struct adapter. +// listDatasetsDetailedFn is the test seam over push.ListDatasetsDetailed — +// same fn-var convention as listDatasetsFn / loadClusterFn. +var listDatasetsDetailedFn = push.ListDatasetsDetailed + +// runDataListArgs is the resolved input to runDataList — a thin +// flag-to-struct adapter, same shape as the other data verbs. type runDataListArgs struct { Kubeconfig string Context string Namespace string + ShowAll bool OutputJSON bool Printer *ui.Printer JSONOut io.Writer } -// newDataListCmd implements `tracebloc data list` — a read-only -// listing of the datasets ingested into the cluster. The kubeconfig -// flags are all zero-value-safe, so the minimal `tracebloc data list` -// runs against the current context + its namespace; the flags only -// override that (same convention as `cluster info`). +// newDataListCmd implements `tracebloc data list` — a read-only listing of the +// datasets ingested into the cluster, with per-dataset size, record count, +// format, split, and freshness. The kubeconfig flags are zero-value-safe, so +// the minimal `tracebloc data list` runs against the current context + its +// namespace (same convention as `cluster info`). func newDataListCmd() *cobra.Command { var ( kubeconfigPath string contextOverride string nsOverride string + showAll bool outputJSON bool ) cmd := &cobra.Command{ Use: "list", - Short: "List datasets ingested in the cluster", - Long: `Lists the datasets ingested into your client — -the tables in ` + push.IngestionDatabase + ` on the cluster. + Short: "List datasets ingested in the cluster, with size / records / format", + Long: `Lists the datasets ingested into your client — the tables in ` + push.IngestionDatabase + ` +on the cluster — grouped by modality, with each dataset's split (train/test), +record count, size, format, and when it was ingested. With no flags it uses your current kubeconfig context and its namespace; the flags below override that, same as ` + "`cluster info`" + ` and ` + "`data ingest`" + `. -For the full catalog (with metadata), see the dashboard at -https://ai.tracebloc.io/metadata. +Framework tables (the ingest-run journal) are hidden unless you pass --all. +For the full catalog, see the dashboard at https://ai.tracebloc.io/metadata. Exit codes: 0 listed successfully (including an empty list) @@ -69,6 +79,7 @@ Exit codes: Kubeconfig: kubeconfigPath, Context: contextOverride, Namespace: nsOverride, + ShowAll: showAll, OutputJSON: outputJSON, Printer: printer, JSONOut: jsonOut, @@ -78,20 +89,21 @@ Exit codes: addKubeconfigFlags(cmd, &kubeconfigPath, &contextOverride, kubeconfigFlagUsage, contextFlagUsage) addNamespaceFlag(cmd, &nsOverride, namespaceFlagUsage) + cmd.Flags().BoolVar(&showAll, "all", false, + "include framework/system tables (e.g. the ingest-run journal), normally hidden") cmd.Flags().BoolVar(&outputJSON, "output-json", false, "emit the dataset list as JSON on stdout (human output → stderr)") return cmd } -// runDataList discovers the cluster, enumerates the ingested tables, -// and renders them. Mirrors the other data verbs' discovery so the -// exit-code contract is consistent. +// runDataList discovers the cluster, enumerates the ingested datasets with +// their metadata, and renders them. Mirrors the other data verbs' discovery so +// the exit-code contract is consistent. func runDataList(ctx context.Context, a runDataListArgs) (err error) { // In --output-json mode, guarantee stdout always carries JSON: the - // success path emits the listing and sets jsonEmitted; this defer - // covers the early-failure returns (kubeconfig, no release, query) - // with a JSON error object, mirroring data ingest. (Bugbot #53) + // success path emits the listing; this defer covers the early-failure + // returns with a JSON error object, mirroring data ingest. (Bugbot #53) jsonEmitted := false defer func() { if a.OutputJSON && err != nil && !jsonEmitted { @@ -114,51 +126,422 @@ func runDataList(ctx context.Context, a runDataListArgs) (err error) { } resolved, cs, release := target.Resolved, target.Clientset, target.Release - tables, err := listDatasetsFn(ctx, cs, resolved.RestConfig, resolved.Namespace) + infos, err := listDatasetsDetailedFn(ctx, cs, resolved.RestConfig, resolved.Namespace) if err != nil { return &exitError{code: exitQueryFailed, err: err} } if a.OutputJSON { - writeDataListJSON(a.JSONOut, resolved.Namespace, release.ReleaseName, tables) + writeDataListJSON(a.JSONOut, resolved.Namespace, release.ReleaseName, infos, a.ShowAll) jsonEmitted = true return nil } - renderDataList(p, resolved.Namespace, tables) + renderDataList(p, resolved.Namespace, infos, a.ShowAll) return nil } -// renderDataList prints the human-facing listing. Split out so it's -// unit-testable with a buffer-backed Printer. -func renderDataList(p *ui.Printer, namespace string, tables []string) { - p.Section(fmt.Sprintf("Datasets in %s (%d)", namespace, len(tables))) - if len(tables) == 0 { +// renderDataList prints the human-facing listing: a summary line, then the +// datasets grouped by their real task (or inferred modality when the task +// wasn't recorded) with per-dataset detail. Split out so it's unit-testable +// with a buffer-backed Printer. +func renderDataList(p *ui.Printer, namespace string, infos []push.DatasetInfo, showAll bool) { + var shown, system []push.DatasetInfo + var totalBytes int64 + for _, d := range infos { + if d.System { + system = append(system, d) + continue + } + shown = append(shown, d) + totalBytes += d.SizeBytes + } + + if len(shown) == 0 && !(showAll && len(system) > 0) { + p.Section(fmt.Sprintf("Datasets in %s (0)", namespace)) p.Newline() p.Para(fmt.Sprintf("No datasets yet — ingest one with `%s data ingest`.", invokedName())) + if len(system) > 0 && !showAll { + p.Hintf("%d system table(s) hidden — show with --all.", len(system)) + } return } - for _, t := range tables { - p.Infof("%s", t) + + header := fmt.Sprintf("Datasets in %s — %d", namespace, len(shown)) + if totalBytes > 0 { + header += " · " + push.HumanBytes(totalBytes) + } + p.Section(header) + if len(system) > 0 && !showAll { + p.Hintf("%d system table(s) hidden — show with --all.", len(system)) + } + + // Column widths sized to the actual rows so no cell overflows its slot and + // shifts the columns after it. Measured in display columns (runes), so the + // em dash and middot don't skew the byte-based padding. Record counts and + // sizes vary widely ("100 documents", "100.00 KiB"), so they're sized here + // too rather than pinned to a guessed constant. + nameW, fmtW, recW, sizeW := 8, 10, 6, 4 + groups := map[string][]push.DatasetInfo{} + groupRank := map[string]int{} // group label → modality rank, for ordering + for _, d := range shown { + m := datasetModality(d) + label := groupLabel(d, m) + groups[label] = append(groups[label], d) + groupRank[label] = modalityRank(m) + if l := dispW(d.Name); l > nameW { + nameW = l + } + if l := dispW(formatCell(d, m)); l > fmtW { + fmtW = l + } + if l := dispW(recordsCell(d, m)); l > recW { + recW = l + } + if l := dispW(sizeCell(d)); l > sizeW { + sizeW = l + } + } + // Names are user-controlled and can be arbitrarily long, so cap + truncate + // (below) to keep the table narrow. Format cells are system-generated and + // naturally bounded ("csv · N cols · M classes"), so they're sized to + // content, not capped — capping without truncating would let a wide format + // overflow and shift the freshness column, the very thing sizing prevents. + if nameW > 24 { + nameW = 24 + } + + // Order groups by modality family (Image→Text→Tabular→Time-series→Other), + // then label — so tasks in the same family cluster and the order is stable. + labels := make([]string, 0, len(groups)) + for l := range groups { + labels = append(labels, l) + } + sort.Slice(labels, func(i, j int) bool { + if ri, rj := groupRank[labels[i]], groupRank[labels[j]]; ri != rj { + return ri < rj + } + return labels[i] < labels[j] + }) + for _, label := range labels { + ds := groups[label] + sort.Slice(ds, func(i, j int) bool { return ds[i].Name < ds[j].Name }) + p.Section(fmt.Sprintf("%s · %d", label, len(ds))) + for _, d := range ds { + p.Para(datasetRow(d, datasetModality(d), nameW, recW, sizeW, fmtW)) + } + } + + if showAll && len(system) > 0 { + // The system group is its own sub-table (name + size only), so size its + // two columns to its own rows rather than the shown datasets'. No cap: + // system names are framework-generated and short, and padRight doesn't + // truncate — a cap here would only reintroduce the overflow it implies + // it prevents. + sysNameW, sysSizeW := 8, 4 + for _, d := range system { + if l := dispW(d.Name); l > sysNameW { + sysNameW = l + } + if l := dispW(sizeCell(d)); l > sysSizeW { + sysSizeW = l + } + } + sort.Slice(system, func(i, j int) bool { return system[i].Name < system[j].Name }) + p.Section(fmt.Sprintf("System · %d", len(system))) + for _, d := range system { + p.Para("· " + padRight(d.Name, sysNameW) + " " + padLeft(sizeCell(d), sysSizeW)) + } + } +} + +// datasetRow formats one dataset as an aligned row: status glyph, name, split, +// record count (with the modality's noun), size, format, and freshness. The +// widths are display-column counts sized by the caller to the widest cell, so +// no value overflows its slot and shifts the columns after it. Cells are padded +// here (by rune count) because fmt's %*s pads by bytes — which would misalign +// the multi-byte em dash / middot. +func datasetRow(d push.DatasetInfo, modality string, nameW, recW, sizeW, fmtW int) string { + glyph := "✔" + if d.Records == 0 { + glyph = "⚠" // ingested-but-empty (e.g. an ingest that dropped every record) + } + name := d.Name + if utf8.RuneCountInString(name) > nameW { + name = string([]rune(name)[:nameW-1]) + "…" + } + split := d.Intent + if split == "" { + split = "—" } + return glyph + " " + + padRight(name, nameW) + " " + + padRight(split, 5) + " " + + padRight(recordsCell(d, modality), recW) + " " + + padLeft(sizeCell(d), sizeW) + " " + + padRight(formatCell(d, modality), fmtW) + " " + + relativeTime(d.CreatedUnix) +} + +// sizeCell renders a dataset's size, or an em dash when the du size is unknown +// (jobs-manager unreachable, or a system table that isn't du-sized). +func sizeCell(d push.DatasetInfo) string { + if d.SizeBytes > 0 { + return push.HumanBytes(d.SizeBytes) + } + return "—" +} + +// dispW is a string's width in display columns (runes), not bytes — so the em +// dash and middot (multi-byte, one column each) each count as one. +func dispW(s string) int { return utf8.RuneCountInString(s) } + +// padRight / padLeft pad s to w display columns. fmt's %*s pads by byte length, +// which over-pads multi-byte glyphs; padding by rune count keeps columns aligned. +func padRight(s string, w int) string { + if n := w - utf8.RuneCountInString(s); n > 0 { + return s + strings.Repeat(" ", n) + } + return s +} + +func padLeft(s string, w int) string { + if n := w - utf8.RuneCountInString(s); n > 0 { + return strings.Repeat(" ", n) + s + } + return s +} + +// frameworkCols are the columns the ingestor adds to every dataset table; the +// rest are the user's schema columns (used for the "N cols" format hint and to +// detect real datasets vs framework tables). +var frameworkCols = map[string]bool{ + "id": true, "created_at": true, "updated_at": true, "status": true, + "label": true, "data_intent": true, "data_id": true, "filename": true, + "extension": true, "annotation": true, "ingestor_id": true, +} + +// datasetModality returns the modality family. When the ingest task is known +// (recorded in the run journal) it's taken from the category registry — the +// same source of truth `data ingest` uses (cli#74), so it can't drift as tasks +// are added. Time-series tasks are FamilyTabular there, so a known one reports +// "Tabular"; the "Time-series" bucket below is only the inference fallback for +// datasets ingested before the task was recorded. Falls back to inferring from +// the on-disk shape: the file extension, else time/sequence columns. +func datasetModality(d push.DatasetInfo) string { + switch { + case push.IsImage(d.Task): + return "Image" + case push.IsText(d.Task): + return "Text" + case push.IsTabular(d.Task): + return "Tabular" + } + switch strings.ToLower(d.Extension) { + case "jpg", "jpeg", "png": + return "Image" + case "txt", "text": // the ingestor accepts both .txt and .text + return "Text" + } + if hasCol(d.Columns, "sequence_id") || hasCol(d.Columns, "timestamp") || + (hasCol(d.Columns, "time") && hasCol(d.Columns, "event")) { + return "Time-series" + } + // A populated dataset with user-schema columns is tabular. Require records: + // an empty (0-row) table has NULL extension/label, so its modality is + // genuinely unknowable — it falls to "Other" rather than a wrong guess (an + // empty image/semseg/keypoint table would otherwise look tabular). + if d.Records > 0 && featureColCount(d.Columns) > 0 { + return "Tabular" + } + return "Other" +} + +// groupLabel is the section header a dataset is grouped under: its real task's +// registry label ("Time-series classification", "Sequence-to-sequence") when +// the journal recorded a known task, the raw task id for a task the registry +// doesn't know, else the inferred modality family. Using the registry label +// keeps the header identical to the rest of the CLI rather than re-deriving it. +func groupLabel(d push.DatasetInfo, modality string) string { + if spec, ok := push.Lookup(d.Task); ok { + return spec.Label + } + if d.Task != "" { + return d.Task + } + return modality +} + +// modalityRank orders the modality families so a group's position is stable and +// related tasks cluster together. +func modalityRank(modality string) int { + switch modality { + case "Image": + return 0 + case "Text": + return 1 + case "Tabular": + return 2 + case "Time-series": + return 3 + default: // Other + return 4 + } +} + +// hasCol reports whether cols contains name (case-insensitive, trimmed). +func hasCol(cols []string, name string) bool { + for _, c := range cols { + if strings.EqualFold(strings.TrimSpace(c), name) { + return true + } + } + return false +} + +// featureColCount is the number of user-schema columns (all columns minus the +// framework-managed ones). +func featureColCount(cols []string) int { + n := 0 + for _, c := range cols { + if !frameworkCols[strings.ToLower(strings.TrimSpace(c))] { + n++ + } + } + return n +} + +// recordsCell renders the record count with the modality's natural noun. +func recordsCell(d push.DatasetInfo, modality string) string { + noun := "rows" + switch modality { + case "Image": + noun = "images" + case "Text": + noun = "documents" + } + return fmt.Sprintf("%d %s", d.Records, noun) +} + +// formatCell renders the format hint: the file extension for file-bearing +// tasks, or "csv · N cols" for tabular/time-series, plus "· N classes" when the +// dataset is labelled. +func formatCell(d push.DatasetInfo, modality string) string { + var base string + switch modality { + case "Image", "Text": + // Usually the recorded file extension. When the modality came from the + // task (not the extension), a file dataset whose extension wasn't + // recorded still lands here — fall back to "files" rather than blank. + base = strings.ToLower(d.Extension) + if base == "" { + base = "files" + } + case "Tabular", "Time-series": + base = fmt.Sprintf("csv · %d cols", featureColCount(d.Columns)) + default: + // Undetermined modality. Row-based tasks (with user feature columns) + // resolve to Tabular/Time-series, so a populated table left here is a + // file dataset whose type we couldn't pin down (e.g. extension not + // recorded) — say "files". An empty table is genuinely unknown; don't + // imply "csv". (A `filename` column can't gate this — it's a framework + // column on every table.) + if d.Records > 0 { + return "files" + } + return "—" + } + // Show classes only when the label actually repeats (classes < records): + // a continuous regression target has ~one distinct value per row, which is + // not a class count. COUNT(DISTINCT label) can't tell the two apart, so this + // guard keeps "N classes" to genuinely categorical datasets. + if d.Classes >= 2 && d.Classes < d.Records { + base += fmt.Sprintf(" · %d classes", d.Classes) + } + return base +} + +// relativeTime renders a UTC epoch (the table's create_time via UNIX_TIMESTAMP, +// which is tz-safe regardless of the MySQL session clock) as a coarse "Xh ago". +// Zero/unknown → an em dash; a future timestamp (clock skew) → "just now". +func relativeTime(epoch int64) string { + if epoch <= 0 { + return "—" + } + d := time.Since(time.Unix(epoch, 0)) + switch { + case d < time.Minute: + return "just now" + case d < time.Hour: + return fmt.Sprintf("%dm ago", int(d.Minutes())) + case d < 24*time.Hour: + return fmt.Sprintf("%dh ago", int(d.Hours())) + default: + return fmt.Sprintf("%dd ago", int(d.Hours()/24)) + } +} + +// ingestedISO renders the ingest time as an explicit UTC RFC3339 stamp from the +// same epoch relativeTime uses. Deriving it from the epoch (not a session-tz +// DATE_FORMAT string) keeps JSON consumers and the human "ago" text in +// agreement regardless of the MySQL session timezone. Empty when unknown. +func ingestedISO(epoch int64) string { + if epoch <= 0 { + return "" + } + return time.Unix(epoch, 0).UTC().Format(time.RFC3339) +} + +// ── JSON output (owned by the CLI layer) ── + +type datasetJSON struct { + Name string `json:"name"` + Task string `json:"task,omitempty"` // real ingest task; omitted for pre-persistence datasets + Modality string `json:"modality"` + Intent string `json:"intent,omitempty"` + Records int64 `json:"records"` + Classes int64 `json:"classes,omitempty"` + Format string `json:"format"` + SizeBytes int64 `json:"size_bytes"` + Ingested string `json:"ingested,omitempty"` + System bool `json:"system,omitempty"` } -// dataListJSON is the --output-json shape (owned by the CLI layer). type dataListJSON struct { - Namespace string `json:"namespace"` - Release string `json:"release"` - Count int `json:"count"` - Datasets []string `json:"datasets"` + Namespace string `json:"namespace"` + Release string `json:"release"` + Count int `json:"count"` + Datasets []string `json:"datasets"` // names — type unchanged (additive-only JSON contract) + Details []datasetJSON `json:"details"` // per-dataset metadata added by the rich listing } -func writeDataListJSON(w io.Writer, namespace, release string, tables []string) { - if tables == nil { - tables = []string{} // emit [] not null +func writeDataListJSON(w io.Writer, namespace, release string, infos []push.DatasetInfo, showAll bool) { + names := []string{} + details := []datasetJSON{} + for _, d := range infos { + if d.System && !showAll { + continue + } + m := datasetModality(d) + names = append(names, d.Name) + details = append(details, datasetJSON{ + Name: d.Name, + Task: d.Task, + Modality: m, + Intent: d.Intent, + Records: d.Records, + Classes: d.Classes, + Format: formatCell(d, m), + SizeBytes: d.SizeBytes, + Ingested: ingestedISO(d.CreatedUnix), + System: d.System, + }) } res := dataListJSON{ Namespace: namespace, Release: release, - Count: len(tables), - Datasets: tables, + Count: len(names), + Datasets: names, + Details: details, } b, err := json.MarshalIndent(res, "", " ") if err != nil { @@ -167,9 +550,9 @@ func writeDataListJSON(w io.Writer, namespace, release string, tables []string) _, _ = fmt.Fprintln(w, string(b)) } -// writeDataListErrorJSON emits a minimal JSON error object for -// --output-json runs that fail before the listing is produced, so -// stdout is never empty on failure (parallels data ingest). (Bugbot #53) +// writeDataListErrorJSON emits a minimal JSON error object for --output-json +// runs that fail before the listing is produced, so stdout is never empty on +// failure (parallels data ingest). (Bugbot #53) func writeDataListErrorJSON(w io.Writer, e error, code int) { res := struct { Status string `json:"status"` diff --git a/internal/cli/data_list_test.go b/internal/cli/data_list_test.go index 715f8731..dac1f241 100644 --- a/internal/cli/data_list_test.go +++ b/internal/cli/data_list_test.go @@ -9,14 +9,32 @@ import ( "path/filepath" "strings" "testing" + "time" + "github.com/tracebloc/cli/internal/push" "github.com/tracebloc/cli/internal/ui" ) -// TestRunDataList_OutputJSONEarlyFailureEmitsJSON: with --output-json, -// a failure before the listing (here a broken kubeconfig, exit 3) still -// writes a JSON error object to stdout — the stdout-always-JSON contract -// that #49 established for data ingest. (Bugbot #53) +// sample datasets spanning every modality + a system table. +func sampleInfos() []push.DatasetInfo { + return []push.DatasetInfo{ + {Name: "image_train", Intent: "train", Records: 20, Classes: 2, Extension: "jpg", SizeBytes: 13210, + CreatedUnix: 1721556000, + Columns: []string{"id", "label", "data_intent", "data_id", "filename", "extension"}}, + {Name: "text_test", Intent: "test", Records: 10, Classes: 2, Extension: "txt", SizeBytes: 770, + Columns: []string{"id", "label", "data_intent", "data_id", "filename", "extension"}}, + {Name: "tabular_train", Intent: "train", Records: 20, Classes: 2, Extension: "", SizeBytes: 206, + Columns: []string{"id", "label", "data_intent", "data_id", "age", "income"}}, + {Name: "timeseries_train", Intent: "train", Records: 36, Classes: 2, Extension: "", SizeBytes: 695, + Columns: []string{"id", "label", "data_intent", "data_id", "sequence_id", "timestamp", "hr", "temp"}}, + {Name: "tracebloc_ingest_runs", SizeBytes: 4096, System: true, + Columns: []string{"ingestor_id", "table_name", "registered"}}, + } +} + +// TestRunDataList_OutputJSONEarlyFailureEmitsJSON: with --output-json, a failure +// before the listing (broken kubeconfig, exit 3) still writes a JSON error +// object to stdout — the stdout-always-JSON contract from #49. (Bugbot #53) func TestRunDataList_OutputJSONEarlyFailureEmitsJSON(t *testing.T) { bad := filepath.Join(t.TempDir(), "broken.yaml") if err := os.WriteFile(bad, []byte("}{ not valid kubeconfig"), 0o644); err != nil { @@ -43,11 +61,9 @@ func TestRunDataList_OutputJSONEarlyFailureEmitsJSON(t *testing.T) { } } -// TestRenderDataList_Empty: the empty listing shows the count and -// points the user at `data ingest`. func TestRenderDataList_Empty(t *testing.T) { var buf bytes.Buffer - renderDataList(ui.New(&buf, ui.WithColor(false)), "ap-workspace", nil) + renderDataList(ui.New(&buf, ui.WithColor(false)), "ap-workspace", nil, false) out := buf.String() if !strings.Contains(out, "Datasets in ap-workspace (0)") { t.Errorf("missing header/count:\n%s", out) @@ -57,39 +73,287 @@ func TestRenderDataList_Empty(t *testing.T) { } } -// TestRenderDataList_Items: a populated listing shows the count and -// every table name. -func TestRenderDataList_Items(t *testing.T) { +// TestRenderDataList_RichAndGrouped: datasets group by modality with per-row +// detail; the system table is hidden by default (with a hint). +func TestRenderDataList_RichAndGrouped(t *testing.T) { var buf bytes.Buffer - renderDataList(ui.New(&buf, ui.WithColor(false)), "tracebloc-templates", []string{"reg_train", "churn_test"}) + renderDataList(ui.New(&buf, ui.WithColor(false)), "test0721", sampleInfos(), false) out := buf.String() - for _, want := range []string{"Datasets in tracebloc-templates (2)", "reg_train", "churn_test"} { + + for _, want := range []string{ + "Datasets in test0721 — 4 · ", // 4 shown (system excluded), total size + "Image · 1", "Text · 1", "Tabular · 1", "Time-series · 1", // modality groups + "image_train", "20 images", "12.90 KiB", "jpg · 2 classes", "train", // rich image row + "tabular_train", "20 rows", "csv · 2 cols", // tabular row + feature-col count + "timeseries_train", "36 rows", // time-series grouped by sequence_id/timestamp + "1 system table(s) hidden", // hint + } { if !strings.Contains(out, want) { - t.Errorf("missing %q:\n%s", want, out) + t.Errorf("missing %q in:\n%s", want, out) } } + if strings.Contains(out, "tracebloc_ingest_runs") { + t.Errorf("system table must be hidden by default:\n%s", out) + } +} + +// --all reveals the system table under its own group. +func TestRenderDataList_ShowAll(t *testing.T) { + var buf bytes.Buffer + renderDataList(ui.New(&buf, ui.WithColor(false)), "test0721", sampleInfos(), true) + out := buf.String() + if !strings.Contains(out, "System · 1") || !strings.Contains(out, "tracebloc_ingest_runs") { + t.Errorf("--all should show the system table:\n%s", out) + } } -// TestWriteDataListJSON: valid JSON with the expected fields, and a -// nil dataset slice marshals as [] (not null) so scripts get an array. +// An ingested-but-empty dataset is flagged with ⚠, not ✔. +func TestDatasetRow_EmptyIsWarned(t *testing.T) { + empty := push.DatasetInfo{Name: "objdet_train", Extension: "jpg", Records: 0, Intent: "train", + Columns: []string{"data_id", "filename", "extension"}} + row := datasetRow(empty, datasetModality(empty), 16, 8, 8, 16) + if !strings.HasPrefix(row, "⚠") { + t.Errorf("0-record dataset should lead with ⚠, got: %q", row) + } +} + +// TestRenderDataList_ColumnsAlign: rows with wide values (≥100 KiB sizes, +// ≥100-document counts) must stay column-aligned — the size and format columns +// should start at the same offset on every row. Regression for the fixed-width +// %9s/%-12s overflow (Bugbot, commit 1cfdcc3). +func TestRenderDataList_ColumnsAlign(t *testing.T) { + infos := []push.DatasetInfo{ + // small: "5 documents" / "0.75 KiB" (both short) + {Name: "text_small", Intent: "train", Records: 5, Classes: 2, Extension: "txt", SizeBytes: 770, + Columns: []string{"id", "label", "data_intent", "data_id", "filename", "extension"}}, + // wide: "100000 documents" (16) + "100.00 KiB" (10) — both overflow the + // old fixed %-12s / %9s and would shift later columns without dynamic sizing. + {Name: "text_big", Intent: "test", Records: 100000, Classes: 2, Extension: "txt", SizeBytes: 102400, + Columns: []string{"id", "label", "data_intent", "data_id", "filename", "extension"}}, + } + var buf bytes.Buffer + renderDataList(ui.New(&buf, ui.WithColor(false)), "ns", infos, false) + out := buf.String() + + // Both rows are in the same "Text" group; find them and confirm the format + // token ("txt · 2 classes") begins at the identical column on each. + var offsets []int + for _, ln := range strings.Split(out, "\n") { + if strings.Contains(ln, "text_small") || strings.Contains(ln, "text_big") { + idx := strings.Index(ln, "txt · 2 classes") + if idx < 0 { + t.Fatalf("row missing format cell: %q", ln) + } + offsets = append(offsets, idx) + } + } + if len(offsets) != 2 { + t.Fatalf("want 2 text rows, found %d in:\n%s", len(offsets), out) + } + if offsets[0] != offsets[1] { + t.Errorf("format column misaligned: offsets %v differ — wide values overflowed:\n%s", offsets, out) + } +} + +func TestDatasetModality(t *testing.T) { + cases := []struct { + d push.DatasetInfo + want string + }{ + {push.DatasetInfo{Extension: "jpg"}, "Image"}, + {push.DatasetInfo{Extension: "PNG"}, "Image"}, + {push.DatasetInfo{Extension: "txt"}, "Text"}, + {push.DatasetInfo{Extension: "text"}, "Text"}, // .text is also a text extension + {push.DatasetInfo{Columns: []string{"sequence_id", "timestamp", "hr"}}, "Time-series"}, + {push.DatasetInfo{Columns: []string{"timestamp", "value"}}, "Time-series"}, + {push.DatasetInfo{Columns: []string{"time", "event"}}, "Time-series"}, + {push.DatasetInfo{Columns: []string{"age", "income"}, Records: 3}, "Tabular"}, + // empty (0-row) file dataset: NULL extension, no schema cols → undetermined + {push.DatasetInfo{Records: 0, Columns: []string{"id", "label", "filename", "extension"}}, "Other"}, + // A recorded task is authoritative — taken from the registry, not + // inferred — and wins even over a misleading on-disk shape. + {push.DatasetInfo{Task: "object_detection", Extension: "jpg"}, "Image"}, + {push.DatasetInfo{Task: "embeddings"}, "Text"}, + {push.DatasetInfo{Task: "tabular_regression"}, "Tabular"}, + // time-series tasks are FamilyTabular in the registry → "Tabular". + {push.DatasetInfo{Task: "time_to_event_prediction"}, "Tabular"}, + {push.DatasetInfo{Task: "time_series_classification", Columns: []string{"sequence_id"}}, "Tabular"}, + {push.DatasetInfo{Task: "semantic_segmentation", Records: 5, Columns: []string{"a", "b"}}, "Image"}, + // Unknown task string → fall back to shape inference. + {push.DatasetInfo{Task: "mystery_task", Extension: "jpg"}, "Image"}, + } + for _, c := range cases { + if got := datasetModality(c.d); got != c.want { + t.Errorf("modality(%+v) = %q, want %q", c.d, got, c.want) + } + } +} + +// groupLabel uses the category registry's canonical label for a known task (so +// headers match the rest of the CLI), the raw id for an unknown task, and the +// inferred modality when no task was recorded. +func TestGroupLabel(t *testing.T) { + cases := []struct { + d push.DatasetInfo + modality string + want string + }{ + {push.DatasetInfo{Task: "image_classification"}, "Image", "Image classification"}, + {push.DatasetInfo{Task: "time_series_classification"}, "Tabular", "Time-series classification"}, + {push.DatasetInfo{Task: "seq2seq"}, "Text", "Sequence-to-sequence"}, + {push.DatasetInfo{Task: "mystery_task"}, "Other", "mystery_task"}, // unknown → verbatim + {push.DatasetInfo{Task: ""}, "Tabular", "Tabular"}, // no task → modality + } + for _, c := range cases { + if got := groupLabel(c.d, c.modality); got != c.want { + t.Errorf("groupLabel(task=%q) = %q, want %q", c.d.Task, got, c.want) + } + } +} + +// Datasets with a recorded task group under the registry's task label (not the +// generic modality), ordered by modality family (Image before Tabular family). +func TestRenderDataList_GroupsByTask(t *testing.T) { + infos := []push.DatasetInfo{ + {Name: "sepsis_train", Task: "time_series_classification", Intent: "train", Records: 4000, Classes: 2, SizeBytes: 20480, + Columns: []string{"id", "label", "data_intent", "data_id", "sequence_id", "timestamp", "hr"}}, + {Name: "xray_train", Task: "image_classification", Intent: "train", Records: 50, Classes: 2, Extension: "jpg", SizeBytes: 1048576, + Columns: []string{"id", "label", "data_intent", "data_id", "filename", "extension"}}, + } + var buf bytes.Buffer + renderDataList(ui.New(&buf, ui.WithColor(false)), "ns", infos, false) + out := buf.String() + + for _, want := range []string{ + "Image classification · 1", // registry label, not the bare "Image" + "Time-series classification · 1", // canonical hyphenated label from the registry + "xray_train", "50 images", + "sepsis_train", "4000 rows", // time-series counts rows + } { + if !strings.Contains(out, want) { + t.Errorf("missing %q in:\n%s", want, out) + } + } + if strings.Contains(out, "\nImage · ") || strings.Contains(out, "\nTabular · ") { + t.Errorf("known-task datasets must not fall back to bare modality headers:\n%s", out) + } + if strings.Index(out, "Image classification") > strings.Index(out, "Time-series classification") { + t.Errorf("Image family should sort before the tabular family:\n%s", out) + } +} + +// A task that resolves to Image but whose extension wasn't recorded still reads +// "files" (the branch is reachable via task-derived modality). +func TestFormatCell_TaskImageNoExtension(t *testing.T) { + d := push.DatasetInfo{Task: "image_classification", Records: 10, Extension: "", + Columns: []string{"data_id", "filename"}} + if got := formatCell(d, datasetModality(d)); got != "files" { + t.Errorf("task-image without a recorded extension format = %q, want \"files\"", got) + } +} + +// The real task is surfaced in JSON (`task`), alongside the derived modality. +func TestWriteDataListJSON_IncludesTask(t *testing.T) { + infos := []push.DatasetInfo{ + {Name: "xray_train", Task: "image_classification", Records: 50, Extension: "jpg", CreatedUnix: 1721556000, + Columns: []string{"data_id", "label", "filename", "extension"}}, + } + var buf bytes.Buffer + writeDataListJSON(&buf, "ns", "tracebloc", infos, false) + var got dataListJSON + if err := json.Unmarshal(buf.Bytes(), &got); err != nil { + t.Fatalf("not JSON: %v\n%s", err, buf.String()) + } + if len(got.Details) != 1 || got.Details[0].Task != "image_classification" || + got.Details[0].Modality != "Image" { + t.Errorf("details[].task/modality should carry the real task: %+v", got.Details) + } +} + +// An undetermined ("Other") dataset must not claim a "csv" format. +func TestFormatCell_OtherIsNeutral(t *testing.T) { + d := push.DatasetInfo{Records: 0, Columns: []string{"id", "label", "filename", "extension"}} + if got := formatCell(d, datasetModality(d)); got != "—" { + t.Errorf("undetermined dataset format = %q, want em dash (not csv)", got) + } +} + +// A populated file table whose extension wasn't recorded can't be typed as +// Image vs Text (so it's "Other"), but it's still clearly file-based — format +// should read "files", not "—". Exercises the reachable default-case fallback. +func TestFormatCell_FileWithoutExtension(t *testing.T) { + d := push.DatasetInfo{Records: 12, Extension: "", Columns: []string{"id", "label", "data_id", "filename"}} + if m := datasetModality(d); m != "Other" { + t.Errorf("extension-less file table modality = %q, want Other", m) + } + if got := formatCell(d, datasetModality(d)); got != "files" { + t.Errorf("file-based (has filename, no extension) format = %q, want \"files\"", got) + } +} + +// TestWriteDataListJSON: `datasets` stays a string array (additive contract); +// the rich objects go in the new `details`. System excluded unless --all; an +// empty result marshals both as [] (not null). func TestWriteDataListJSON(t *testing.T) { var buf bytes.Buffer - writeDataListJSON(&buf, "ns1", "tracebloc", []string{"a", "b"}) + writeDataListJSON(&buf, "ns1", "tracebloc", sampleInfos(), false) var got dataListJSON if err := json.Unmarshal(buf.Bytes(), &got); err != nil { t.Fatalf("not JSON: %v\n%s", err, buf.String()) } - if got.Namespace != "ns1" || got.Release != "tracebloc" || got.Count != 2 { - t.Errorf("unexpected: %+v", got) + if got.Namespace != "ns1" || got.Release != "tracebloc" || got.Count != 4 { + t.Errorf("unexpected top-level: %+v", got) } - if len(got.Datasets) != 2 || got.Datasets[0] != "a" { - t.Errorf("datasets wrong: %+v", got.Datasets) + // `datasets` is still a []string of names (contract-preserving). + if len(got.Datasets) != 4 { + t.Errorf("datasets (names) count = %d, want 4", len(got.Datasets)) + } + foundName := false + for _, n := range got.Datasets { + if n == "image_train" { + foundName = true + } + } + if !foundName { + t.Errorf("datasets should list names incl image_train: %v", got.Datasets) + } + // `details` carries the rich objects. + var img *datasetJSON + for i := range got.Details { + if got.Details[i].Name == "image_train" { + img = &got.Details[i] + } + if got.Details[i].System { + t.Errorf("system table must be excluded without --all: %+v", got.Details[i]) + } + } + if img == nil || img.Modality != "Image" || img.Records != 20 || img.Format != "jpg · 2 classes" { + t.Errorf("image dataset detail wrong: %+v", img) + } + // `ingested` must be timezone-explicit (UTC, Z-suffixed RFC3339) so JSON + // consumers can't misread it as a naive local time. (Bugbot: JSON tz.) + if img != nil { + if !strings.HasSuffix(img.Ingested, "Z") { + t.Errorf("ingested %q must be Z-suffixed UTC, not a naive timestamp", img.Ingested) + } + if _, err := time.Parse(time.RFC3339, img.Ingested); err != nil { + t.Errorf("ingested %q is not valid RFC3339: %v", img.Ingested, err) + } + } + + buf.Reset() + writeDataListJSON(&buf, "ns1", "tracebloc", nil, false) + for _, want := range []string{`"datasets": []`, `"details": []`} { + if !strings.Contains(buf.String(), want) { + t.Errorf("nil result should marshal %s:\n%s", want, buf.String()) + } } buf.Reset() - writeDataListJSON(&buf, "ns1", "tracebloc", nil) - if !strings.Contains(buf.String(), `"datasets": []`) { - t.Errorf("nil datasets should marshal as []:\n%s", buf.String()) + writeDataListJSON(&buf, "ns1", "tracebloc", sampleInfos(), true) + _ = json.Unmarshal(buf.Bytes(), &got) + if got.Count != 5 { + t.Errorf("--all JSON should include the system table (count 5), got %d", got.Count) } } diff --git a/internal/push/list_detailed.go b/internal/push/list_detailed.go new file mode 100644 index 00000000..4f5266f8 --- /dev/null +++ b/internal/push/list_detailed.go @@ -0,0 +1,347 @@ +package push + +import ( + "bytes" + "context" + "fmt" + "path" + "regexp" + "strconv" + "strings" + + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" +) + +// DatasetInfo is one dataset's metadata for the rich `data list` view. It is +// assembled from read-only queries against the mysql pod: an information_schema +// pass (name/db-size/create-time/columns), a per-table data pass +// (intent/record-count/classes/extension), and the ingest-run journal (task). +// Everything here already exists in the cluster — no backend round-trip. +type DatasetInfo struct { + Name string // table name = dataset name + Intent string // "train" / "test" / "" (from the data_intent column) + Task string // the ingest task/category (e.g. image_classification) from the run journal; "" if not recorded (pre-persistence datasets) + Records int64 // COUNT(*) — images / documents / rows, per modality + Classes int64 // COUNT(DISTINCT label); 0 when unlabelled + Extension string // per-row file extension (jpg/png/txt); "" for CSV tasks + SizeBytes int64 // dataset size: du of the shared PVC for file datasets, else the DB data_length; 0 if unavailable + DBBytes int64 // information_schema.data_length — the size source for row-based (non-file) datasets + CreatedUnix int64 // create_time as a UTC epoch (tz-safe) — the sole time SoT, for both "ago" and JSON + Columns []string // all column names — drives modality inference + System bool // a framework table (no data_id), e.g. the ingest-run journal +} + +// identRe guards table names before they are interpolated into the data query. +// Ingest already restricts dataset names to this shape; the guard is defence in +// depth so a surprising name can never break out of the SQL. +var identRe = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`) + +// ListDatasetsDetailed returns per-dataset metadata for the rich listing, +// reusing the same mysql-pod exec seam + pod discovery as ListDatasets. +func ListDatasetsDetailed(ctx context.Context, cs kubernetes.Interface, cfg *rest.Config, namespace string) ([]DatasetInfo, error) { + mysqlPod, mysqlContainer, err := findRunningPod(ctx, cs, namespace, "mysql") + if err != nil { + return nil, fmt.Errorf("locating mysql pod: %w", err) + } + exec := &SPDYExecutor{Config: cfg, Client: cs} + infos, err := listDatasetsDetailedWith(ctx, exec, namespace, mysqlPod, mysqlContainer) + if err != nil { + return nil, err + } + applyDatasetSizes(infos, datasetSizesFromShared(ctx, exec, cs, namespace)) + return infos, nil +} + +// applyDatasetSizes picks each dataset's size from where its data actually +// lives. File-bearing datasets keep their bytes on the shared PVC, so the real +// size is the du of that PVC (duSizes) — the DB holds only metadata rows. +// Row-based datasets (tabular / time-series) live entirely in the table, so +// their information_schema.data_length (DBBytes) IS the size. +// +// The file-vs-row signal is the per-row file Extension, NOT a `filename` +// column: filename/extension/annotation are framework columns present on EVERY +// dataset table (tabular included), so only a non-empty extension actually +// marks staged files. du is best-effort: a file dataset with no du entry +// (jobs-manager unreachable) shows no size rather than the misleading +// metadata-row DBBytes. +// +// DBBytes is used only for row-based datasets that actually have rows. An empty +// table (0 records) — whether a row-based or a file dataset whose ingest landed +// nothing — has an empty extension too, but its data_length is just InnoDB's +// one-page allocation, not real data; leave it sizeless (rendered "—", matching +// its ⚠ empty flag) rather than implying it holds a page of data. +func applyDatasetSizes(infos []DatasetInfo, duSizes map[string]int64) { + for i := range infos { + // The Extension gate runs FIRST: a row-based table can still have a dest + // dir on the shared PVC (ingest scaffolding), and that stray du entry must + // not override its real DBBytes. + if infos[i].Extension != "" { + if b, ok := duSizes[infos[i].Name]; ok { + infos[i].SizeBytes = b // file dataset: real PVC size + } + } else if infos[i].Records > 0 { + infos[i].SizeBytes = infos[i].DBBytes // row-based with rows: DB data_length + } + } +} + +// datasetSizesFromShared returns real dataset byte sizes by du-ing the shared +// PVC on the jobs-manager pod (which mounts it). Best-effort: any failure +// yields a nil map and the caller renders without sizes. +func datasetSizesFromShared(ctx context.Context, exec Executor, cs kubernetes.Interface, namespace string) map[string]int64 { + pod, container, err := findRunningPod(ctx, cs, namespace, "jobs-manager") + if err != nil { + return nil + } + var stdout, stderr bytes.Buffer + // `|| true`: du exits non-zero if ANY entry is unreadable (jobs-manager can + // hit EACCES on some shared-PVC paths), but the readable entries are already + // on stdout — keep them rather than blanking every dataset's size. + if err := exec.Exec(ctx, namespace, pod, container, + []string{"sh", "-c", "du -sk " + SharedRoot + "/* 2>/dev/null || true"}, + nil, &stdout, &stderr); err != nil { + return nil + } + return parseDuOutput(stdout.String()) +} + +// parseDuOutput parses `du -sk` output ("\t" per line) into a +// name→bytes map keyed by the path's basename (the dataset name). +func parseDuOutput(raw string) map[string]int64 { + m := map[string]int64{} + for _, line := range strings.Split(raw, "\n") { + fields := strings.Fields(line) // " "; dataset dirs have no spaces + if len(fields) < 2 { + continue + } + kb, err := strconv.ParseInt(fields[0], 10, 64) + if err != nil { + continue + } + m[path.Base(fields[len(fields)-1])] = kb * 1024 + } + return m +} + +// listDatasetsDetailedWith runs the two queries through the given Executor so +// the exec + parse path is unit-testable with a fake Executor. A dataset table +// can be dropped (a concurrent `data delete`) between the schema snapshot and +// the data pass, which would fail the whole UNION ALL; retry the pair once — a +// fresh snapshot drops the vanished table and the rebuilt data query succeeds. A +// genuinely unreachable mysql fails the retry too and the error surfaces. +func listDatasetsDetailedWith(ctx context.Context, exec Executor, namespace, pod, container string) ([]DatasetInfo, error) { + infos, err := queryDatasetsDetailed(ctx, exec, namespace, pod, container) + if err != nil { + infos, err = queryDatasetsDetailed(ctx, exec, namespace, pod, container) + } + return infos, err +} + +// queryDatasetsDetailed runs the schema pass then the per-table data pass once. +func queryDatasetsDetailed(ctx context.Context, exec Executor, namespace, pod, container string) ([]DatasetInfo, error) { + // data_length is fetched but used ONLY for row-based datasets (tabular / + // time-series), whose data really lives in the table. For a file-bearing + // dataset it's just the metadata rows (images/text live on the PVC) and for + // a tiny table it's InnoDB's padded page allocation — both misleading — so + // those fall back to a `du` of the PVC in ListDatasetsDetailed. + // Raise group_concat_max_len (default 1024) so a wide table's column list + // isn't truncated mid-name — that would under-count feature columns and drop + // late modality markers. Runs as a leading statement over the same stdin. + schemaQ := "SET SESSION group_concat_max_len = 1048576; " + + "SELECT t.table_name," + + " COALESCE(MAX(UNIX_TIMESTAMP(t.create_time)),0)," + // tz-safe epoch — sole time SoT + " COALESCE(MAX(t.data_length),0)," + // DB size — used for row-based datasets + " COALESCE(GROUP_CONCAT(c.column_name ORDER BY c.ordinal_position SEPARATOR ','),'')" + + " FROM information_schema.tables t" + + " LEFT JOIN information_schema.columns c" + + " ON c.table_schema=t.table_schema AND c.table_name=t.table_name" + + " WHERE t.table_schema='" + IngestionDatabase + "'" + + " GROUP BY t.table_name ORDER BY t.table_name" + schemaOut, err := runMySQLQuery(ctx, exec, namespace, pod, container, schemaQ) + if err != nil { + return nil, err + } + infos := parseSchemaRows(schemaOut) + if len(infos) == 0 { + return infos, nil + } + + // Second pass: intent/count/classes/extension for the REAL dataset tables + // (those with a data_id column). System tables (the run journal, salt store) + // lack data_id and carry no ingest metadata, so they're marked System and + // excluded from the data query — selecting data_intent there would error. + var selects []string + for i := range infos { + d := &infos[i] + // A framework table — the ingest-run journal / ingest-meta store + // (reservedTables, shared with ListDatasets) or any table lacking a + // data_id column — carries no dataset rows. Mark it System so it's + // hidden by default and kept out of the data query below, which selects + // data_intent (a column these tables don't have). + if _, reserved := reservedTables[d.Name]; reserved || !hasColumn(d.Columns, "data_id") { + d.System = true + continue + } + if !identRe.MatchString(d.Name) { + continue // never interpolate a non-identifier table name + } + // Reference only columns this table actually has. `label` is optional — + // self-supervised tasks (MLM/CLM/seq2seq/embeddings) omit it — and + // data_intent/extension can be absent on older tables. A missing column + // would fail the whole UNION ALL and take the entire listing down (exit + // 7), so fall back to a constant when a column isn't present. + intentExpr, classesExpr, extExpr := "''", "0", "''" + if hasColumn(d.Columns, "data_intent") { + intentExpr = "COALESCE(MAX(data_intent),'')" + } + if hasColumn(d.Columns, "label") { + classesExpr = "COUNT(DISTINCT label)" + } + if hasColumn(d.Columns, "extension") { + extExpr = "COALESCE(MAX(extension),'')" + } + // Qualify the table with the schema: mysql runs without -D (so an empty + // cluster with no ingestion DB still lists cleanly), and a bare table + // reference would fail with "No database selected". + selects = append(selects, fmt.Sprintf( + "SELECT '%s',%s,COUNT(*),%s,%s FROM `%s`.`%s`", + d.Name, intentExpr, classesExpr, extExpr, IngestionDatabase, d.Name)) + } + if len(selects) > 0 { + dataOut, err := runMySQLQuery(ctx, exec, namespace, pod, container, strings.Join(selects, " UNION ALL ")) + if err != nil { + return nil, err + } + applyDataRows(infos, dataOut) + } + applyTaskLabels(ctx, exec, namespace, pod, container, infos) + return infos, nil +} + +// applyTaskLabels tags each dataset with its ingest task from the run journal +// (tracebloc_ingest_runs.task, persisted by data-ingestors). Best-effort +// enrichment: a cluster whose ingestor predates the task column, or a dataset +// ingested before it shipped, keeps an empty Task and the caller falls back to +// inferred modality. Guarded on the column's presence — known from the schema +// pass — so a pre-persistence journal is never queried for a column it lacks. +func applyTaskLabels(ctx context.Context, exec Executor, namespace, pod, container string, infos []DatasetInfo) { + hasTaskColumn := false + for i := range infos { + if infos[i].Name == ingestRunsTable && hasColumn(infos[i].Columns, "task") { + hasTaskColumn = true + break + } + } + if !hasTaskColumn { + return + } + out, err := runMySQLQuery(ctx, exec, namespace, pod, container, fmt.Sprintf( + "SELECT table_name, COALESCE(MAX(task),'') FROM `%s`.`%s` "+ + "WHERE task IS NOT NULL GROUP BY table_name", + IngestionDatabase, ingestRunsTable)) + if err != nil { + return // best-effort: the listing still renders with inferred modality + } + tasks := parseTaskRows(out) + for i := range infos { + if t, ok := tasks[infos[i].Name]; ok { + infos[i].Task = t + } + } +} + +// parseTaskRows parses the "table_name\ttask" TSV into a name→task map, skipping +// blank tasks (a run journalled by a pre-persistence ingestor). +func parseTaskRows(raw string) map[string]string { + m := map[string]string{} + for _, line := range strings.Split(raw, "\n") { + line = strings.TrimRight(line, "\r") + if strings.TrimSpace(line) == "" { + continue + } + f := strings.Split(line, "\t") + if len(f) < 2 { + continue + } + if task := strings.TrimSpace(f[1]); task != "" { + m[strings.TrimSpace(f[0])] = task + } + } + return m +} + +// runMySQLQuery feeds the query to `mysql -N` over stdin (not -e) so table names +// and string literals — including backtick-quoted identifiers — never pass +// through the shell, sidestepping quoting/injection entirely. +func runMySQLQuery(ctx context.Context, exec Executor, namespace, pod, container, query string) (string, error) { + var stdout, stderr bytes.Buffer + script := `mysql -uroot -p"$MYSQL_ROOT_PASSWORD" -N` + if err := exec.Exec(ctx, namespace, pod, container, + []string{"sh", "-c", script}, strings.NewReader(query), &stdout, &stderr); err != nil { + return "", fmt.Errorf("querying datasets: %w%s", err, stderrSuffix(&stderr)) + } + return stdout.String(), nil +} + +// parseSchemaRows turns the `mysql -N` TSV of the schema query into DatasetInfos +// (name, create-time epoch, DB size, columns). Malformed/short lines are skipped. +func parseSchemaRows(raw string) []DatasetInfo { + var out []DatasetInfo + for _, line := range strings.Split(raw, "\n") { + line = strings.TrimRight(line, "\r") + if strings.TrimSpace(line) == "" { + continue + } + f := strings.Split(line, "\t") + if len(f) < 4 { + continue + } + d := DatasetInfo{Name: strings.TrimSpace(f[0])} + d.CreatedUnix, _ = strconv.ParseInt(strings.TrimSpace(f[1]), 10, 64) + d.DBBytes, _ = strconv.ParseInt(strings.TrimSpace(f[2]), 10, 64) + if cols := strings.TrimSpace(f[3]); cols != "" { + d.Columns = strings.Split(cols, ",") + } + out = append(out, d) + } + return out +} + +// applyDataRows merges the data query's TSV (name, intent, count, classes, ext) +// back onto the matching DatasetInfo by name. +func applyDataRows(infos []DatasetInfo, raw string) { + by := make(map[string]*DatasetInfo, len(infos)) + for i := range infos { + by[infos[i].Name] = &infos[i] + } + for _, line := range strings.Split(raw, "\n") { + if strings.TrimSpace(line) == "" { + continue + } + f := strings.Split(line, "\t") + if len(f) < 5 { + continue + } + d := by[strings.TrimSpace(f[0])] + if d == nil { + continue + } + d.Intent = strings.TrimSpace(f[1]) + d.Records, _ = strconv.ParseInt(strings.TrimSpace(f[2]), 10, 64) + d.Classes, _ = strconv.ParseInt(strings.TrimSpace(f[3]), 10, 64) + // The ingestor stores the extension with a leading dot (".jpg"); drop it + // so callers can match/display a bare "jpg". + d.Extension = strings.TrimPrefix(strings.TrimSpace(f[4]), ".") + } +} + +// hasColumn reports whether cols contains name (case-insensitive). +func hasColumn(cols []string, name string) bool { + for _, c := range cols { + if strings.EqualFold(strings.TrimSpace(c), name) { + return true + } + } + return false +} diff --git a/internal/push/list_detailed_test.go b/internal/push/list_detailed_test.go new file mode 100644 index 00000000..79b5b9e9 --- /dev/null +++ b/internal/push/list_detailed_test.go @@ -0,0 +1,268 @@ +package push + +import ( + "context" + "fmt" + "io" + "strings" + "testing" +) + +// seqExecutor is an Executor that returns queued stdout per call and captures +// the SQL each call received on stdin — enough to drive the two-query +// listDatasetsDetailedWith path. Calls whose index is in errCalls return an +// error instead, to exercise the retry-on-vanished-table path. +type seqExecutor struct { + outs [][]byte // stdout for call 0, 1, … + queries []string // stdin (the SQL) captured per call + errCalls map[int]bool // 0-based call indices that should fail + call int +} + +func (e *seqExecutor) Exec(_ context.Context, _, _, _ string, _ []string, + stdin io.Reader, stdout, _ io.Writer) error { + idx := e.call + e.call++ + if stdin != nil { + b, _ := io.ReadAll(stdin) + e.queries = append(e.queries, string(b)) + } + if e.errCalls[idx] { + return fmt.Errorf("simulated exec failure on call %d", idx) + } + if idx < len(e.outs) && stdout != nil { + _, _ = stdout.Write(e.outs[idx]) + } + return nil +} + +func TestParseSchemaRows(t *testing.T) { + raw := "image_train\t1721556000\t40960\tid,label,data_id,extension\n" + + "empty_cols\t0\t0\t\n" + + " \n" // blank line skipped + got := parseSchemaRows(raw) + if len(got) != 2 { + t.Fatalf("want 2 rows, got %d: %#v", len(got), got) + } + d := got[0] + if d.Name != "image_train" || d.CreatedUnix != 1721556000 || d.DBBytes != 40960 { + t.Errorf("row0 wrong: %+v", d) + } + if len(d.Columns) != 4 || d.Columns[0] != "id" { + t.Errorf("row0 columns wrong: %#v", d.Columns) + } + if got[1].Name != "empty_cols" || got[1].DBBytes != 0 || len(got[1].Columns) != 0 { + t.Errorf("row1 (no columns) wrong: %+v", got[1]) + } +} + +func TestApplyDataRows(t *testing.T) { + infos := []DatasetInfo{{Name: "a"}, {Name: "b"}} + applyDataRows(infos, "a\ttrain\t20\t2\tjpg\nb\ttest\t5\t0\t\n") + if infos[0].Intent != "train" || infos[0].Records != 20 || infos[0].Classes != 2 || infos[0].Extension != "jpg" { + t.Errorf("a wrong: %+v", infos[0]) + } + if infos[1].Intent != "test" || infos[1].Records != 5 || infos[1].Classes != 0 || infos[1].Extension != "" { + t.Errorf("b wrong: %+v", infos[1]) + } +} + +func TestParseDuOutput(t *testing.T) { + // `du -sk` output: "\t", keyed by basename → bytes. + raw := "80\t/data/shared/image_train\n" + + "8\t/data/shared/clm_test\n" + + "40\t/data/shared/.tracebloc-staging\n" + + "junk line\n" + m := parseDuOutput(raw) + if m["image_train"] != 80*1024 { + t.Errorf("image_train = %d, want %d", m["image_train"], 80*1024) + } + if m["clm_test"] != 8*1024 { + t.Errorf("clm_test = %d, want %d", m["clm_test"], 8*1024) + } + if _, ok := m["junk"]; ok { + t.Errorf("non-numeric line should be skipped, got %v", m) + } +} + +// End-to-end over the two-query path: a real dataset (has data_id) is merged +// with its data row; a system table (no data_id) is marked System and excluded +// from the second (data) query, whose SQL selects data_intent — which a system +// table lacks. +func TestListDatasetsDetailedWith(t *testing.T) { + schema := "image_train\t1721556000\t131072\tid,label,data_intent,data_id,filename,extension\n" + + "tracebloc_ingest_runs\t1721552400\t16384\tingestor_id,table_name,registered\n" + data := "image_train\ttrain\t20\t2\t.jpg\n" // leading dot must be stripped + fe := &seqExecutor{outs: [][]byte{[]byte(schema), []byte(data)}} + + infos, err := listDatasetsDetailedWith(context.Background(), fe, "tracebloc", "mysql-0", "mysql") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(infos) != 2 { + t.Fatalf("want 2 infos, got %d: %#v", len(infos), infos) + } + + var img, sys *DatasetInfo + for i := range infos { + switch infos[i].Name { + case "image_train": + img = &infos[i] + case "tracebloc_ingest_runs": + sys = &infos[i] + } + } + if img == nil || img.System || img.Records != 20 || img.Classes != 2 || img.Intent != "train" || + img.Extension != "jpg" { + t.Errorf("image_train merged wrong: %+v", img) + } + if sys == nil || !sys.System || sys.Records != 0 { + t.Errorf("system table should be flagged System with no data: %+v", sys) + } + + // First query is the schema pass; second is the data pass over real tables only. + if len(fe.queries) != 2 { + t.Fatalf("want 2 queries, got %d", len(fe.queries)) + } + if !strings.Contains(fe.queries[0], "information_schema") { + t.Errorf("query 0 should hit information_schema: %s", fe.queries[0]) + } + if !strings.Contains(fe.queries[1], "image_train") { + t.Errorf("data query should include the real dataset: %s", fe.queries[1]) + } + if strings.Contains(fe.queries[1], "tracebloc_ingest_runs") { + t.Errorf("data query must exclude the system table (no data_id): %s", fe.queries[1]) + } +} + +// A table dropped between the schema snapshot and the data pass fails the whole +// UNION ALL; listDatasetsDetailedWith retries with a fresh snapshot, which no +// longer lists the vanished table, and succeeds. +func TestListDatasetsDetailedWith_RetriesOnVanishedTable(t *testing.T) { + // Attempt 1 lists a + b; its data query (call 1) fails as if b was just + // dropped. Attempt 2's fresh snapshot lists only a, and its data succeeds. + schema1 := "a\t1721556000\t8192\tdata_id,label,data_intent\n" + + "b\t1721556000\t8192\tdata_id,label,data_intent\n" + schema2 := "a\t1721556000\t8192\tdata_id,label,data_intent\n" + data2 := "a\ttrain\t7\t2\t\n" + fe := &seqExecutor{ + outs: [][]byte{[]byte(schema1), nil, []byte(schema2), []byte(data2)}, + errCalls: map[int]bool{1: true}, // the first data query fails + } + + infos, err := listDatasetsDetailedWith(context.Background(), fe, "ns", "mysql-0", "mysql") + if err != nil { + t.Fatalf("expected retry to succeed after a vanished table, got: %v", err) + } + if len(infos) != 1 || infos[0].Name != "a" || infos[0].Records != 7 { + t.Fatalf("want dataset a with 7 records after retry, got: %#v", infos) + } + if fe.call != 4 { + t.Errorf("want 4 execs (schema, failed data, schema, data), got %d", fe.call) + } +} + +func TestApplyDatasetSizes(t *testing.T) { + // filename/extension are framework columns on EVERY table, so the file-vs-row + // signal is the per-row Extension: set → file dataset (PVC/du), empty → + // row-based (DB data_length). + infos := []DatasetInfo{ + {Name: "img_train", Extension: "jpg", Records: 20, DBBytes: 4096, + Columns: []string{"data_id", "filename", "extension"}}, + {Name: "tab_train", Extension: "", Records: 20, DBBytes: 24576, + Columns: []string{"data_id", "filename", "extension", "age", "income"}}, // has filename col yet is row-based + {Name: "img_nodu", Extension: "jpg", Records: 20, DBBytes: 4096, + Columns: []string{"data_id", "filename", "extension"}}, + {Name: "empty_ds", Extension: "", Records: 0, DBBytes: 16384, + Columns: []string{"data_id", "filename", "extension"}}, // 0 rows → sizeless, not the page allocation + {Name: "tab_dir", Extension: "", Records: 20, DBBytes: 24576, + Columns: []string{"data_id", "filename", "extension", "age"}}, // row-based WITH a stray PVC dir + } + applyDatasetSizes(infos, map[string]int64{"img_train": 1048576, "tab_dir": 999999}) + if infos[0].SizeBytes != 1048576 { + t.Errorf("file dataset should take the du size, got %d", infos[0].SizeBytes) + } + if infos[1].SizeBytes != 24576 { + t.Errorf("row-based dataset (empty extension) should take DBBytes despite the framework filename column, got %d", infos[1].SizeBytes) + } + if infos[2].SizeBytes != 0 { + t.Errorf("file dataset without a du entry must stay 0 (—), not the misleading DBBytes, got %d", infos[2].SizeBytes) + } + if infos[3].SizeBytes != 0 { + t.Errorf("empty dataset (0 rows) must stay 0 (—), not the InnoDB page size, got %d", infos[3].SizeBytes) + } + if infos[4].SizeBytes != 24576 { + t.Errorf("row-based dataset with a stray PVC dir must keep DBBytes, not the du size, got %d", infos[4].SizeBytes) + } +} + +func TestParseTaskRows(t *testing.T) { + raw := "xray_train\timage_classification\n" + + "vitals_train\ttime_series_classification\n" + + "blank_task\t\n" + // blank task skipped (pre-persistence run) + "short\n" // malformed line skipped + m := parseTaskRows(raw) + if m["xray_train"] != "image_classification" || + m["vitals_train"] != "time_series_classification" { + t.Errorf("task map wrong: %#v", m) + } + if _, ok := m["blank_task"]; ok { + t.Errorf("blank task should be skipped: %#v", m) + } + if _, ok := m["short"]; ok { + t.Errorf("malformed line should be skipped: %#v", m) + } +} + +// When the runs journal carries a task column, a third query maps each dataset +// to its real task and tags the DatasetInfo. +func TestListDatasetsDetailedWith_AppliesTask(t *testing.T) { + schema := "xray_train\t1721556000\t131072\tdata_id,label,data_intent,filename,extension\n" + + "tracebloc_ingest_runs\t1721552400\t16384\tingestor_id,table_name,registered,task\n" + data := "xray_train\ttrain\t50\t2\t.jpg\n" + taskMap := "xray_train\timage_classification\n" + fe := &seqExecutor{outs: [][]byte{[]byte(schema), []byte(data), []byte(taskMap)}} + + infos, err := listDatasetsDetailedWith(context.Background(), fe, "ns", "mysql-0", "mysql") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + var x *DatasetInfo + for i := range infos { + if infos[i].Name == "xray_train" { + x = &infos[i] + } + } + if x == nil || x.Task != "image_classification" { + t.Fatalf("want xray_train tagged image_classification, got: %#v", x) + } + if len(fe.queries) != 3 { + t.Fatalf("want 3 queries (schema, data, task), got %d", len(fe.queries)) + } + if !strings.Contains(fe.queries[2], "tracebloc_ingest_runs") || + !strings.Contains(fe.queries[2], "task") { + t.Errorf("third query should read task from the runs journal: %s", fe.queries[2]) + } +} + +// A journal predating the task column (no task in its schema) is never queried +// for it — the datasets keep an empty Task and the caller infers modality. +func TestListDatasetsDetailedWith_SkipsTaskWhenColumnAbsent(t *testing.T) { + schema := "xray_train\t1721556000\t131072\tdata_id,label,data_intent,filename,extension\n" + + "tracebloc_ingest_runs\t1721552400\t16384\tingestor_id,table_name,registered\n" + data := "xray_train\ttrain\t50\t2\t.jpg\n" + fe := &seqExecutor{outs: [][]byte{[]byte(schema), []byte(data)}} + + infos, err := listDatasetsDetailedWith(context.Background(), fe, "ns", "mysql-0", "mysql") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + for i := range infos { + if infos[i].Task != "" { + t.Errorf("no task column → Task must stay empty, got %q", infos[i].Task) + } + } + if len(fe.queries) != 2 { + t.Errorf("task lookup must be skipped when the column is absent; got %d queries", len(fe.queries)) + } +}