Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 73 additions & 0 deletions internal/cli/coverage_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -367,3 +367,76 @@ func TestRunClusterInfo_BadKubeconfigExitsThree(t *testing.T) {
t.Errorf("exit code = %d, want 3 (kubeconfig/local-input error)", ee.Code())
}
}

// ---------------------------------------------------------------------------
// backend#2895 — --output-json must name the table that was CREATED.
//
// `spec["table"]` is the operator's --name. Under per-ingestion tables the
// physical table is a `ds_<hex>` handle, so echoing the label emitted
// `"table": "<label>"` for a table that does not exist — and feeding that value
// to `data delete` failed on a dataset just created. The delete path already
// reports "the REAL (case-resolved) spelling, not the raw argument"; ingest
// now holds itself to the same standard.
// ---------------------------------------------------------------------------

func TestWritePushJSON_ReportsPhysicalTableWhenIngestorNamesIt(t *testing.T) {
spec := map[string]any{"table": "dropcheck_train", "category": "tabular_classification", "intent": "train"}
s := &submit.Summary{
IngestorID: "46ad219b-8192-4be3-a0eb-3d9120fa10b4",
DestinationTable: "ds_46ad219b81924be3a0eb3d9120fa10b4",
TotalRecords: 30000, InsertedRecords: 30000, APISentRecords: 30000,
}

var buf bytes.Buffer
writePushJSON(&buf, "succeeded", spec, s, "ns1", "ingest-job-x")

var got pushJSONResult
if err := json.Unmarshal(buf.Bytes(), &got); err != nil {
t.Fatalf("output is not valid JSON: %v\n%s", err, buf.String())
}
if got.Table != "ds_46ad219b81924be3a0eb3d9120fa10b4" {
t.Errorf("Table = %q, want the physical handle the ingestor reported", got.Table)
}
// The whole point: the emitted value must not be the label, because the
// label is not addressable by `data delete` / `data list`.
if got.Table == "dropcheck_train" {
t.Error("Table echoed the operator's --name, which names no table that exists")
}
}

func TestWritePushJSON_FallsBackToRequestedNameWhenUnreported(t *testing.T) {
// Older ingestor, or a run that failed before the banner: the field is
// empty and output must stay byte-identical to today's rather than
// emitting an empty table.
spec := map[string]any{"table": "reg_train", "category": "tabular_regression", "intent": "train"}
s := &submit.Summary{IngestorID: "run-1", TotalRecords: 240, InsertedRecords: 240}

var buf bytes.Buffer
writePushJSON(&buf, "succeeded", spec, s, "ns1", "ingest-job-x")

var got pushJSONResult
if err := json.Unmarshal(buf.Bytes(), &got); err != nil {
t.Fatalf("output is not valid JSON: %v\n%s", err, buf.String())
}
if got.Table != "reg_train" {
t.Errorf("Table = %q, want the requested name as fallback", got.Table)
}
}

func TestWritePushJSON_FallsBackWhenThereIsNoSummaryAtAll(t *testing.T) {
// A submit/auth failure emits JSON with a nil summary. Dereferencing it to
// read the table would panic on the error path — the one place output
// matters most.
spec := map[string]any{"table": "reg_train", "category": "tabular_regression", "intent": "train"}

var buf bytes.Buffer
writePushJSON(&buf, "submit_error", spec, nil, "ns1", "")

var got pushJSONResult
if err := json.Unmarshal(buf.Bytes(), &got); err != nil {
t.Fatalf("output is not valid JSON: %v\n%s", err, buf.String())
}
if got.Table != "reg_train" {
t.Errorf("Table = %q, want the requested name when no summary exists", got.Table)
}
}
15 changes: 14 additions & 1 deletion internal/cli/data_ingest_output.go
Original file line number Diff line number Diff line change
Expand Up @@ -89,9 +89,22 @@ type pushJSONSummary struct {
// --output-json mode). Errors are dropped: marshaling our own struct
// can't fail in practice, and the exit code remains the contract.
func writePushJSON(w io.Writer, status string, spec map[string]any, s *submit.Summary, ns, jobName string) {
// THE TABLE THAT WAS ACTUALLY CREATED, not the one that was asked for
// (tracebloc/backend#2895). `spec["table"]` is the operator's --name; under
// the cluster's per-ingestion-tables mode the physical table is a `ds_<hex>`
// handle, so echoing the label named a table that does not exist — and
// `data delete <that value>` then failed on a dataset just created.
//
// The ingestor reports the real name in its banner; prefer it, and fall back
// to the requested name when it is absent (older ingestor, or a run that
// failed before the banner) so output stays byte-identical there.
table := fmt.Sprintf("%v", spec["table"])
if s != nil && s.DestinationTable != "" {
table = s.DestinationTable
}
res := pushJSONResult{
Status: status,
Table: fmt.Sprintf("%v", spec["table"]),
Table: table,
Category: fmt.Sprintf("%v", spec["category"]),
Intent: fmt.Sprintf("%v", spec["intent"]),
Namespace: ns,
Expand Down
1 change: 1 addition & 0 deletions internal/cli/testdata/golden/zz-all-strings.golden
Original file line number Diff line number Diff line change
Expand Up @@ -504,6 +504,7 @@ screen. %s/%d are runtime placeholders.
"decoding submit response (got body %q): %w"
"deleting stage Pod %s/%s: %w"
"destination"
"destination table"
"docker info reported cores=%d mem=%d"
"docker info returned %q, want two fields"
"docker info: %w"
Expand Down
32 changes: 32 additions & 0 deletions internal/submit/summary.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,21 @@ type Summary struct {
// "you can grep cluster logs for this ID."
IngestorID string

// DestinationTable is the PHYSICAL table the run wrote into, as
// reported by the ingestor (tracebloc/backend#2895).
//
// Under the cluster's per-ingestion-tables mode this is a
// `ds_<hex>` handle and the operator's --name is only a label, so
// the label names no table that exists: `data delete <--name>`
// fails on a dataset that was just created, and a follow-up
// `data list | grep <--name>` finds nothing — which reads as
// "already clean". Both signals agree and both are wrong.
//
// Empty when the ingestor did not report it (older ingestor, or a
// run that failed before the banner). Callers must fall back to
// the requested name rather than surfacing an empty table.
DestinationTable string

// TotalRecords is the row count the ingestor saw in the
// source data. Includes every row regardless of outcome.
TotalRecords int64
Expand Down Expand Up @@ -169,6 +184,11 @@ var numberRE = regexp.MustCompile(`([0-9][0-9,]*)\s*$`)
// pattern.
var ingestorIDRE = regexp.MustCompile(`Ingestor ID:\s*(.+?)\s*$`)

// destTableRE matches the destination-table line. Like the ingestor
// ID the value is an identifier rather than a number, so it needs its
// own pattern rather than the trailing-number one.
var destTableRE = regexp.MustCompile(`Destination table:\s*(.+?)\s*$`)

// SummaryParser is a streaming parser for the 📊 banner. Feed it
// log lines as they arrive (any chunk size, any line splitting);
// Result() returns the accumulated Summary at any point. The
Expand Down Expand Up @@ -339,6 +359,15 @@ func (p *SummaryParser) feedLine(rawLine string) {
return
}

// Destination table — the name the operator can actually pass to
// `data delete` / `data list`. Handled beside the ingestor ID (and
// like it, not counted as a numeric field) because it is an
// identifier line, not a statistic.
if m := destTableRE.FindStringSubmatch(line); m != nil {
p.summary.DestinationTable = strings.TrimSpace(m[1])
return
}

// Otherwise: try each field pattern. The prefix match is
// linear over a 7-element slice — microscopic overhead per
// line, and the fixed order matches the ingestor's print
Expand Down Expand Up @@ -420,6 +449,9 @@ func RenderSummary(p *ui.Printer, s *Summary) {
}

p.Section("Ingestion summary")
if s.DestinationTable != "" {
p.Field("destination table", s.DestinationTable)
}
if s.IngestorID != "" {
p.Field("ingestor ID", s.IngestorID)
}
Expand Down
67 changes: 67 additions & 0 deletions internal/submit/summary_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -325,3 +325,70 @@ func TestCommaSep(t *testing.T) {
}
}
}

// ---------------------------------------------------------------------------
// backend#2895 — the parser must pick up the destination table.
//
// Under the cluster's per-ingestion-tables mode the physical table is a
// `ds_<hex>` handle while the operator's --name is only a label. The banner is
// the ONLY channel this parser sees, so without this line the CLI cannot name
// the table it just created — and `data delete <--name>` fails on a dataset
// that exists, while `data list | grep <--name>` finds nothing. Both read as
// "already clean", which is why the failure is silent.
// ---------------------------------------------------------------------------

// realBannerWithDestTable mirrors realBanner, plus the destination-table line
// the ingestor prints (tracebloc/data-ingestors#549). Kept as its own fixture
// so the flag-off banner above still pins the legacy shape.
const realBannerWithDestTable = "some earlier log line\n" +
"\x1b[36m" + "════════════════════════════════════════════════════════════" + "\x1b[0m\n" +
"\x1b[1m\x1b[36m📊 INGESTION SUMMARY 📊\x1b[0m\n" +
"\x1b[36m" + "════════════════════════════════════════════════════════════" + "\x1b[0m\n" +
"\x1b[1mIngestor ID:\x1b[0m \x1b[34m46ad219b-8192-4be3-a0eb-3d9120fa10b4\x1b[0m\n" +
"\x1b[1mDestination table:\x1b[0m \x1b[34mds_46ad219b81924be3a0eb3d9120fa10b4\x1b[0m\n" +
"\x1b[1m📈 Total Records Found:\x1b[0m \x1b[34m30,000\x1b[0m\n" +
"\x1b[1m✅ Successfully Processed:\x1b[0m \x1b[32m30,000\x1b[0m\n" +
"\x1b[1m💾 Inserted to Database:\x1b[0m \x1b[32m30,000\x1b[0m\n" +
"\x1b[1m🚀 Sent to API:\x1b[0m \x1b[32m30,000\x1b[0m\n" +
"\x1b[1m⏭️ Skipped Records:\x1b[0m \x1b[33m0\x1b[0m\n" +
"\x1b[1m📁 File Transfer Failures:\x1b[0m \x1b[32m0\x1b[0m\n" +
"\x1b[1m❌ Failed DB Insertion:\x1b[0m \x1b[31m0\x1b[0m\n" +
"\x1b[36m" + "════════════════════════════════════════════════════════════" + "\x1b[0m\n"

func TestSummaryParser_ExtractsDestinationTable(t *testing.T) {
p := NewSummaryParser()
p.Feed([]byte(realBannerWithDestTable))
got := p.Result()
if got == nil {
t.Fatal("parser returned nil summary")
}
const want = "ds_46ad219b81924be3a0eb3d9120fa10b4"
if got.DestinationTable != want {
t.Errorf("DestinationTable = %q, want %q", got.DestinationTable, want)
}
// The handle must not be confused with the ingestor ID it derives from —
// they differ only by `ds_` and the hyphens, so a sloppy regex could
// capture the wrong line and still look plausible.
if got.IngestorID != "46ad219b-8192-4be3-a0eb-3d9120fa10b4" {
t.Errorf("IngestorID = %q, want the hyphenated uuid", got.IngestorID)
}
// The rest of the banner must still parse with the extra line present.
if got.TotalRecords != 30000 || got.InsertedRecords != 30000 {
t.Errorf("counters broke with the new line: total=%d inserted=%d",
got.TotalRecords, got.InsertedRecords)
}
}

func TestSummaryParser_DestinationTableAbsentOnLegacyBanner(t *testing.T) {
// An older ingestor prints no such line. The field must stay empty so
// callers fall back to the requested name rather than reporting "".
p := NewSummaryParser()
p.Feed([]byte(realIngestorBanner))
got := p.Result()
if got == nil {
t.Fatal("parser returned nil summary")
}
if got.DestinationTable != "" {
t.Errorf("DestinationTable = %q, want empty on a legacy banner", got.DestinationTable)
}
}
Loading