Mismatch of spark.catalog.tableExists and SHOW TABLES in streams

[i4ata]

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.

[balajij8]

@i4ata Its generally due to client side metadata caching within the Spark Connect architecture. Spark Connect checks a local client side cache instead of doing a fresh look up against the meta store when calling methods like spark.catalog.tableExists(). Since table creation in the foreachBatch executes on the server side directly against Unity Catalog, it did not trigger an automatic invalidation of client session's catalog cache. As a result, subsequent calls to tableExists() continue reading the stale client-side state.

Using SQL commands such as SHOW TABLES bypasses client cache sending a direct query to the server and reading the latest state from Unity Catalog in real time. Spark Connect’s decoupled client-server model is to minimize repetitive network trips for catalog operations as the general behavior

You can rely on server-side SQL operations such as SHOW TABLES or wrapping a DESCRIBE TABLE query in error handling to ensure using the live state of the Unity Catalog tables in jobs.