✅ Source Code used in sample here:

# Databricks notebook source
################################################################################################
# 1) Function to create sample DataFrame and overwrite results in "people_table" Delta Table 
################################################################################################
from pyspark.sql import DataFrame
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

def create_or_reset_sample_df() -> DataFrame:
    schema = StructType([
        StructField("Id", IntegerType(), False),
        StructField("Name", StringType(), True),
        StructField("Surname", StringType(), True),
        StructField("Age", IntegerType(), True)
    ])
    data = [
        (1, "Alice", "Smith", 25),
        (2, "Bob", "Jordan", 30),
        (3, "Charlie", "Durant", 35)
    ]
    df = spark.createDataFrame(data, schema)
    df.write.mode("overwrite").saveAsTable("people_table")
    return df

def print_separator():
    print("=" * 100)

create_or_reset_sample_df().show()


# COMMAND ----------

################################################################################################
# 2) Inspect plans WITHOUT caching 
################################################################################################
create_or_reset_sample_df()

# Get Dataframe from Delta Table
df = spark.sql("select * from people_table where Age >= 18")

# Action 1
print(f"Total People: {df.count()}\n")
df.explain()

# Filter and Action 2
name1 = 'Alice'
df1 = df.filter(f"Name = '{name1}' ")
print(f"Number of people with name '{name1}': {df1.count()}\n")
df1.explain()

# Filter and Action 3
name2 = 'Bob'
df2 = df.filter(f"Name = '{name2}'")
print(f"Number of people with name '{name2}': {df2.count()}\n")
df2.explain()

print_separator()
print("CONCLUSION: All DFs (df, df1, df2) are created based on scans over parquet files from delta table")
print("IMPORTANT: Up to 3 different queries are performed over original delta table due to Lazy Evaluation")
print_separator()


# COMMAND ----------


################################################################################################
# 3) Inspect Plans WITHOUT caching BUT replaced 'df' by original Spark SQL Query  
################################################################################################
create_or_reset_sample_df()

# Get Dataframe from Delta Table
df = spark.sql("select * from people_table where Age >= 18")

# Action 1
print(f"Total People: {df.count()}\n")
df.explain()

# Filter and Action 2 - Reference to 'df' replaced by original query
name1 = 'Alice'
df1 = spark.sql("select * from people_table where Age >= 18").filter(f"Name = '{name1}' ")
print(f"Number of people with name '{name1}': {df1.count()}\n")
df1.explain()

# Filter and Action 3 - Reference to 'df' replaced by original query
name2 = 'Bob'
df2 = spark.sql("select * from people_table where Age >= 18").filter(f"Name = '{name2}'")
print(f"Number of people with name '{name2}': {df2.count()}\n")
df2.explain()

print_separator()
print("CONCLUSION: We get exactly same plans as SAMPLE 2")
print_separator()


# COMMAND ----------


################################################################################################
# 4) Inspect plans WITH caching (.persist() or .cache() methods)
################################################################################################
create_or_reset_sample_df()

# Get Dataframe from Delta Table
df = spark.sql("select * from people_table where Age >= 18")

# Action 1
print(f"Total People: {df.count()}\n")
df.explain()

# Persist Dataframe
df.persist()

# Filter and Action 2 - 'df' already cached / persisted
name1 = 'Alice'
df1 = df.filter(f"Name = '{name1}' ")
print(f"Number of people with name '{name1}': {df1.count()}\n")
df1.explain()

# Filter and Action 3 - 'df' already cached / persisted
name2 = 'Bob'
df2 = df.filter(f"Name = '{name2}'")
print(f"Number of people with name '{name2}': {df2.count()}\n")
df2.explain()

# Unpersist Dataframe
df.unpersist()

print_separator()
print(f"CONCLUSION: After caching ('persist()') rest of DFs are created based on cached 'df'")
print(f"IMPORTANT: Only one query is performed over parquet files from original delta table")
print_separator()




# COMMAND ----------

########################################################################################################################
# 5) WHAT IF underlying Delta Table is updated? (Cluster Runtime: 16.4 LTS)
# A) Update underlying Delta Table with rows NOT MATCHING 'where' clauses used to create DFs
########################################################################################################################
CreateOrResetSampleDF()

# Get DataFrame from Delta Table
df = spark.sql("select * from people_table where Age >= 18")

# Set DataFrame to be persisted
df.persist()

# Insert new record in underlying Delta Table over which DataFrame was cached.
# New record (4,'Bob', 'Carrey', 15) does not match 'where Age >= 18' filter as 'Age = 15' is used
total_people_before_insert = df.count()
print(f"Total People in Cached DataFrame BEFORE inserting new record in underlying Delta Table: {total_people_before_insert}")
spark.sql("insert into people_table values(4,'Bob', 'Carrey', 15)") 
total_people_after_insert = df.count()
print(f"Total People in Cached DataFrame AFTER inserting new record in underlying Delta Table: {total_people_after_insert}")
print(f"Inserted Record (4,'Bob', 'Carrey', 15) in underlying Delta Table IS NOT present in Cached DataFrame as 15 does not match 'Age >= 18' where clause")
df.show()
# As inserted record does not match 'where' filter, we do not expect cache gets invalidated / refreshed remaining "stale"
assert(total_people_before_insert == total_people_after_insert) 

# Filter and action for name 'Bob'
name = 'Bob'
df2 = df.filter(f"Name = '{name}'")
total_people_with_name = df2.count()
print(f"People with name '{name}': {df2.count()}") 
assert(total_people_with_name == 1)

# Unpersist DataFrame
df.unpersist()

print_separator()
print(f"CONCLUSION: If new inserted record in underlying Delta Table does not match 'where' clause as from which Cached DataFrame was generated -> CACHE is NOT REFRESHED")
print_separator()

# COMMAND ----------



########################################################################################################################
# 5) WHAT IF underlying Delta Table is updated? (Cluster Runtime: 16.4 LTS)
# B) Update underlying Delta Table with rows MATCHING 'where' clauses used to create DFs
########################################################################################################################
CreateOrResetSampleDF()

# Get DataFrame from Delta Table
df = spark.sql("select * from people_table where Age >= 18")
# Set DataFrame to be persisted
df.persist()

# Insert new record in underlying Delta Table over which DataFrame was cached.
# New record (4,'Bob', 'Wilkins', 45) match 'where Age >= 18' filter as 'Age = 45' is used
total_people_before_insert = df.count()
print(f"Total People in Cached DataFrame BEFORE inserting new record in underlying Delta Table: {total_people_before_insert}")
spark.sql("insert into people_table values(4,'Bob', 'Wilkins', 45)") 
total_people_after_insert = df.count()
print(f"Total People in Cached DataFrame AFTER inserting new record in underlying Delta Table: {total_people_after_insert}")
print(f"New inserted record (4,'Bob', 'Wilkins', 45) into underlying Delta Table IS present in Cached DataFrame as 45 matches 'Age >= 18' where clause")
df.show()
# As inserted record match 'where' filter, we expect cache gets invalidated / refreshed.
assert(total_people_before_insert < total_people_after_insert) 

# Filter and Action for name 'Bob'
name = 'Bob'
df2 = df.filter(f"Name = '{name}'")
total_people_with_name = df2.count()
print(f"People with name '{name}': {df2.count()}") 
assert(total_people_with_name == 2) # We expect new record to appear in cached DataFrame as a result of cache invalidation

# Unpersist DataFrame
df.unpersist()

print_separator()
print("CONCLUSION: If new inserted record in underlying Delta Table MATCHES 'where' clause as from which Cached DF was generated -> CACHE is INVALIDATED / REFRESHED.")
print("IMPORTANT: Cached DataFrame is only refreshed if Delta Table update (INSERT, DELETE, UPDATE, MERGE) is performed IN SAME CLUSTER where DF was cached.")
print("IMPORTANT: All of this is only valid if Cluster Runtime Version is >= 12.2 LTS. Otherwise, Cached DataFrame remains stale.")
print_separator()