Your schemaHint isn't a cast: the Auto Loader mixed-type trap

from pyspark.sql.functions import col

# Illustrative files in /Volumes/catalog/schema/vol/raw/orders/:
# file-a.json: {"order_id": 101, "response_time": "450", "region": "eu-west"}
# file-b.json: {"order_id": 102, "response_time": 512,   "region": "eu-west"}

trap = (
    spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation",
            "/Volumes/catalog/schema/vol/checkpoints/orders/_schema")
    .option("cloudFiles.schemaHints",
            "response_time LONG, order_id LONG, region STRING")
    .option("cloudFiles.schemaEvolutionMode", "rescue")
    .option("rescuedDataColumn", "_rescued_data")
    .load("/Volumes/catalog/schema/vol/raw/orders/")
)

# Illustrative result:
# order 101 → response_time NULL, "450" quarantined in _rescued_data
# order 102 → response_time 512
# The hint did not convert "450". The hint is why the column is NULL.

from pyspark.sql.functions import col

# Read loose: hint the volatile field as STRING so mixed types
# ("450" and 450) both parse without rescue.
bronze = (
    spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation",
            "/Volumes/catalog/schema/vol/checkpoints/orders/_schema")
    .option("cloudFiles.schemaHints",
            "response_time STRING, order_id LONG, region STRING")
    .option("cloudFiles.schemaEvolutionMode", "rescue")
    .option("rescuedDataColumn", "_rescued_data")
    .load("/Volumes/catalog/schema/vol/raw/orders/")
)

# Cast strict: convert downstream, where a failed cast is YOUR
# error to catch — not a silent quarantine.
clean = bronze.withColumn("response_time_ms", col("response_time").cast("long"))

# The decision table (verified against Databricks docs):
#
# addNewColumns  (default when you DON'T pin a schema)
#   New column  → stream STOPS with UnknownFieldException.
#                 Schema location is updated; restart resumes with the
#                 new schema. Existing column types NEVER evolve.
#                 This is a signal, not a bug — run on Lakeflow Jobs
#                 so the restart is automatic.
#
# rescue         Stream never fails on schema changes; schema never
#                evolves. Every new column lands in _rescued_data.
#                Best for a bronze layer you want always-on.
#
# failOnNewColumns
#                Stream fails and stays down until you update the
#                provided schema or remove the offending file.
#                Use when a new column must be a human decision.
#
# none           Silent. New columns are ignored — not even rescued
#                unless rescuedDataColumn is set. The most dangerous
#                option, and the DEFAULT when you pin a schema.
bronze = (
    spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation",
            "/Volumes/catalog/schema/vol/checkpoints/orders/_schema")
    .option("cloudFiles.schemaEvolutionMode", "rescue")
    .option("rescuedDataColumn", "_rescued_data")
    .load("/Volumes/catalog/schema/vol/raw/orders/")
)

-- SAVE THIS: the daily rescued-data audit.
-- The blob carries the rescued columns AND the source file path,
-- so each row names its culprit file.
SELECT _rescued_data
FROM catalog.bronze.orders
WHERE _rescued_data IS NOT NULL;