# Pushing and reading dataset rows from workflows

Bug bounty hunting is a long game. The subdomain you discover today might host
the admin panel that gets you a critical in six months. Datasets let a project
accumulate knowledge across runs: every scan reconciles its observations into
an evolving table of assets.

## Pushing Observations: `step.update_dataset`

```bhw
step.update_dataset(
    table = "subdomains",
    source = path.file("subs-clean.txt").lines(),
    pipeline = transform.trim().lower().unique(),
    mapping = {
        "host": row.value,
        "source": "subfinder",
    },
    mode = "upsert",
)
```

Parameters:

<table>
<thead>
  <tr>
    <th>
      Parameter
    </th>
    
    <th>
      Meaning
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        table
      </code>
    </td>
    
    <td>
      Name of an existing project dataset table
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        source
      </code>
    </td>
    
    <td>
      Where the data comes from — a path or typed transform expression
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        pipeline
      </code>
    </td>
    
    <td>
      Typed transform producing the records to push
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        mapping
      </code>
    </td>
    
    <td>
      <code>
        {column: expression}
      </code>
      
       projection from each pipeline record (<code>
        row
      </code>
      
      ) into dataset columns
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        mode
      </code>
    </td>
    
    <td>
      <code>
        "upsert"
      </code>
      
       or <code>
        "sync"
      </code>
    </td>
  </tr>
</tbody>
</table>

The workflow only provides source, pipeline, mapping, and mode. Everything
else — identity, reconciliation, stale tracking, provenance, change events — is
owned by the dataset service.

Dataset names must start with a letter and contain only letters, digits, and
hyphens (`live-hosts`, not `live_hosts`). Column names are identifiers and may
contain letters, digits, and underscores, but not the reserved names
`__ingested_at` or `__deleted`.

## Mapping

The mapping dict's keys are dataset column names; each value is an expression
evaluated against the current record, bound as `row`. With a JSONL source,
fields come straight off the record:

```bhw
step.exec(command = ["httpx", "-l", "subs.txt", "-jsonl", "-o", "live.jsonl"])

step.update_dataset(
    table = "live-hosts",
    source = path.file("live.jsonl").jsonl(),
    pipeline = transform.unique(),
    mapping = {
        "url": row.url,
        "status_code": row.status_code,
        "title": row.title,
    },
    mode = "upsert",
)
```

With line-based sources, `row.value` is the raw string:

```bhw
mapping = {"host": row.value}
```

## Modes

`mode` accepts the strings `"upsert"` and `"sync"`. The `dataset.upsert()` /
`dataset.sync()` values are not accepted by `step.update_dataset`.

### `"upsert"`

Create or update records matching known identities; records absent from this
push are left untouched. This is what you want for accumulating observations:

```bhw
# Every nightly run adds new subdomains; existing ones stay.
mode = "upsert"
```

### `"sync"`

The push represents the **complete** current state: identities that were known
but are absent from this push become stale. Use it when the source is
authoritative:

```bhw
# After a full rescan, hosts no longer resolving are marked stale.
mode = "sync"
```

## Reading a Dataset: `step.download_dataset`

`step.download_dataset` exports the current logical state of a dataset into a
file in the activity workspace, so a run can start from knowledge accumulated by
previous runs:

```bhw
step.download_dataset(
    table = "subdomains",
    destination = path.file("known-subdomains.csv"),
    format = "csv",
)
```

Parameters:

<table>
<thead>
  <tr>
    <th>
      Parameter
    </th>
    
    <th>
      Meaning
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        table
      </code>
    </td>
    
    <td>
      Name of an existing project dataset table
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        destination
      </code>
    </td>
    
    <td>
      Workspace file the export is written to (<code>
        path.file(...)
      </code>
      
      )
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        format
      </code>
    </td>
    
    <td>
      <code>
        "csv"
      </code>
      
       (default), <code>
        "jsonl"
      </code>
      
      , or <code>
        "json"
      </code>
    </td>
  </tr>
</tbody>
</table>

Later steps read the file like any other workspace input, and the step exposes
it as `run.output`. The export covers the current logical state: latest
observation per identity, stale records excluded.

## A Complete Asset-Tracking Workflow

```bhw
workflow(
    name = "asset-tracker",
    activities = [
        activity(
            name = "discover-and-record",
            triggers = [trigger.cron("0 0 * * *")],
            steps = [
                # 1. Enumerate.
                step.exec(command = ["subfinder", "-d", config.var("target_domain"), "-silent", "-o", "subs-raw.txt"]),
                # 2. Normalize.
                step.transform(
                    source = path.file("subs-raw.txt").lines(),
                    pipeline = transform.trim().lower().unique(),
                    destination = path.file("subs-clean.txt"),
                ),
                # 3. Record into the project's asset inventory.
                step.update_dataset(
                    table = "subdomains",
                    source = path.file("subs-clean.txt").lines(),
                    pipeline = transform.unique(),
                    mapping = {
                        "host": row.value,
                        "source": "subfinder",
                        "program": config.var("program"),
                    },
                    mode = "upsert",
                ),
            ],
        ),
    ],
)
```

Each night, new subdomains appear in the dataset with their source and program
recorded as provenance; nothing is ever lost between runs. When a later scan
uses `mode = "sync"` after a full validation pass, dead assets age out on
their own instead of polluting your target lists forever.

## Querying the data you pushed

Pushes are only half of the feature — datasets are queried with DuckDB SQL and
typed filters. The Datasets section covers schema design, SQL, filters, views,
and automation:

- [Datasets overview](/datasets/overview): current state vs history, identity,
reserved columns, and limits.
- [Schemas and identity](/datasets/schema): column kinds and identity columns.
- [SQL reference](/datasets/sql): relations, pagination, result types, joins,
aggregates, and time ranges.
- [Structured filters](/datasets/filters): the filter AST used by the Hub and
API.
- [Saved views](/datasets/views): reusable queries across datasets.
- [API and CLI](/datasets/api-cli): `DatasetService` and `bh dataset`.