influxdb(seed): filter the colliding "source" field before pivot

Per-day chunking from 106d551 ran successfully for 39 minutes (got
through 2018-01-01 to 2024-06-03) before hitting:

  runtime error @3:13-3:83: pivot: value "source" appears in a column
  key column, but a column named "source" already exists; consider
  renaming "source" to something else before pivoting

HA writes `source` as both a tag (integration name, e.g. "Home
Assistant") AND, on some entities, as a field (entity attribute) on
the same measurement. pivot() turns _field values into column names,
so when it processes a row with _field=source it tries to add a
"source" column — but the existing tag-derived "source" column is
already there. Pivot errors out.

Drop the field via `filter(fn: (r) => r._field != "source")` before
pivot. Tag `source` is preserved (more useful — present on every
point, drives Grafana datasource filtering); the field `source` is
lost (rare, unique-per-entity attribute). Net data shape stays
queryable for HA's downstream consumers.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

kubernetes/seed/influxdb/seed-job.yaml

@@ -131,12 +131,23 @@ spec:
               while [[ "$CUR" < "$END" ]]; do
                 NEXT=$(date -u -d "$CUR +1 day" +%Y-%m-%dT00:00:00Z)
                 log "Streaming $CUR → $NEXT ..."
+                # Filter out the `_field == "source"` rows: HA writes
+                # `source` as both a tag (which integration emitted the
+                # point) AND a field (some entity attributes) on the
+                # same measurement. pivot() cannot materialise a field-
+                # derived column whose name collides with an existing
+                # tag column, so it errors out at the first row that
+                # has _field=source. Dropping that one field keeps the
+                # `source` tag intact (the more useful of the two) at
+                # the cost of losing the entity-attribute field, which
+                # is rarely populated and unique per entity.
                 influx query --raw \
                     --host "$LEGACY_HOST" \
                     --token "$LEGACY_TOKEN" \
                     --org "$LEGACY_ORG" \
                     "from(bucket: \"$LEGACY_BUCKET\")
                        |> range(start: $CUR, stop: $NEXT)
+                       |> filter(fn: (r) => r._field != \"source\")
                        |> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")
                        |> drop(columns: [\"_start\", \"_stop\"])" \
                   | influx write \