The core issue with your prototype is that the file discovery step is being duplicated: Auto Loader already does incremental, stateful file discovery internally (and can use cloud-native file notifications instead of directory listing), but your dbutils.fs.ls loop is a separate manual full listing that doesn't scale and has to be re-run to find new objecttypes. The fix is to stop doing objecttype discovery before the stream and instead do it inside a single stream.

Recommended pattern: one Auto Loader stream landing everything, fan-out in foreachBatch

One Auto Loader stream reads the whole directory, not per-objecttype:
python
raw = (spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("cloudFiles.useNotifications", "true") # avoid repeated full listing as volume grows
.option("cloudFiles.schemaLocation", schema_loc)
.option("header", "true")
.schema(generic_string_schema) # or land everything as strings — see below
.load(source_path)
)

Since each objecttype has a different real schema, don't try to get Auto Loader to infer one unified schema across all of them — that's what was pushing you toward per-objecttype streams in the first place. Instead, either:

Land every column as string (schema inference will do this by default for CSV if you don't force it otherwise) plus _rescued_data, or
Land the whole row as a single VARIANT column — Databricks explicitly supports this pattern for exactly this kind of "many shapes landing in one place" ingestion, and it's schema-agnostic by construction.
Extract objecttype from the filename inside the stream, using the built-in _metadata column rather than a separate listing pass:
python
from pyspark.sql.functions import regexp_extract, col

raw = raw.withColumn(
"objecttype",
regexp_extract(col("_metadata.file_name"), r"^([^_]+)_", 1)
)

This means new objecttypes are picked up automatically the moment their files land — no code change, no stream restart, no separate scan.

Fan out to per-objecttype tables in foreachBatch, dynamically, based on whatever objecttypes actually appear in each micro-batch:
python
def route_batch(batch_df, batch_id):
for obj_type in [r.objecttype for r in batch_df.select("objecttype").distinct().collect()]:
subset = batch_df.filter(col("objecttype") == obj_type)
target_table = f"bronze.{obj_type}"
(subset.write.format("delta")
.mode("append")
.option("mergeSchema", "true")
.saveAsTable(target_table)) # creates the table on first appearance

raw.writeStream.foreachBatch(route_batch).option("checkpointLocation", checkpoint_loc).start()

Why this scales where your prototype doesn't:

One stream, one checkpoint — no per-objecttype glob patterns to maintain or restart as new types appear.
No separate full-directory listing to enumerate objecttypes — that work is now done incrementally by Auto Loader's own file-tracking state, and cheaply, per micro-batch, only on files that already got pulled in.
New objecttypes just work — the first time a new <objecttype>_*.csv.gz lands, foreachBatch sees a new distinct value and creates the table, no code deploy required.

One thing to decide up front: if downstream consumers need real typed columns (not everything as strings/VARIANT), you'll want a lightweight per-objecttype schema registry (even just a small control Delta table: objecttype -> expected_schema_json) that the foreachBatch function looks up to cast columns properly before writing — otherwise you're pushing the "what's the real schema" problem one layer downstream instead of solving it. That's a reasonable v2; landing as strings/VARIANT + rescue column first, then adding typed casting once you've stabilized the objecttype list, keeps the initial rollout simple.