A data engineer has the following streaming query with a blank: spark . readStream . table ( "orders_cleaned" ) ____________________________ . groupBy ( "order_timestamp" , "author" ) . agg ( count ( "order_id" ). alias ( "orders_count" ), avg ( "quantity" ). alias ( "avg_quantity" )) . writeStream . option ( "checkpointLocation" , "dbfs:/path/checkpoint" ) . table ( "orders_stats" ) For handling late-arriving data, they want to maintain the streaming state information for 30 minutes. Which option correctly fills in the blank to meet this requirement ?