Convert Teradata BTEQ and SQL to PySpark on Databricks

.LOGON tdserver/dbc,dbc;

COLLECT STATISTICS ON customer_orders
  COLUMN (customer_id);

CREATE VOLATILE TABLE latest_orders AS (
  SELECT customer_id, order_date, amount,
         order_status
  FROM customer_orders
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY order_date DESC
  ) = 1
) WITH DATA ON COMMIT PRESERVE ROWS;

MERGE INTO customer_summary tgt
USING latest_orders src
ON tgt.customer_id = src.customer_id
WHEN MATCHED THEN UPDATE SET
  last_order_date = src.order_date,
  last_amount     = src.amount,
  status          = src.order_status
WHEN NOT MATCHED THEN INSERT VALUES (
  src.customer_id, src.order_date,
  src.amount, src.order_status
);

.LOGOFF;
.QUIT;

# BTEQ → PySpark on Databricks
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from delta.tables import DeltaTable

spark.sql("ANALYZE TABLE customer_orders
  COMPUTE STATISTICS FOR COLUMNS customer_id")

w = Window.partitionBy("customer_id") \
          .orderBy(F.col("order_date").desc())

latest_orders = (
    spark.read.table("customer_orders")
    .withColumn("_rn", F.row_number().over(w))
    .filter(F.col("_rn") == 1)
    .drop("_rn")
)

tgt = DeltaTable.forName(spark, "customer_summary")
tgt.alias("tgt").merge(
    latest_orders.alias("src"),
    "tgt.customer_id = src.customer_id"
).whenMatchedUpdate(set={
    "last_order_date": "src.order_date",
    "last_amount":     "src.amount",
    "status":          "src.order_status"
}).whenNotMatchedInsertAll().execute()