add cleanup stmt when a data prodsuct is persisted to fix ingestion worker leak