Datasets

Datasets give Radish apps a built-in analytics aggregation layer for data that lives outside the app — in a data warehouse, a sibling MongoDB collection, or an HTTP API. Enable the feature, define your datasets as data (not code), and query rollups, trends, and rankings through a typed service API.

This is deliberately not a general-purpose BI tool. Datasets solve a specific problem: your app needs to show charts, KPIs, and leaderboards derived from an upstream source, without coupling your frontend to that source's query language or connection details.

When to Use Datasets

Use case	Datasets?	Why
Dashboard showing sales by category from a data warehouse	✅ Yes	Classic rollup over remote data
Time-series chart of order volume by month	✅ Yes	Trend query with grain bucketing
"Top 10 customers by revenue" widget	✅ Yes	TopN ranking
Full-text search across products	❌ No	Use Search instead
Counting records in your own entities	❌ No	Use the entity service's `count()` method
Real-time event stream processing	❌ No	Datasets sync periodically, not in real time

Quick Start

1. Enable the feature

Add datasets to your app blueprint's features:

{
  "features": {
    "datasets": true
  }
}

Or pass it on the command line:

radish-cli create datalayer . --schema blueprints/app.types.json --features datasets

2. Generate

radish-cli create datalayer . --schema blueprints/app.types.json

This generates three entities (DatasetDefinition, DatasetField, DatasetRow), two services (DatasetService, DatasetSyncService), and the corresponding API routes.

3. Create a dataset definition

import { DatasetDefinitionService, systemAuth } from '@generated/datalayer/services';

const svc = new DatasetDefinitionService(systemAuth());
await svc.create({
  datasetName: 'invoiceActivity',
  kind: 'fact',
  source: 'databricks',
  sourceRef: 'analytics.commerce.invoice_activity_v2',
  syncMode: 'scheduled',
  syncSchedule: '0 6 * * *',
  grainColumns: ['invoice_id', 'line_num', 'shipment_skey'],
  enabled: true,
});

4. Register field metadata

import { DatasetFieldService, systemAuth } from '@generated/datalayer/services';

const svc = new DatasetFieldService(systemAuth());

// Dimensions — groupable columns
await svc.create({ datasetId: defId, columnName: 'marketCategory', label: 'Market Category', role: 'dimension', dataType: 'string', provenance: 'curated' });
await svc.create({ datasetId: defId, columnName: 'invoiceDate',    label: 'Invoice Date',    role: 'dimension', dataType: 'date', grain: 'day', provenance: 'curated' });

// Measures — aggregatable values
await svc.create({ datasetId: defId, columnName: 'netSales', label: 'Net Sales', role: 'measure', dataType: 'float', agg: 'sum', provenance: 'curated' });
await svc.create({ datasetId: defId, columnName: 'cogs',     label: 'COGS',      role: 'measure', dataType: 'float', agg: 'sum', provenance: 'curated' });
await svc.create({ datasetId: defId, columnName: 'avgDays',  label: 'Avg Days',  role: 'measure', dataType: 'float', agg: 'avg', provenance: 'curated' });

5. Sync data from the source

import { syncDataset } from '@generated/datalayer/services/dataset-sync.service';

const result = await syncDataset('invoiceActivity', async (ctx) => {
  // Your source-specific fetch logic
  return fetchFromDatabricks(client, ctx.definition.sourceRef);
}, {
  grainColumns: ['invoice_id', 'line_num', 'shipment_skey'],
});

console.log(`Synced ${result.inserted} inserted, ${result.updated} updated`);

6. Query aggregations

import { DatasetService, systemAuth } from '@generated/datalayer/services';

const svc = new DatasetService(systemAuth());

// Rollup: sales by category
const rollup = await svc.rollup({
  dataset: 'invoiceActivity',
  by: ['marketCategory'],
  measures: ['netSales', 'cogs'],
  where: { bucketDate: { $gte: '2026-06-01', $lte: '2026-08-31' } },
});

// Trend: monthly sales
const trend = await svc.trend({
  dataset: 'invoiceActivity',
  grain: 'month',
  measures: ['netSales'],
});

// Top N: biggest customers
const topN = await svc.topN({
  dataset: 'invoiceActivity',
  by: 'customerId',
  measure: 'netSales',
  n: 25,
});

Architecture

Three Entities

Datasets use three entities that share one MongoDB collection pattern:

Entity	Purpose	Key fields
DatasetDefinition	What to sync, where from, how often	`datasetName`, `source`, `sourceRef`, `grainColumns`, `syncSchedule`
DatasetField	Per-column metadata: role, type, aggregation	`datasetId`, `columnName`, `role`, `agg`, `provenance`
DatasetRow	The actual data — dimensions + measures per grain	`datasetId`, `grainKey`, `bucketDate`, `dimensions`, `measures`

Two Services

Service	Responsibility	Auth model
DatasetService	Read path — `rollup()`, `trend()`, `topN()`	Requires `datasetrow:view:all` permission
DatasetSyncService	Write path — `syncDataset()` function	Uses `systemAuth()` (no user context)

Data Flow

┌──────────────────┐       ┌──────────────────┐      ┌───────────────┐
│  Upstream Source  │──────▶│  DatasetSync     │─────▶│  DatasetRow   │
│  (Databricks,    │ fetch │  Service          │ map  │  Collection   │
│   Mongo, HTTP)   │       │  ┌─field plan─┐  │  +   │               │
└──────────────────┘       │  │ dims: [...]│  │ upsert└───────┬───────┘
                           │  │ measures:[]│  │               │
                           │  │ grain: ... │  │               │ aggregate
                           │  └────────────┘  │               │ pipeline
                           └──────────────────┘       ┌───────▼───────┐
                                                      │ DatasetService│
                                                      │  rollup()     │
                                                      │  trend()      │
                                                      │  topN()       │
                                                      └───────────────┘

DatasetDefinition

A definition describes one logical dataset — where it comes from, how it syncs, and what makes a row unique.

Fields

Field	Type	Required	Description
`datasetName`	string	✅	Unique identifier (e.g. `invoiceActivity`, `qrScans`)
`kind`	enum	✅	`fact` (measures over time) or `dimension` (lookup data)
`source`	enum	✅	`databricks`, `mongo`, or `http`
`sourceRef`	string	✅	Upstream identifier — catalog path, collection name, or URL
`syncMode`	enum	✅	`scheduled` or `manual`
`syncSchedule`	string		Cron expression for scheduled sync (e.g. `0 6 * * *`)
`grainColumns`	string[]		Upstream columns that form the natural key (see Grain Keys)
`enabled`	boolean		Whether scheduled sync runs (default `true`)
`baseFilter`	any		Default filter applied to every query
`indexedDims`	string[]		Dimension fields with dedicated compound indexes
`schemaVersion`	int		Incremented when field layout changes (default `1`)
`syncStatus`	enum		`ok`, `drift`, `error`, or `pending` (default `pending`)
`lastSyncedAt`	isoDate		Timestamp of last successful sync
`lastSyncError`	string		Error message from the last failed sync
`rowCount`	int		Number of rows in the local store (default `0`)
`description`	string		Human-readable description

Grain Keys

The grainColumns field is the most important configuration on a dataset definition. It declares which upstream columns form the natural key — the combination of values that uniquely identifies a row.

{
  "grainColumns": ["invoice_id", "line_num", "shipment_skey"]
}

The sync engine hashes these values to produce a grainKey on each DatasetRow. This key drives upsert matching: on subsequent syncs, rows with the same grain key are updated rather than duplicated.

Getting the grain wrong is dangerous. If the grain is too coarse (missing a key column), distinct upstream rows collapse into one — silently discarding data. If the grain is too fine (including a volatile column), every sync inserts new rows instead of updating.

DatasetField

Field metadata describes each column in the dataset: what role it plays, how it aggregates, and whether a human has reviewed the classification.

Fields

Field	Type	Required	Description
`datasetId`	objectId	✅	Reference to DatasetDefinition
`columnName`	string	✅	Column name as it appears upstream
`label`	string	✅	Human-readable display label
`role`	enum	✅	`dimension`, `measure`, or `excluded`
`dataType`	enum	✅	`string`, `int`, `float`, `boolean`, or `date`
`agg`	enum		Aggregation function: `sum`, `avg`, `min`, `max`, `count`, `none` (default `none`)
`grain`	enum		For date dimensions: `day`, `week`, `month`, `quarter`, `year`
`provenance`	enum	✅	`generated` (overwritable on sync) or `curated` (preserved on sync)
`needsReview`	boolean		Flagged for human review (default `false`)
`lastSeenAt`	isoDate		Last time this column appeared in the upstream schema
`position`	int		Display order (default `0`)

Roles

dimension — A groupable column. Used in rollup({ by: [...] }) and topN({ by: '...' }). Examples: marketCategory, customerId, region.
measure — A numeric value that gets aggregated. Each measure declares its agg function so callers cannot accidentally SUM an average. Examples: netSales (sum), avgDaysToShip (avg), orderCount (count).
excluded — Ignored by the query engine. Use for columns that exist upstream but aren't useful for analytics (raw IDs, internal flags).

Provenance

Provenance controls what happens to a field during re-sync:

generated — Auto-classified by the sync engine. Will be overwritten if the upstream schema changes.
curated — Manually reviewed and set by a human. Preserved across syncs even if the column disappears temporarily.

DatasetRow

The row store holds the actual data. All datasets share one collection, discriminated by datasetId.

Fields

Field	Type	Required	Description
`datasetId`	objectId	✅	Which dataset this row belongs to
`grainKey`	string	✅	Natural key for upsert (hash of grain column values)
`bucketDate`	isoDate	✅	Time dimension for time-series queries
`dimensions`	object	✅	Groupable values: `{ marketCategory: 'Frozen', region: 'West' }`
`measures`	object	✅	Numeric values: `{ netSales: 1234.56, qty: 42 }`
`syncedAt`	isoDate	✅	When this row was last synced

Indexes

Index	Purpose
`{ datasetId, grainKey }` (unique)	Upsert matching — the natural key
`{ datasetId, bucketDate }`	Time-range queries in trend()
`{ datasetId, 'dimensions.$**' }` (wildcard)	Fast drilldown on any dimension value

The wildcard index on dimensions.$** means queries that filter on dimension values (e.g. where: { marketCategory: 'Frozen' }) hit an index rather than scanning the collection.

DatasetService (Read Path)

The DatasetService compiles queries against DatasetField metadata into push-down MongoDB aggregation pipelines. The service validates dimension and measure names against the field catalog, so typos and invalid columns fail fast with clear error messages.

All three methods require the datasetrow:view:all permission.

rollup()

Aggregate rows by one or more dimensions.

interface RollupQuery {
  dataset: string;       // Dataset name (matches DatasetDefinition.datasetName)
  by: string[];          // Dimension fields to group by
  where?: Record<string, any>;  // Filters on dimensions and time range
  measures?: string[];   // Measure fields to aggregate (defaults to all)
}

Each measure is aggregated according to its declared agg function from the DatasetField metadata.

const result = await svc.rollup({
  dataset: 'invoiceActivity',
  by: ['marketCategory'],
  measures: ['netSales', 'cogs'],
  where: {
    bucketDate: { $gte: '2026-06-01', $lte: '2026-08-31' },
    region: 'West',
  },
});

// result.rows → [
//   { marketCategory: 'Frozen', netSales: 45231.89, cogs: 32100.00 },
//   { marketCategory: 'Dairy',  netSales: 28790.12, cogs: 19500.00 },
// ]

trend()

Time-series aggregation — bucket measures by a time grain.

interface TrendQuery {
  dataset: string;
  grain: 'day' | 'week' | 'month' | 'quarter' | 'year';
  measures?: string[];
  where?: Record<string, any>;
}

const result = await svc.trend({
  dataset: 'invoiceActivity',
  grain: 'month',
  measures: ['netSales'],
  where: { marketCategory: 'Frozen' },
});

// result.rows → [
//   { bucket: '2026-06', netSales: 12000.00 },
//   { bucket: '2026-07', netSales: 15400.00 },
//   { bucket: '2026-08', netSales: 17831.89 },
// ]

topN()

Rank a single dimension by a single measure.

interface TopNQuery {
  dataset: string;
  by: string;            // Dimension to rank
  measure: string;       // Measure to sort by
  n?: number;            // Results to return (default: 10, max: 1000)
  sort?: 'asc' | 'desc'; // Sort direction (default: 'desc')
  where?: Record<string, any>;
}

const result = await svc.topN({
  dataset: 'invoiceActivity',
  by: 'customerId',
  measure: 'netSales',
  n: 5,
});

// result.rows → [
//   { customerId: 'C-4021', netSales: 98234.56 },
//   { customerId: 'C-1187', netSales: 72100.00 },
//   ...
// ]

Filtering

The where clause supports dimension filters and time ranges:

where: {
  // Time range — applied directly to bucketDate
  bucketDate: { $gte: '2026-01-01', $lte: '2026-06-30' },

  // Dimension filters — matched against dimensions.{key}
  region: 'West',
  marketCategory: { $in: ['Frozen', 'Dairy'] },
}

DatasetSyncService (Write Path)

The sync service is a standalone function (not a class method) that runs with system auth. It is source-agnostic: you provide a fetchSource callback that returns rows from whatever upstream system you use.

syncDataset()

import { syncDataset } from '@generated/datalayer/services/dataset-sync.service';

const result = await syncDataset(
  'invoiceActivity',          // dataset name
  fetchSource,                 // callback: (ctx) => Promise<SourceRow[]>
  { grainColumns: ['invoice_id', 'line_num', 'shipment_skey'] },
  { batchSize: 500, log: console.log }
);

Parameters

Parameter	Type	Description
`datasetName`	string	Name matching a DatasetDefinition
`fetchSource`	function	Callback that receives `{ definition, since?, maxRows? }` and returns rows
`config`	object	`{ grainColumns: string[], bucketColumn?: string }`
`opts`	SyncOptions	Optional: `since`, `maxRows`, `batchSize`, `mode`, `dryRun`, `log`

SyncOptions

Option	Type	Default	Description
`since`	string		ISO date — only sync rows newer than this
`maxRows`	number		Hard cap on source rows to fetch
`batchSize`	number	`1000`	Rows per bulkUpsert batch
`mode`	string	`'upsert'`	`'upsert'` updates existing, `'replace'` wipes and re-inserts
`dryRun`	boolean	`false`	Run the full pipeline but write nothing
`log`	function	no-op	Logger for progress messages

SyncResult

interface SyncResult {
  dataset: string;
  sourceRows: number;
  inserted: number;
  updated: number;
  skipped: number;
  errors: string[];
  durationMs: number;
  dryRun: boolean;
}

Sync Pipeline

The sync engine runs a seven-step pipeline:

Resolve definition — Look up the DatasetDefinition by name
Set status — Mark syncStatus: 'pending'
Load field plan — Read DatasetField rows, classify into dimensions/measures
Fetch source — Call your fetchSource callback
Map rows — Transform source rows into DatasetRow shape using the field plan
Write rows — Batch upsert into the DatasetRow collection
Status writeback — Update syncStatus, lastSyncedAt, rowCount

If any step throws, the engine writes syncStatus: 'error' and lastSyncError back to the definition before re-throwing.

Field Plan

The sync engine builds a field plan from DatasetField rows:

interface FieldPlan {
  dimensions: { column: string; name: string }[];
  measures: { column: string; name: string; agg: string }[];
  bucketColumn: string | null;
  grainColumns: string[];
}

Fields with role: 'excluded' are skipped
Date-typed dimension fields auto-detect as the bucketColumn
Non-finite measure values (NaN, Infinity) are dropped to prevent poisoning $sum rollups
Fields with needsReview: true are logged as warnings but still included

Writing a fetchSource Callback

The callback receives context about the dataset and returns flat rows:

const fetchSource: FetchSource = async (ctx) => {
  // ctx.definition — the full DatasetDefinition document
  // ctx.since — ISO date string (if incremental sync)
  // ctx.maxRows — row cap (if set)

  const { sourceRef } = ctx.definition;

  // Databricks example
  const rows = await databricksClient.query(
    `SELECT * FROM ${sourceRef}` +
    (ctx.since ? ` WHERE updated_at >= '${ctx.since}'` : '') +
    (ctx.maxRows ? ` LIMIT ${ctx.maxRows}` : '')
  );

  return rows;
};

The returned rows should be flat objects keyed by column name:

[
  { invoice_id: 'INV-001', line_num: 1, marketCategory: 'Frozen', netSales: 1234.56, invoiceDate: '2026-08-15' },
  { invoice_id: 'INV-001', line_num: 2, marketCategory: 'Dairy',  netSales: 890.00,  invoiceDate: '2026-08-15' },
]

Dry Run

Use dryRun: true to validate the full pipeline without writing:

const result = await syncDataset('invoiceActivity', fetchSource, config, {
  dryRun: true,
  log: console.log,
});

console.log(`Would sync ${result.inserted} rows (${result.skipped} skipped, ${result.errors.length} errors)`);

API Routes

When datasets is enabled, Radish generates standard CRUD routes for all three entities plus analytics endpoints on DatasetService:

Method	Path	Description
`GET`	`/api/v1/datasetdefinitions`	List dataset definitions
`POST`	`/api/v1/datasetdefinitions`	Create a definition
`GET`	`/api/v1/datasetdefinitions/:id`	Get a definition
`PATCH`	`/api/v1/datasetdefinitions/:id`	Update a definition
`DELETE`	`/api/v1/datasetdefinitions/:id`	Delete a definition
`GET`	`/api/v1/datasetfields`	List dataset fields
`POST`	`/api/v1/datasetfields`	Create a field
`GET`	`/api/v1/datasetrows`	List raw rows
`POST`	`/api/v1/datasetrows/rollup`	Rollup aggregation
`POST`	`/api/v1/datasetrows/trend`	Trend aggregation
`POST`	`/api/v1/datasetrows/topn`	Top-N ranking

Rollup endpoint

curl -X POST http://localhost:5173/api/v1/datasetrows/rollup \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "dataset": "invoiceActivity",
    "by": ["marketCategory"],
    "measures": ["netSales"],
    "where": { "bucketDate": { "$gte": "2026-06-01" } }
  }'

Integration with Scheduled Jobs

Datasets pair naturally with Scheduled Jobs for automated sync. The scheduler and handler registry are app code you write (see Scheduled Jobs: What You Build). Here is the recommended pattern for a dataset sync handler:

// YOUR APP CODE — register a handler in your job registry
import { registerJobHandler } from './jobs/registry';
import { syncDataset } from '@generated/datalayer/services/dataset-sync.service';

registerJobHandler('datasets:sync-invoices', async (context) => {
  const result = await syncDataset('invoiceActivity', fetchFromDatabricks, {
    grainColumns: ['invoice_id', 'line_num', 'shipment_skey'],
  }, {
    since: context.lastRunAt,
    log: (msg) => context.log(msg),
  });

  // Optionally record sync metrics as a job snapshot
  if (context.snapshot) {
    context.snapshot({}, {
      rows_synced: result.inserted + result.updated,
      errors: result.errors.length,
      duration_ms: result.durationMs,
    });
  }
}, {
  name: 'Invoice Activity Sync',
});

Then create the corresponding ScheduledJob record (via API, admin panel, or seed script) with handler: 'datasets:sync-invoices' and schedules: ['0 6 * * *'].

Best Practices

DO:

✅ Set grainColumns on every dataset — it prevents silent duplicates
✅ Mark field provenance as curated after manual review
✅ Use dryRun: true on the first sync of a new dataset
✅ Set agg on every measure field — prevents accidental SUM of averages
✅ Filter by bucketDate in queries to limit the aggregation window
✅ Use incremental sync (since option) for large datasets

DON'T:

❌ Use datasets for data already in your app's entities — query those directly
❌ Skip DatasetField setup — the sync engine refuses to run without fields
❌ Use replace mode in production without understanding the data gap — rows are deleted before new ones are inserted
❌ Set grain columns to include volatile fields (timestamps, sequence numbers) — this causes unbounded row growth

Next Steps

Job Snapshots — Record sync metrics alongside your scheduled jobs
Services — How the generated service layer works
Routes — API route generation details
Access Control — Permission model for dataset queries

When to Use Datasets​

Quick Start​

1. Enable the feature​

2. Generate​

3. Create a dataset definition​

4. Register field metadata​

5. Sync data from the source​

6. Query aggregations​

Architecture​

Three Entities​

Two Services​

Data Flow​

DatasetDefinition​

Fields​

Grain Keys​

DatasetField​

Fields​

Roles​

Provenance​

DatasetRow​

Fields​

Indexes​

DatasetService (Read Path)​

rollup()​

trend()​

topN()​

Filtering​

DatasetSyncService (Write Path)​

syncDataset()​

Parameters​

SyncOptions​

SyncResult​

Sync Pipeline​

Field Plan​

Writing a fetchSource Callback​

Dry Run​

API Routes​

Rollup endpoint​

Integration with Scheduled Jobs​

Best Practices​

DO:​

DON'T:​

Next Steps​

When to Use Datasets

Quick Start

1. Enable the feature

2. Generate

3. Create a dataset definition

4. Register field metadata

5. Sync data from the source

6. Query aggregations

Architecture

Three Entities

Two Services

Data Flow

DatasetDefinition

Fields

Grain Keys

DatasetField

Fields

Roles

Provenance

DatasetRow

Fields

Indexes

DatasetService (Read Path)

rollup()

trend()

topN()

Filtering

DatasetSyncService (Write Path)

syncDataset()

Parameters

SyncOptions

SyncResult

Sync Pipeline

Field Plan

Writing a fetchSource Callback

Dry Run

API Routes

Rollup endpoint

Integration with Scheduled Jobs

Best Practices

DO:

DON'T:

Next Steps