I am using the Feature Engineering client when writing to a time series feature table. Then I have cried two data bricks jobs with the below code. I am running with different run_dates (e.g. '2016-01-07' and '2016-01-08'). When they run concurrently, one of the jobs throws: 

ConcurrentAppendException: Files were added to the root of the table by a concurrent update. Please try the operation again.

I think this is happening because of the merge operation that the feature engineering client is using. One solution that I have read is Avoid conflicts using partitioning and disjoint command conditions. However, I cannot partition the table because my table is a Time Series Feature table, which should not be partitioned according to this doc:

A time series feature table must have one timestamp key and cannot have any partition columns. The timestamp key column must be of TimestampType or DateType.

https://docs.databricks.com/en/machine-learning/feature-store/time-series.html#create-a-time-series-...

In addition, the code snippet solution described by using the below code cannot be used as well since the write_table handles the merge operation, and is abstracted. 

"s.user_id = t.user_id AND s.date = t.date AND s.country = t.country AND t.date = '" + <date> + "' AND t.country = '" + <country> + "'")

How can I write concurrently to a time series feature table then?

# Databricks notebook source
from pyspark.sql import SparkSession, DataFrame
import pyspark.sql.functions as F
from pyspark.sql.types import FloatType, IntegerType, StringType

from delta.tables import DeltaTable

from databricks.feature_engineering import FeatureEngineeringClient

def get_inputs(spark, run_date: str) -> list[DataFrame]:
    return {
        "silver_taxi_trips": (
            spark.read.load(
                path="/databricks-datasets/nyctaxi-with-zipcodes/subsampled", 
                format="delta"
            )
            .filter(
                F.col("tpep_pickup_datetime").cast("date") <= run_date
            )
        ),
    }


def main(spark, run_date: str) -> None:
    inputs = get_inputs(spark, run_date)

    df_taxi_trips = inputs["silver_taxi_trips"]

    df_pickupzip_features = (
        df_taxi_trips.groupBy(
            "pickup_zip", F.hour("tpep_pickup_datetime").alias("hour_slot")
        )
        .agg(
            F.mean("fare_amount").alias("mean_fare_hourly_pickup_zip"),
            F.count("*").alias("count_trips_hourly_pickup_zip"),
        )
        .select(
            F.col("pickup_zip").alias("zip"),
            F.col("hour_slot"),
            F.col("mean_fare_hourly_pickup_zip").cast(FloatType()),
            F.col("count_trips_hourly_pickup_zip").cast(IntegerType()),
        )
    )

    df_pickupzip_features = df_pickupzip_features.withColumn('run_date', F.to_date(F.lit(run_date)))
    df_pickupzip_features = df_pickupzip_features.withColumn('writing_ts', F.current_timestamp())

    fe = FeatureEngineeringClient()
    fe.write_table(
        name="dbc_mlops_premium.taxi_example.trip_pickup_features",
        df=df_pickupzip_features
    )

the table was created with:

%sql
CREATE TABLE IF NOT EXISTS dbc_mlops_premium.taxi_example.trip_pickup_features(
  -- entity
  zip INT NOT NULL,
  hour_slot INT NOT NULL,
  -- features
  mean_fare_hourly_pickup_zip FLOAT,
  count_trips_hourly_pickup_zip INT,
  -- time metadata
  run_date DATE NOT NULL,
  writing_ts TIMESTAMP NOT NULL,
  CONSTRAINT trip_pickup_features_pk PRIMARY KEY (zip, hour_slot, run_date TIMESERIES)
)
COMMENT "Taxi pick up features";