Two sources joined

Join two earthquake feeds on an event identifier, and check that the join did not fan out.

The question

Which of the earthquakes reported in the last twenty-four hours does the USGS also count as significant this month?

Two feeds answer half of it each. The first lists every located event of the last day. The second lists the month’s significant events with a score. The join makes one table out of them.

The sources

	Last day	Significant month
Address	`…/summary/all_day.geojson`	`…/summary/significant_month.geojson`
Format	GeoJSON, served as `application/json; charset=utf-8`	the same
Cadence	rebuilt every minute	rebuilt every minute
Rights	a USGS product, published without copyright as a work of the federal government	the same

Both sources pin json.tabular@1.1.0 rather than geojson.vector@1.0.0, because this table wants scalar properties and a point’s two coordinates, not a geometry column. The GeoJSON Reader carries the geometry itself.

The two feeds share an identifier, each one fits inside its own fetch ceiling, and anybody can state the rights basis.

The recipe

{
  "dataset": {
    "name": "Today's earthquakes, flagged against the significant feed",
    "description": "Each row is one earthquake the USGS reported in the last day, carrying its magnitude and location and whether the same event also appears in the significant-events feed for the month."
  },
  "question": {
    "text": "Which of the earthquakes reported in the last day does the USGS also count as significant this month?"
  },
  "table": {
    "name": "daily_events_flagged",
    "description": "One row per event in the last-day feed.",
    "grain": [
      "event_id"
    ],
    "columns": [
      {
        "name": "event_id",
        "type": "string",
        "nullable": false,
        "description": "The USGS identifier for the event.",
        "presentation": {
          "chart": "none"
        }
      },
      {
        "name": "occurred_at",
        "type": "timestamp",
        "nullable": false,
        "description": "When the event happened, in UTC.",
        "presentation": {
          "chart": "timeline",
          "bucket": "hour"
        }
      },
      {
        "name": "magnitude",
        "type": "decimal",
        "nullable": true,
        "description": "The magnitude the USGS reports for the event.",
        "presentation": {
          "chart": "histogram",
          "bins": 10,
          "story": "Almost every event is small and the few large ones sit far to the right."
        }
      },
      {
        "name": "place",
        "type": "string",
        "nullable": true,
        "description": "The place description the USGS prints for the event.",
        "presentation": {
          "chart": "top_values"
        }
      },
      {
        "name": "latitude",
        "type": "decimal",
        "nullable": false,
        "description": "Event latitude."
      },
      {
        "name": "longitude",
        "type": "decimal",
        "nullable": false,
        "description": "Event longitude."
      },
      {
        "name": "depth_km",
        "type": "decimal",
        "nullable": true,
        "description": "Depth below ground level."
      },
      {
        "name": "is_significant",
        "type": "boolean",
        "nullable": false,
        "description": "Whether this event also appears in the month's significant-events feed.",
        "presentation": {
          "chart": "share"
        }
      },
      {
        "name": "significance",
        "type": "integer",
        "nullable": true,
        "description": "The significance score the significant-events feed carries, null for an event that is not in it."
      }
    ]
  },
  "sources": [
    {
      "name": "all_day",
      "description": "Every earthquake the USGS located in the last twenty-four hours.",
      "source_class": "user_api",
      "data_classification": "public",
      "locator": {
        "kind": "https_url",
        "display_locator": "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_day.geojson"
      },
      "rights_claim": {
        "claimed_basis": "public_domain_asserted",
        "claim_evidence_digest": "sha256:1111111111111111111111111111111111111111111111111111111111111111",
        "claim_note": "A United States Geological Survey product, published without copyright as a work of the federal government."
      },
      "connector": {
        "adapter_id": "public.https",
        "credential_mode": "none",
        "origin": "https://earthquake.usgs.gov",
        "parameters": [
          {
            "name": "reader.family_id",
            "value": "json.tabular"
          },
          {
            "name": "reader.family_version",
            "value": "1.1.0"
          },
          {
            "name": "reader.decode_options",
            "value": "{\"columns\":[{\"name\":\"event_id\",\"pointer\":\"/id\",\"required\":true},{\"name\":\"magnitude\",\"pointer\":\"/properties/mag\",\"required\":false},{\"name\":\"place\",\"pointer\":\"/properties/place\",\"required\":false},{\"name\":\"occurred_at_ms\",\"pointer\":\"/properties/time\",\"required\":true},{\"name\":\"longitude\",\"pointer\":\"/geometry/coordinates/0\",\"required\":true},{\"name\":\"latitude\",\"pointer\":\"/geometry/coordinates/1\",\"required\":true},{\"name\":\"depth_km\",\"pointer\":\"/geometry/coordinates/2\",\"required\":false}],\"document_format\":\"json\",\"expand\":[],\"records_pointer\":\"/features\"}"
          }
        ]
      },
      "limits": {
        "max_source_bytes": 3145728,
        "max_rows": 50000,
        "max_requests": 2
      }
    },
    {
      "name": "significant_month",
      "description": "The earthquakes the USGS counts as significant over the last month.",
      "source_class": "user_api",
      "data_classification": "public",
      "locator": {
        "kind": "https_url",
        "display_locator": "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/significant_month.geojson"
      },
      "rights_claim": {
        "claimed_basis": "public_domain_asserted",
        "claim_evidence_digest": "sha256:1111111111111111111111111111111111111111111111111111111111111111",
        "claim_note": "A United States Geological Survey product, published without copyright as a work of the federal government."
      },
      "connector": {
        "adapter_id": "public.https",
        "credential_mode": "none",
        "origin": "https://earthquake.usgs.gov",
        "parameters": [
          {
            "name": "reader.family_id",
            "value": "json.tabular"
          },
          {
            "name": "reader.family_version",
            "value": "1.1.0"
          },
          {
            "name": "reader.decode_options",
            "value": "{\"columns\":[{\"name\":\"event_id\",\"pointer\":\"/id\",\"required\":true},{\"name\":\"significance\",\"pointer\":\"/properties/sig\",\"required\":false}],\"document_format\":\"json\",\"expand\":[],\"records_pointer\":\"/features\"}"
          }
        ]
      },
      "limits": {
        "max_source_bytes": 3145728,
        "max_rows": 5000,
        "max_requests": 2
      }
    }
  ],
  "transform": {
    "engine": "duckdb_sql",
    "steps": [
      {
        "step_id": "daily_events",
        "sql": "select event_id, to_timestamp(cast(occurred_at_ms as bigint) / 1000) as occurred_at, try_cast(magnitude as decimal(6, 3)) as magnitude, place, try_cast(latitude as decimal(9, 5)) as latitude, try_cast(longitude as decimal(9, 5)) as longitude, try_cast(depth_km as decimal(9, 3)) as depth_km from all_day",
        "description": "Cast the last-day feed. The publisher's time is epoch milliseconds, so it is divided before it becomes an instant."
      },
      {
        "step_id": "significant_events",
        "sql": "select event_id, cast(try_cast(significance as decimal(10, 0)) as integer) as significance from significant_month",
        "description": "Cast the significant feed down to the identity and the score, which is all this table joins on."
      },
      {
        "step_id": "daily_events_flagged",
        "sql": "select d.event_id, d.occurred_at, d.magnitude, d.place, d.latitude, d.longitude, d.depth_km, s.event_id is not null as is_significant, s.significance from daily_events d left join significant_events s on d.event_id = s.event_id",
        "description": "Keep every event of the last day and attach the significance score where the other feed carries one. A left join is what makes the unmatched events survive."
      }
    ]
  },
  "checks": [
    {
      "check_id": "one_row_per_event",
      "kind": "key_uniqueness",
      "enforcement": "required",
      "columns": [
        "event_id"
      ],
      "description": "One row per event; a duplicate here means the right relation was not unique on the join key."
    },
    {
      "check_id": "magnitude_mostly_present",
      "kind": "null_ceiling",
      "enforcement": "required",
      "columns": [
        "magnitude"
      ],
      "max_null_ppm": 50000
    },
    {
      "check_id": "magnitude_plausible",
      "kind": "value_range",
      "enforcement": "required",
      "columns": [
        "magnitude"
      ],
      "min_value": "-2",
      "max_value": "12"
    },
    {
      "check_id": "events_present",
      "kind": "row_expectation",
      "enforcement": "required",
      "min_rows": 1
    }
  ],
  "units": [
    {
      "column": "depth_km",
      "unit": "km"
    }
  ],
  "timezone": "UTC"
}

Part by part

Two sources, two Reader pins

Each source pins its own decode options, and the two differ. The left feed projects seven columns, the right projects two. Two sources reading the same publisher’s format may project different fields.

longitude, latitude and depth_km come from /geometry/coordinates/0, /1 and /2. A pointer that still selects an array refuses the decode, so /geometry/coordinates on its own is refused. Indexing into it selects a number.

depth_km is optional because a coordinate array may carry two members rather than three.

Transform, in three steps

The first two steps cast each feed on its own, before the join rather than inside it. A bad cast then fails in a step named after the feed it came from.

select event_id,
       to_timestamp(cast(occurred_at_ms as bigint) / 1000) as occurred_at,
       try_cast(magnitude as decimal(6, 3)) as magnitude,
       …
from all_day

The publisher’s time is epoch milliseconds, so the step divides it before making an instant. to_timestamp returns TIMESTAMP WITH TIME ZONE, which a column declared timestamp accepts.

The third step joins:

select d.event_id, d.occurred_at, d.magnitude, d.place, d.latitude, d.longitude, d.depth_km,
       s.event_id is not null as is_significant,
       s.significance
from daily_events d
left join significant_events s on d.event_id = s.event_id

A left join keeps the unmatched events. An inner join here answers a different question, “which significant events happened today”, with a table that looks complete.

The flag is s.event_id is not null, not s.significance is not null. A matched row with an absent score is still a match, and testing the score reports it as unmatched. The expression returns BOOLEAN, which the declared boolean column accepts.

Checks

Four checks.

key_uniqueness on event_id catches a join that fanned out. A right relation holding two rows for one identifier makes the left join emit two rows for that event. The table looks fine, every total computed from it is wrong, and nothing else in the document notices.

null_ceiling on magnitude at five per cent tolerates the events the USGS has not yet given a magnitude. value_range between "-2" and "12" bounds a magnitude, as decimal strings, because the document includes no fractional JSON number. row_expectation with min_rows: 1 catches a feed that returned an empty collection.

Each check kind has a shape. The acquire stage validates the check dialect, not registration, so a wrong shape costs a queued run.

`kind`	Columns	Also required
`key_uniqueness`	any number, falling back to `grain`	none
`null_ceiling`	exactly one	`max_null_ppm`
`value_range`	exactly one	both `min_value` and `max_value`
`continuity`	exactly one	`max_gap_seconds`
`row_expectation`	none, and naming one is refused	at least one of `min_rows`, `max_rows`

Units

Only depth_km gets one, as km. A magnitude is a number on a named scale rather than a physical quantity, and a latitude is a coordinate. units omits both rather than declaring them none.

Cadence

Both feeds show their current state. A normal refresh does not automatically re-read an unwindowed source; current refresh rules require an incremental window or return RESYNC_REQUIRED. Use an explicit full run or resync when you deliberately want to replace the current-state inputs. That leaves no history of the prior answer unless the resulting table is retained separately.

To ask what changed, declare a snapshot window on each source. The worker stamps every decoded row with the UTC day it was acquired on and keeps one partition per day. A transition becomes a lag() over that column:

select event_id, observed_date, is_significant,
       lag(is_significant) over (partition by event_id order by observed_date) as previous
from unioned
qualify is_significant is distinct from previous

is distinct from rather than <> keeps the first day. A row’s earliest covered day has no previous reading, and null <> true is null rather than true, which drops the row.

Register, run, inspect

mr-data dataset create --name "Today's earthquakes, flagged against the significant feed" --json
mr-data recipe recipe.json --json

mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --sample --max-rows 5000 --json
mr-data peek RUN_ID --json
mr-data checks RUN_ID --json
mr-data query RUN_ID "select is_significant, count(*) from run_table group by is_significant" --json

mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --full --json
mr-data diff RUN_A RUN_B --json

What the run reports

mr-data receipt RUN_ID --json carries one entry per source with its byte count, digest, Reader name and version, validated options digest and truncated flag.

mr-data checks RUN_ID --json reads the check results out of the receipt and exits non-zero when one did not pass. A check declared required never reaches this command, because a required check that fails ends the run with failure_code: CHECK_FAILED.

checks_scope says which rows the pass covered. version means every row of the table the run composed. Anything else means the rows that run read, not the ones it inherited. Never report “the checks passed” over an append without saying which rows they covered.

Traps

A row ceiling cuts each source separately, and the join then lies. Two sources cut at 2,000 rows each are two partial spans. Events present in the left feed lose their match because the right feed was truncated, and is_significant comes back false for rows that are significant. Measure the sources, then set the ceiling over the largest.

A left join whose right side is not unique on the key multiplies rows. Reduce the right side to one row per key in its own step, and keep the uniqueness check on the output.

try_cast on a join key hides a broken join. Use a bare cast on the key so the run stops, and try_cast on measurements so a bad cell becomes null.

Qualify a column a publisher spelled like an engine value. An unqualified current_date resolves to the engine’s value rather than the column, whatever else is in scope. Write all_day.current_date.

A step may name only its sources and earlier steps. The engine refuses a step_id equal to a source’s name rather than resolving it. The two would name one relation, and registration order would decide which one the SQL meant.