Materialize a dataset's query surface to Parquet [REPORT-113]

cmd/dataset.go

@@ -21,6 +21,7 @@ func newDatasetCmd() *cobra.Command {
 		newDatasetPurgeCmd(),
 		newDatasetShowCmd(),
 		newDatasetListCmd(),
+		newDatasetMaterializeCmd(),
 		newReindexCmd(),
 	)
 	return cmd

cmd/dataset_list.go

@@ -45,9 +45,13 @@ func renderList(list *dataset.ListJSON) {
 	tw := tabwriter.NewWriter(out, 0, 2, 2, ' ', 0)
 	fmt.Fprintln(tw, "REF\tDESCRIPTION\tAGE\tANSWERS\tHISTORY\tREPORT\tATTACH\tSIZE")
 	for _, d := range list.Datasets {
+		size := humanBytes(d.SizeBytes)
+		if d.MaterializedBytes > 0 {
+			size += fmt.Sprintf(" (%s materialized)", humanBytes(d.MaterializedBytes))
+		}
 		fmt.Fprintf(tw, "%s\t%s\t%s\t%d\t%d\t%d\t%d\t%s\n",
 			d.Ref, d.Description, humanAge(d.AgeSeconds),
-			d.Totals["answers"], d.Totals["history"], d.Totals["report"], d.Totals["attachments"], humanBytes(d.SizeBytes))
+			d.Totals["answers"], d.Totals["history"], d.Totals["report"], d.Totals["attachments"], size)
 	}
 	tw.Flush()
 }

cmd/dataset_materialize.go

@@ -0,0 +1,113 @@
+package cmd
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"sort"
+	"strings"
+
+	"github.com/concord-consortium/cc-data-cli/internal/dataset"
+	"github.com/concord-consortium/cc-data-cli/internal/duck"
+	"github.com/concord-consortium/cc-data-cli/internal/output"
+	"github.com/spf13/cobra"
+)
+
+func newDatasetMaterializeCmd() *cobra.Command {
+	var force, allowPartial bool
+	cmd := &cobra.Command{
+		Use:   "materialize <ref>",
+		Short: "Write each of a dataset's views to a Parquet file queries can read",
+		Long: "Write each of a dataset's file-backed views to " + dataset.MaterializedDir +
+			"/<view>.parquet. Queries read a view's Parquet while it is fresh and fall back to the raw " +
+			"artifacts otherwise, so materializing never changes an answer. The files are ordinary Parquet " +
+			"and any tool can read them by path without going through cc-data.",
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			cfg, root, err := loadRuntime()
+			if err != nil {
+				return err
+			}
+			ref, err := resolveExistingRef(cfg, root, args[0])
+			if err != nil {
+				return err
+			}
+			echoRef(ref)
+			d := dataset.Open(root, ref)
+			if !d.Exists() {
+				return notFound(ref)
+			}
+			opts := duck.MaterializeOptions{
+				Force:        force,
+				AllowPartial: allowPartial,
+				Progress: func(view string, i, n int) {
+					output.Progressf("[%d/%d] %s", i, n, view)
+				},
+			}
+			ctx, stop := interruptible()
+			defer stop()
+			res, err := duck.Materialize(ctx, d, opts, output.Stderr())
+			if err != nil {
+				return mutationErr(err)
+			}
+			return reportMaterialize(res)
+		},
+	}
+	cmd.Flags().BoolVar(&force, "force", false, "re-materialize a view whose inputs are unchanged")
+	cmd.Flags().BoolVar(&allowPartial, "allow-partial", false, "build a view even though a file it declares is missing")
+	return cmd
+}
+
+// interruptible returns a context the first interrupt cancels, so a run that is
+// waiting to commit, or mid-copy, cleans up its temp files instead of leaving
+// them for the next run's sweep. The handler is removed once it has fired, so a
+// second interrupt kills the process as usual; the caller's stop removes it
+// when the run ends without one.
+func interruptible() (context.Context, context.CancelFunc) {
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	go func() {
+		<-ctx.Done()
+		stop()
+	}()
+	return ctx, stop
+}
+
+// reportMaterialize renders the per-view outcome and returns a non-zero exit when
+// any view was refused, so a set -euo pipefail recipe stops rather than consuming
+// a surface that is missing a view.
+func reportMaterialize(res duck.MaterializeResult) error {
+	written, fresh := res.Written(), res.Fresh()
+	discarded, refused := res.Discarded(), res.Refused()
+	output.Progressf("written %d, already fresh %d, discarded %d, refused %d",
+		len(written), len(fresh), len(discarded), len(refused))
+	if len(written) > 0 {
+		output.Progressf("  written: %s", strings.Join(written, ", "))
+	}
+	for _, o := range res.Views {
+		if (o.Status == duck.StatusWritten || o.Status == duck.StatusFresh) && o.Reason != "" {
+			output.Progressf("  %s: %s", o.View, o.Reason)
+		}
+	}
+	if len(discarded) > 0 {
+		output.Progressf("  discarded, because a concurrent change moved their inputs; run again to pick them up: %s",
+			strings.Join(discarded, ", "))
+	}
+	if !res.Refusals() {
+		return nil
+	}
+	views := make([]string, 0, len(refused))
+	for view := range refused {
+		views = append(views, view)
+	}
+	sort.Strings(views)
+	for _, view := range views {
+		output.Progressf("view %s %s", view, refused[view])
+	}
+	// Exit 1, the table's "internal/other" class: a refusal is a local condition,
+	// so the server-contract class would be a lie to anyone scripting on the code.
+	return &output.CLIError{
+		ExitCode: output.ExitInternal,
+		Code:     "MATERIALIZE_REFUSED",
+		Message:  "refused to materialize " + strings.Join(views, ", "),
+	}
+}

cmd/dataset_materialize_test.go

@@ -0,0 +1,126 @@
+package cmd
+
+import (
+	"os"
+	"runtime"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/concord-consortium/cc-data-cli/internal/config"
+	"github.com/concord-consortium/cc-data-cli/internal/dataset"
+	"github.com/concord-consortium/cc-data-cli/internal/output"
+)
+
+func materializeFixture(t *testing.T) *dataset.Dataset {
+	t.Helper()
+	home := t.TempDir()
+	t.Setenv("HOME", home)
+	if runtime.GOOS == "windows" {
+		t.Setenv("USERPROFILE", home)
+	}
+	root := t.TempDir()
+	t.Setenv("CC_DATA_ROOT", root)
+
+	d, err := dataset.Create(root, dataset.Ref{Portal: config.MustPortal("learn.concord.org"), Name: "ds"}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, run := range []int{100, 200} {
+		name := writeReportCSV(t, d, run)
+		rowCount, cols, order, dialect, err := dataset.DetectCSV(d.Path(name), "answers")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := d.UpsertDownload(dataset.Download{
+			Type: "report", RunID: run, ReportType: "answers", Slug: "slug",
+			Files: []string{name}, RowCount: &rowCount, Columns: cols, ColumnOrder: order,
+			CSVDialect: &dialect, Complete: true, FetchedAt: time.Unix(int64(run), 0).UTC(),
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// A dimension view reading its own CSV, so one view stays clean when another
+	// view's input is deleted and the refusal can be shown to be per view.
+	mapping := "report_300.csv"
+	body := "learner_id,user_id,primary_user_id,student_id,class_id,offering_id,runnable_url,run_remote_endpoint\n" +
+		"1,11,11,s1,30,70,https://act/1,https://portal/dataservice/external_activity_data/AAA\n"
+	if err := os.WriteFile(d.Path(mapping), []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	rowCount, cols, order, dialect, err := dataset.DetectCSV(d.Path(mapping), dataset.ReportTypePortal)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := d.UpsertDownload(dataset.Download{
+		Type: "report", RunID: 300, ReportType: dataset.ReportTypePortal, Slug: "student-id-mapping",
+		Files: []string{mapping}, RowCount: &rowCount, Columns: cols, ColumnOrder: order,
+		CSVDialect: &dialect, Complete: true, FetchedAt: time.Unix(300, 0).UTC(),
+	}); err != nil {
+		t.Fatal(err)
+	}
+	return d
+}
+
+// writeReportCSV writes one report CSV and returns its dataset-relative name.
+func writeReportCSV(t *testing.T, d *dataset.Dataset, run int) string {
+	t.Helper()
+	name := "report_" + strconv.Itoa(run) + ".csv"
+	body := "student_id,res_1_q1_answer\nPrompt,What?\nCorrect answer,42\n1,hello\n"
+	if err := os.WriteFile(d.Path(name), []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return name
+}
+
+func TestDatasetMaterializeReportsAndExitsZero(t *testing.T) {
+	materializeFixture(t)
+
+	_, errOut, code := runArgs(t, "dataset", "materialize", "learn.concord.org/ds")
+	if code != output.ExitSuccess {
+		t.Fatalf("clean run exit = %d, want 0: %s", code, errOut)
+	}
+	for _, want := range []string{"reports", "report_prompts", "written:"} {
+		if !strings.Contains(errOut, want) {
+			t.Fatalf("output should name %q: %s", want, errOut)
+		}
+	}
+}
+
+// TestDatasetMaterializeExitsNonZeroOnARefusal is the assertion that matters: the
+// exit code is what stops a set -euo pipefail recipe from consuming a surface
+// that is missing a view.
+func TestDatasetMaterializeExitsNonZeroOnARefusal(t *testing.T) {
+	d := materializeFixture(t)
+	if err := os.Remove(d.Path("report_200.csv")); err != nil {
+		t.Fatal(err)
+	}
+
+	_, errOut, code := runArgs(t, "dataset", "materialize", "learn.concord.org/ds")
+	if code == output.ExitSuccess {
+		t.Fatalf("a refused view must not exit 0: %s", errOut)
+	}
+	if !strings.Contains(errOut, "report_200.csv") {
+		t.Fatalf("the message must name the missing file: %s", errOut)
+	}
+	if !strings.Contains(errOut, "re-fetch run 200") || !strings.Contains(errOut, "dataset reindex") {
+		t.Fatalf("the message must carry both remedies: %s", errOut)
+	}
+	if !strings.Contains(errOut, "view reports declares 3 inputs") {
+		t.Fatalf("the message must name the view and the shortfall: %s", errOut)
+	}
+	// The view that does not read the deleted CSV is still written, which is what
+	// makes the refusal per view rather than per command.
+	if !strings.Contains(errOut, "written: student_id_mapping") {
+		t.Fatalf("a clean view should still be written: %s", errOut)
+	}
+
+	_, partialOut, partialCode := runArgs(t, "dataset", "materialize", "learn.concord.org/ds", "--allow-partial")
+	if partialCode != output.ExitSuccess {
+		t.Fatalf("--allow-partial should exit 0, got %d: %s", partialCode, partialOut)
+	}
+	if !strings.Contains(partialOut, "reports: built from 2 of 3 declared inputs") {
+		t.Fatalf("--allow-partial should report the shortfall as a count: %s", partialOut)
+	}
+}

cmd/dataset_show.go

@@ -6,6 +6,7 @@ import (
 	"text/tabwriter"
 
 	"github.com/concord-consortium/cc-data-cli/internal/dataset"
+	"github.com/concord-consortium/cc-data-cli/internal/duck"
 	"github.com/concord-consortium/cc-data-cli/internal/output"
 	"github.com/spf13/cobra"
 )
@@ -29,7 +30,7 @@ func newDatasetShowCmd() *cobra.Command {
 			if !d.Exists() {
 				return notFound(ref)
 			}
-			summary, err := d.BuildShowJSON(full)
+			summary, err := duck.ShowJSON(d, full)
 			if err != nil {
 				return output.Internalf("%v", err)
 			}
@@ -51,7 +52,11 @@ func renderShow(s *dataset.ShowJSON, full bool) {
 	if s.Description != "" {
 		fmt.Fprintf(out, "%s\n", s.Description)
 	}
-	fmt.Fprintf(out, "created %s, %s on disk\n\n", s.CreatedAt.Format("2006-01-02"), humanBytes(s.SizeBytes))
+	disk := humanBytes(s.SizeBytes)
+	if s.MaterializedBytes > 0 {
+		disk += fmt.Sprintf(" (%s materialized)", humanBytes(s.MaterializedBytes))
+	}
+	fmt.Fprintf(out, "created %s, %s on disk\n\n", s.CreatedAt.Format("2006-01-02"), disk)
 
 	fmt.Fprintln(out, "Totals:")
 	types := make([]string, 0, len(s.Totals))

docs/researcher-guide.md

@@ -344,6 +344,48 @@ When you (or Claude) query a dataset, the data is exposed as a set of SQL
 | `student_metadata` | One row per learner from Student Metadata runs: name, username, class, school, teachers, permission forms. |
 | `run_membership`, `downloads` | Provenance: which run's fetch covered which records, and what each download was, including whether its run hid names. |
 
+### Speeding up a large dataset
+
+Every query re-reads the JSONL and CSV files your downloads produced. That is
+fine for a few thousand answers and slow once a dataset runs to gigabytes.
+`cc-data dataset materialize <ref>` writes each view to a compressed Parquet file
+that queries read instead, while the copy is still current:
+
+```
+<data-root>/<portal>/datasets/<name>/materialized/<view>.parquet
+```
+
+A few things worth knowing about that folder.
+
+- **It never changes an answer.** A query returns the same rows either way. When a
+  later download changes a file a view reads, that view quietly goes back to
+  reading the raw files, and `cc-data dataset show <ref>` tells you which views
+  are worth refreshing.
+- **The path is stable, and the files are ordinary Parquet.** Any tool that reads
+  Parquet, DuckDB, pandas or Polars, can open one directly without going through
+  cc-data, which is the point of the fixed path: analysis scripts can hardcode it.
+- **You can delete the folder whenever you like**; queries go back to the raw
+  files, `cc-data dataset show <ref>` names the views that did, and materializing
+  again rebuilds them. cc-data itself removes a finished `.parquet` only when you
+  purge or delete the whole dataset, so a script of yours can rely on one being
+  there until you say otherwise. The one exception is housekeeping: if a run is
+  interrupted partway through, the next run clears the half-written file it left
+  behind.
+- **It is bigger than the one view you care about**, for two reasons. `reports`
+  unions the same log CSVs `logs` reads, so both carry those rows (on a
+  200,000-row log fixture, a 5.4 MB `reports.parquet` beside a 9.5 MB
+  `logs.parquet`). And a view's derived columns cost nothing until they are
+  written down: `logs` carries both `parameters` and its parsed `parameters_json`,
+  which together with the other derived columns is about 45% of that file.
+- **`dataset show` and `dataset list` report the folder's size separately**, so
+  you can see at a glance what deleting it would reclaim.
+
+If a view reads a file that is no longer on disk, materializing refuses that view
+rather than writing a short Parquet, names the file, and tells you both ways to
+fix it; other views are still written, and the command exits non-zero so a script
+stops. `--allow-partial` builds from what is there anyway and says how much of
+the declared input it used.
+
 ### Report types
 
 You generate report runs in the report server, and `cc-data` downloads whichever