Hey,
we are experiencing issues with writing to external tables when using the Unity Catalogue and Row Level Security.
As soon as we stop using the serverless compute instance, we receive the following error for writing (Overwrite, append and upsert):
ErrorClass=INVALID_PARAMETER_VALUE.PATH_BASED_ACCESS_NOT_SUPPORTED_FOR_TABLES_WITH_ROW_COLUMN_ACCESS_POLICIES] Path-based access to table catalog_x.schema_y.rls_dbx_test with row filter or column mask not supported.
RLS and write to external table:
- works with a serverless cluster
- does not work with LTS 13.4
- does not work with LTS 15.4
- does not work with LTS 16.4
Is there any way to make the RLS with an external table and a dedicated cluster work?
Using dynamic views as an alternative introduces a lot more work and user complexity.
This is our test setup:
CREATE OR REPLACE FUNCTION us_filter(rls_entity STRING)
RETURN IF(CURRENT_USER()=='user_x', true, rls_entity='us');
ALTER TABLE catalog_x.schema_y.rls_dbx_test SET ROW FILTER us_filter ON (rls_entity);
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from delta.tables import DeltaTable
# Define catalog and schema
catalog = "catalog_x"
schema_name = "schema_y"
table_name = "rls_dbx_test"
### CASE OVERWRITE ###
# Sample data
data = [
("logger_01", 1909, "us"),
("logger_02", 1909, "row"),
("logger_03", 1909, "us"),
("logger_04", 1909, "bra")
]
# Define schema
schema = StructType([
StructField("data_object_name", StringType(), True),
StructField("year", IntegerType(), True),
StructField("rls_entity", StringType(), True)
])
# Create DataFrame
df = spark.createDataFrame(data, schema)
# Set current catalog and schema
spark.sql(f"USE CATALOG {catalog}")
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {schema_name}")
spark.sql(f"USE {schema_name}")
spark.sql(f"""
CREATE TABLE IF NOT EXISTS `{catalog}`.`{schema_name}`.`{table_name}`
USING DELTA
LOCATION '{"abfss://placeholder_x@placeholder_y}'
""")
# Write as Delta table
df.write.format("delta").mode("overwrite").option("mergeSchema", "true").saveAsTable(f"{catalog}.{schema_name}.{table_name}")
### CASE APPEND ###
data = [
("logger_05", 1909, "us"),
("logger_06", 1909, "row"), ]
# Define schema
schema = StructType([
StructField("data_object_name", StringType(), True),
StructField("year", IntegerType(), True),
StructField("rls_entity", StringType(), True)
])
# Create DataFrame
df = spark.createDataFrame(data, schema)
# Write as Delta table
df.write.format("delta").mode("append").saveAsTable(f"{catalog}.{schema_name}.{table_name}")
### CASE UPSERT ###
data = [
("logger_04", 1909, "ch"),
("logger_07", 1909, "us"),
("logger_08", 1909, "eu"), ]
# Define schema
schema = StructType([
StructField("data_object_name", StringType(), True),
StructField("year", IntegerType(), True),
StructField("rls_entity", StringType(), True)
])
# Create DataFrame
df = spark.createDataFrame(data, schema)
full_table_name = f"{catalog}.{schema_name}.{table_name}"
delta_table = DeltaTable.forName(spark, full_table_name)
delta_table.alias("target").merge(
source=df.alias("source"),
condition="target.data_object_name == source.data_object_name"
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
Thanks
