(Dear Moderators, why do you remove this question? It is a genuine question. Do not, please. )

We have a continuous dlt pipeline that has tables updating every minute and partitioned by "partition_key" column. Table is 4 TB. Has 16k files. Sometimes we are getting below error caused by predictive optimization conflict. After exception, pipeline restarts itself and continues operation but we want to avoid this and we also don't want to disable predictive optimization. What do you recommend?

Error:

[DELTA_METADATA_CHANGED] MetadataChangedException: The metadata of the Delta table has been changed by a concurrent update. Please try the operation again.
Conflicting commit: {"timestamp":1771277469269,"userId":"******","userName":"*********","operation":"SET TBLPROPERTIES","operationParameters":{"properties":{"delta.workloadBasedColumns.optimizerStatistics":"`partition_key`"}},"job":{"jobId":"*****","jobName":"Predictive Optimization Job-*****","jobRunId":"*****","runId":"*****","jobOwnerId":"*****","triggerType":"manual"},"clusterId":"*****","readVersion":247493,"isolationLevel":"WriteSerializable","isBlindAppend":true,"operationMetrics":{},"tags":{"maintenance":"true","delta.rowTracking.preserved":"true"},"engineInfo":"Databricks-Runtime/18.0.x-aarch64-photon-scala2.13","txnId":"******"}

Code is this:

@DP.table(
    name="***",
    comment="***",  
    partition_cols=["partition_key"],
    table_properties={
        "description": "***",
        # thought it can solve the problem, appearently it did not
        "delta.workloadBasedColumns.optimizerStatistics": "partition_key"
    }
)
def ***():
    df = (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        # we intentionally do not use schema since this stream has overall 700 columns and constantly evolving.
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.useManagedFileEvents","true")
        .option("cloudFiles.schemaEvolutionMode", "rescue")
        .load(f"***")
    )