According to this https://iceberg.apache.org/docs/latest/spark-structured-streaming/ , we can stream from iceberg tables. I have ensured that my source table is Iceberg version 3, but no matter what I do, I get Iceberg does not streaming reads. Looking at V3 limitations here https://docs.databricks.com/gcp/en/iceberg/iceberg-v3?language=Managed+Iceberg+table#limitations , it does not mention that anywhere as a limitation. For version 2, one of limitation listed here https://docs.databricks.com/gcp/en/iceberg/ states that "Iceberg doesn't support change data feed. As a result, incremental processing isn't supported when reading managed Iceberg tables as a source for:"

But my destination is another iceberg table (also tried delta)

Error: data source iceberg does not support streamed reading

-- create source table
spark.sql("""CREATE OR REPLACE TABLE engagement.`sandbox-client-feedback`.dummy_iceberg_source_stream (c1 INT)
USING iceberg
TBLPROPERTIES ('format-version' = 3)""")

insert into engagement.`sandbox-client-feedback`.dummy_iceberg_source_stream values (2);

-- create destination table
spark.sql("""CREATE OR REPLACE TABLE engagement.`sandbox-client-feedback`.dummy_iceberg_destination (c1 INT)
USING iceberg""")

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import sys
from databricks.connect import DatabricksSession
import logging
import datetime


def stream_iceberg_to_iceberg(
    spark,
    source_table,
    target_table,
    checkpoint_location,
    trigger_interval_seconds=60,
    stream_start_timestamp=None
):
    """
    Stream data from source Iceberg table to target Iceberg table
    
    Args:
        spark: SparkSession instance
        source_table: Source Iceberg table name (e.g., 'catalog.database.source_table')
        target_table: Target Iceberg table name (e.g., 'catalog.database.target_table')
        checkpoint_location: Path for checkpoint files
        trigger_interval_seconds: Processing trigger interval in seconds (default: 60)
        stream_start_timestamp: Optional timestamp in milliseconds to start streaming from
    """
    
    # Configure streaming read from source Iceberg table
    read_stream_builder = (spark.readStream 
        .format("iceberg")
    )
    
    # Optional: Start from a specific timestamp
    if stream_start_timestamp:
        read_stream_builder = (read_stream_builder 
            .option("stream-from-timestamp", str(stream_start_timestamp))
        )
    
    # Optional: Configure streaming behavior
    read_stream_builder = (read_stream_builder 
        .option("streaming-skip-overwrite-snapshots", "true") 
        .option("streaming-skip-delete-snapshots", "true")
    )
    
    # Read the stream
    source_stream = read_stream_builder.load(source_table)
    
    # Optional: Apply transformations
    # Example: Add processing timestamp and filter
    transformed_stream = (source_stream 
        .withColumn("processing_time", current_timestamp()) 
        .withColumn("processing_date", current_date())
    )
    # Add your custom transformations here
    # .filter(col("some_column").isNotNull())
    # .select("col1", "col2", "col3")
    
    # Write stream to target Iceberg table
    query = (transformed_stream.writeStream 
        .format("iceberg") 
        .outputMode("append") 
        .trigger(availableNow=True) 
        .option("checkpointLocation", checkpoint_location) 
        .option("fanout-enabled", "true") 
        .toTable(target_table)
    )
    
    return query



def main():
    """
    Main execution function
    """
    # Configuration
    SOURCE_TABLE = "engagement.`sandbox-client-feedback`.dummy_iceberg_source_stream"
    TARGET_TABLE = "engagement.`sandbox-client-feedback`.dummy_iceberg_destination"
    CHECKPOINT_LOCATION = "dbfs:/dbfs/tmp/sum_streaming_iceberg_test_stream"
    
    start_time = int(datetime.datetime(2026, 1, 1).timestamp() * 1000)
    START_TIMESTAMP = start_time  # Set to None to start from latest snapshot
    
    # Create Spark session
    spark = DatabricksSession.builder.getOrCreate()
    spark.sparkContext.setLogLevel("INFO")
    
    logging.info(f"Starting streaming from {SOURCE_TABLE} to {TARGET_TABLE}")
    logging.info(f"Checkpoint location: {CHECKPOINT_LOCATION}")
    
    try:
        # Start streaming query
        query = stream_iceberg_to_iceberg(
            spark=spark,
            source_table=SOURCE_TABLE,
            target_table=TARGET_TABLE,
            checkpoint_location=CHECKPOINT_LOCATION,
            stream_start_timestamp=START_TIMESTAMP
        )
        
        logging.info(f"Streaming query started: {query.name}")
        logging.info("Press Ctrl+C to stop the streaming query")
        
        query.awaitTermination()
        
    except Exception as e:
        print(f"Error in streaming job: {str(e)}")


if __name__ == "__main__":
    main()

TBLPROPERTIES ('format-version' = 3);