"Cannot write incompatible data to table 'spark_catalog.default.store_returns':\n- Cannot safely cast 'sr_ticket_number': string to int"

 StructField("sr_ticket_number", IntegerType(), nullable=False),

StructField("sret_ticket_number", CharType(20)),

CREATE view srv as
SELECT d_date_sk sr_returned_date_sk
 ,t_time_sk sr_return_time_sk
 ,i_item_sk sr_item_sk
 ,c_customer_sk sr_customer_sk
 ,c_current_cdemo_sk sr_cdemo_sk
 ,c_current_hdemo_sk sr_hdemo_sk
 ,c_current_addr_sk sr_addr_sk
 ,s_store_sk sr_store_sk
 ,r_reason_sk sr_reason_sk
 ,sret_ticket_number sr_ticket_number
 ,sret_return_qty sr_return_quantity
 ,sret_return_amt sr_return_amt
 ,sret_return_tax sr_return_tax
 ,sret_return_amt + sret_return_tax sr_return_amt_inc_tax
 ,sret_return_fee sr_fee
 ,sret_return_ship_cost sr_return_ship_cost
 ,sret_refunded_cash sr_refunded_cash
 ,sret_reversed_charge sr_reversed_charge
 ,sret_store_credit sr_store_credit
 ,sret_return_amt+sret_return_tax+sret_return_fee
 -sret_refunded_cash-sret_reversed_charge-sret_store_credit sr_net_loss
FROM s_store_returns 
LEFT OUTER JOIN date_dim 
 ON (cast(sret_return_date as date) = d_date)
LEFT OUTER JOIN time_dim 
 ON (( cast(substr(sret_return_time,1,2) AS integer)*3600
 +cast(substr(sret_return_time,4,2) AS integer)*60
 +cast(substr(sret_return_time,7,2) AS integer)) = t_time)
LEFT OUTER JOIN item ON (sret_item_id = i_item_id)
LEFT OUTER JOIN customer ON (sret_customer_id = c_customer_id)
LEFT OUTER JOIN store ON (sret_store_id = s_store_id)
LEFT OUTER JOIN reason ON (sret_reason_id = r_reason_id)
WHERE i_rec_end_date IS NULL
 AND s_rec_end_date IS NULL;
------------------------------------------------
insert into store_returns (select * from srv);

2.2.2.1 Each column employs one of the following datatypes: a) Identifier means that the column shall be able to hold any key value generated for that column.

2.2.2.3 The implementation chosen by the test sponsor for a particular datatype definition shall be applied consistently to all the instances of that datatype definition in the schema, except for identifier columns, whose datatype may be selected to satisfy database scaling requirements.

delete from catalog_returns 
   where cr_order_number in 
      (select cs_order_number from 
             catalog_sales, date_dim 
                 where cs_sold_date_sk=d_date_sk and d_date between 'DATE1' and 'DATE2');

 java.lang.ClassCastException: org.apache.spark.sql.catalyst.plans.logical.Project cannot be cast to org.apache.spark.sql.execution.SparkPlan
	at scala.collection.immutable.List.map(List.scala:293)
	at org.apache.spark.sql.execution.SparkPlanInfo$.fromSparkPlan(SparkPlanInfo.scala:75)
	at org.apache.spark.sql.execution.SparkPlanInfo$.$anonfun$fromSparkPlan$3(SparkPlanInfo.scala:75)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)
	at scala.collection.Iterator.foreach(Iterator.scala:943)
	at scala.collection.Iterator.foreach$(Iterator.scala:943)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
	at scala.collection.IterableLike.foreach(IterableLike.scala:74)

cs_order_number = spark.sql("""
          select cs_order_number from catalog_sales, date_dim where cs_sold_date_sk=d_date_sk and d_date between '2000-05-20' and '2000-05-21'
          """).collect()

cs_order_number_lst = [x['cs_order_number'] for x in cs_order_number]

spark.sql(f"""
          delete from catalog_returns
          where cr_order_number in ({str(cs_order_number_lst)[1:-1]});
          """)

java.lang.ClassCastException: org.apache.spark.sql.catalyst.plans.logical.Project cannot be cast to org.apache.spark.sql.execution.SparkPlan