Consider this minimal example of a streaming workflow:

import os
from pyspark.sql import functions as F, DataFrame
from typing import Literal

# Change these if you want to replicate
CATALOG, SCHEMA = 'partner_demo_catalog', 'ad_sales_demo'

spark.sql(f'use {CATALOG}.{SCHEMA}')
log_file = os.path.join(os.getcwd(), 'debug.txt')

def exists(table_name: str, method: Literal['spark', 'sql']) -> bool:

    # Bugged
    if method == 'spark':
        return spark.catalog.tableExists(table_name)
    
    # Working fine
    return spark.sql('show tables').filter(F.col('tableName') == table_name).count() > 0

def foreach_batch(batch_df: DataFrame, batch_id: int) -> None:

    # Write logs to a text file for simplicity
    with open(log_file, 'a') as f:
        
        f.write(f'batch_id: {batch_id}\n')
        
        # Check if the table is in the catalog
        if exists('t', 'sql'): f.write('Before stream, SQL: IN\n')
        else: f.write('Before stream, SQL: OUT\n')
        if exists('t', 'spark'): f.write('Before stream, Spark: IN!\n')
        else: f.write('Before stream, Spark: OUT!\n')

        # Write the table to the catalog
        batch_df.write.format("delta").mode('overwrite').saveAsTable('t')
        
        # Check if the table is in the catalog again
        if exists('t', 'sql'): f.write('After stream, SQL: IN\n')
        else: f.write('After stream, SQL: OUT\n')
        if exists('t', 'spark'): f.write('After stream, Spark: IN!\n')
        else: f.write('After stream, Spark: OUT!\n')
        
        f.write('\n')

def test_stream(df: str, i: int) -> None:
    (
        spark
        .readStream.table(df)
        .writeStream.foreachBatch(foreach_batch)

        # Pass a different checkpoint location just to trigger the stream again with no changes
        .option("checkpointLocation", f'{os.getcwd()}/checkpoints_test_{i}')

        .trigger(availableNow=True)
        .start()
        .awaitTermination()
    )

spark.sql('drop table if exists t')
spark.sql('drop table if exists src')

df_src=spark.createDataFrame([('Alice', 1)], ['name', 'age'])
df_src.write.format("delta").mode('overwrite').saveAsTable('src')

test_stream('src', 0)
test_stream('src', 1)

As it can be seen, the script creates a toy table `src`, from which a stream is created and triggered twice. Inside `foreachBatch` I create a new toy table `t`. I check whether spark can see the new table in two ways: `spark.catalog.tableExists()` and Spark SQL's `SHOW TABLES`. For simplicity I am logging to a regular .txt file. The contents of it after running the script (using serverless compute and unity catalog) are the following:

batch_id: 0
Before stream, SQL: OUT
Before stream, Spark: OUT!
After stream, SQL: IN
After stream, Spark: IN!

batch_id: 0
Before stream, SQL: IN
Before stream, Spark: OUT! # Key line here
After stream, SQL: IN
After stream, Spark: IN!

This means that `spark.catalog.tableExists()` fails to see the table `t` even though it clearly exists in the catalog. Could anyone clarify what is going on and whether that behavior is intended? According to Genie, that is "a bug in the Spark Connect foreachBatch implementation".

I am using different checkpoint paths just so that I can trigger the full stream again without changing the source table. Still, using the same path and updating the table between streaming runs does not change the behavior. Using `batch_df`'s spark session has no effect either.