feat: Batch OPA column mask calls (#827)

* add batchColumnMasks OPA endpoint, add schema definition, adjust operator to use new batched endpoint

* add/adjust tests, add crd field to disable column masking

* incorporate decision feedback

* adjust tests

* make `opa` within `authorization` a required enum variant

* pre-commit fixes

* refactor code, add documenation

* add changelog entries

* cleanup test and unnecessary function

* fix docs

* remove version 451 test handling

* Update rust/operator-binary/src/crd/mod.rs

Co-authored-by: Siegfried Weber <mail@siegfriedweber.net>

* Update tests/templates/kuttl/opa-authorization/trino_rules/trino/verification.rego

Co-authored-by: Siegfried Weber <mail@siegfriedweber.net>

* Update tests/templates/kuttl/opa-authorization/trino_rules/trino/verification.rego

Co-authored-by: Siegfried Weber <mail@siegfriedweber.net>

* remove unneeded line wrap

* fix docs

* update config overrides list

* fix rego test

* disable columnmasking in example

* pre-commit fixes

---------

Co-authored-by: Siegfried Weber <mail@siegfriedweber.net>

Author: xeniape

CHANGELOG.md

@@ -9,13 +9,19 @@ All notable changes to this project will be documented in this file.
 - Support objectOverrides using `.spec.objectOverrides`.
   See [objectOverrides concepts page](https://docs.stackable.tech/home/nightly/concepts/overrides/#object-overrides) for details ([#831]).
 - Enable the [restart-controller](https://docs.stackable.tech/home/nightly/commons-operator/restarter/), so that the Pods are automatically restarted on config changes ([#833]).
+- Add `enabledColumnMasking` field to `opa` configuration in `authorization` ([#827]).
+- Support batched column masks in Rego rules ([#827]).
 
 ### Changed
 
 - Pin k8s-openapi to `0.26.0` ([#831]).
+- BREAKING: The field `opa` in `authorization` is now a mandatory enum variant instead of being optional ([#827]).
+- BREAKING: The operator no longer sets `opa.policy.column-masking-uri` in `access-control.properties` but
+  `opa.policy.batch-column-masking-uri` instead, allowing Trino to fetch multiple column masks in a single request ([#827]).
 
 [#831]: https://github.com/stackabletech/trino-operator/pull/831
 [#833]: https://github.com/stackabletech/trino-operator/pull/833
+[#827]: https://github.com/stackabletech/trino-operator/pull/827
 
 ## [25.11.0] - 2025-11-07
 

deploy/helm/trino-operator/crds/crds.yaml

@@ -70,20 +70,26 @@ spec:
                         Authorization options for Trino.
                         Learn more in the [Trino authorization usage guide](https://docs.stackable.tech/home/nightly/trino/usage-guide/security#authorization).
                       nullable: true
+                      oneOf:
+                        - required:
+                            - opa
                       properties:
                         opa:
                           description: |-
                             Configure the OPA stacklet [discovery ConfigMap](https://docs.stackable.tech/home/nightly/concepts/service_discovery)
                             and the name of the Rego package containing your authorization rules.
                             Consult the [OPA authorization documentation](https://docs.stackable.tech/home/nightly/concepts/opa)
                             to learn how to deploy Rego authorization rules with OPA.
-                          nullable: true
                           properties:
                             configMapName:
                               description: |-
                                 The [discovery ConfigMap](https://docs.stackable.tech/home/nightly/concepts/service_discovery)
                                 for the OPA stacklet that should be used for authorization requests.
                               type: string
+                            enableColumnMasking:
+                              default: true
+                              description: Whether to set the OPA batched column masking URI for Trino queries; defaults to true
+                              type: boolean
                             package:
                               description: The name of the Rego package containing the Rego rules for the product.
                               nullable: true

docs/modules/trino/pages/reference/security.adoc

@@ -0,0 +1,37 @@
+= Security
+
+== Authorization
+
+=== OPA
+
+==== Column masking
+
+===== CRD configuration
+
+[source,yaml]
+----
+apiVersion: trino.stackable.tech/v1alpha1
+kind: TrinoCluster
+spec:
+  clusterConfig:
+    authorization:
+      opa:
+        enableColumnMasking: true # default
+----
+
+===== Result
+
+In the `access-control.properties` file, the following value is set when `enableColumnMasking` is set to `true`:
+
+[source]
+----
+opa.policy.batch-column-masking-uri=<opa-url>/v1/data/<package>/batchColumnMasks # <1> <2>
+----
+
+<1> `<opa-url>` is read from the OPA discovery ConfigMap
+<2> `<package>` is read from `spec.clusterConfig.authorization.opa.package` if set, otherwise defaults to the TrinoCluster name
+
+===== Considerations
+
+The default setting for `enableColumnMasking` assumes a `batchColumnMasks` rule is defined in the Rego rules for the TrinoCluster.
+If no such rule is defined, Trino queries that utilize the column masking endpoint will fail.

docs/modules/trino/pages/usage-guide/configuration.adoc

@@ -13,7 +13,6 @@ For a role or role group, at the same level of `config`, you can specify `config
 * `access-control.properties`
 * `config.properties`
 * `node.properties`
-* `password-authenticator.properties`
 * `security.properties`
 * `exchange-manager.properties`
 * `spooling-manager.properties`

docs/modules/trino/pages/usage-guide/overrides.adoc

@@ -27,8 +27,12 @@ Confiuration overrides are applied like so:
 
 Configuration overrides can be applied to:
 
+* `access-control.properties`
 * `config.properties`
 * `node.properties`
+* `security.properties`
+* `exchange-manager.properties`
+* `spooling-manager.properties`
 
 === Configuration overrides in the TrinoCatalog
 

docs/modules/trino/partials/nav.adoc

@@ -33,5 +33,6 @@
 ** xref:trino:reference/crds.adoc[]
 *** {crd-docs}/trino.stackable.tech/trinocluster/v1alpha1/[TrinoCluster {external-link-icon}^]
 *** {crd-docs}/trino.stackable.tech/trinocatalog/v1alpha1/[TrinoCatalog {external-link-icon}^]
+** xref:trino:reference/security.adoc[]
 ** xref:trino:reference/commandline-parameters.adoc[]
 ** xref:trino:reference/environment-variables.adoc[]

examples/simple-trino-cluster-authentication-opa-authorization-s3.yaml

@@ -11,6 +11,7 @@ spec:
       - authenticationClass: simple-trino-users
     authorization:
       opa:
+        enableColumnMasking: false
         configMapName: simple-opa
         package: trino
     catalogLabelSelector:

rust/operator-binary/src/authorization/opa.rs

@@ -1,13 +1,11 @@
 use std::collections::BTreeMap;
 
 use stackable_operator::{
-    client::Client,
-    commons::opa::{OpaApiVersion, OpaConfig},
-    k8s_openapi::api::core::v1::ConfigMap,
+    client::Client, commons::opa::OpaApiVersion, k8s_openapi::api::core::v1::ConfigMap,
     kube::ResourceExt,
 };
 
-use crate::crd::v1alpha1::TrinoCluster;
+use crate::crd::v1alpha1;
 
 pub const OPA_TLS_VOLUME_NAME: &str = "opa-tls";
 
@@ -23,10 +21,10 @@ pub struct TrinoOpaConfig {
     /// `http://localhost:8081/v1/data/trino/rowFilters` - if not set,
     /// no row filtering will be applied
     pub(crate) row_filters_connection_string: Option<String>,
-    /// URI for fetching column masks, e.g.
-    /// `http://localhost:8081/v1/data/trino/columnMask` - if not set,
+    /// URI for fetching columns masks in batches, e.g.
+    /// `http://localhost:8081/v1/data/trino/batchColumnMasks` - if not set,
     /// no masking will be applied
-    pub(crate) column_masking_connection_string: Option<String>,
+    pub(crate) batched_column_masking_connection_string: Option<String>,
     /// Whether to allow permission management (GRANT, DENY, ...) and
     /// role management operations - OPA will not be queried for any
     /// such operations, they will be bulk allowed or denied depending
@@ -41,13 +39,15 @@ pub struct TrinoOpaConfig {
 impl TrinoOpaConfig {
     pub async fn from_opa_config(
         client: &Client,
-        trino: &TrinoCluster,
-        opa_config: &OpaConfig,
+        trino: &v1alpha1::TrinoCluster,
+        opa_config: &v1alpha1::TrinoAuthorizationOpaConfig,
     ) -> Result<Self, stackable_operator::commons::opa::Error> {
         let non_batched_connection_string = opa_config
+            .opa
             .full_document_url_from_config_map(client, trino, Some("allow"), OpaApiVersion::V1)
             .await?;
         let batched_connection_string = opa_config
+            .opa
             .full_document_url_from_config_map(
                 client,
                 trino,
@@ -57,6 +57,7 @@ impl TrinoOpaConfig {
             )
             .await?;
         let row_filters_connection_string = opa_config
+            .opa
             .full_document_url_from_config_map(
                 client,
                 trino,
@@ -65,19 +66,27 @@ impl TrinoOpaConfig {
                 OpaApiVersion::V1,
             )
             .await?;
-        let column_masking_connection_string = opa_config
-            .full_document_url_from_config_map(
-                client,
-                trino,
-                // Sticking to https://github.com/trinodb/trino/blob/455/plugin/trino-opa/src/test/java/io/trino/plugin/opa/TestOpaAccessControlDataFilteringSystem.java#L47
-                Some("columnMask"),
-                OpaApiVersion::V1,
+
+        let batched_column_masking_connection_string = if opa_config.enable_column_masking {
+            Some(
+                opa_config
+                    .opa
+                    .full_document_url_from_config_map(
+                        client,
+                        trino,
+                        // Sticking to https://github.com/trinodb/trino/blob/455/plugin/trino-opa/src/test/java/io/trino/plugin/opa/TestOpaAccessControlDataFilteringSystem.java#L48
+                        Some("batchColumnMasks"),
+                        OpaApiVersion::V1,
+                    )
+                    .await?,
             )
-            .await?;
+        } else {
+            None
+        };
 
         let tls_secret_class = client
             .get::<ConfigMap>(
-                &opa_config.config_map_name,
+                &opa_config.opa.config_map_name,
                 trino.namespace().as_deref().unwrap_or("default"),
             )
             .await
@@ -89,7 +98,7 @@ impl TrinoOpaConfig {
             non_batched_connection_string,
             batched_connection_string,
             row_filters_connection_string: Some(row_filters_connection_string),
-            column_masking_connection_string: Some(column_masking_connection_string),
+            batched_column_masking_connection_string,
             allow_permission_management_operations: true,
             tls_secret_class,
         })
@@ -113,10 +122,12 @@ impl TrinoOpaConfig {
                 Some(row_filters_connection_string.clone()),
             );
         }
-        if let Some(column_masking_connection_string) = &self.column_masking_connection_string {
+        if let Some(batched_column_masking_connection_string) =
+            &self.batched_column_masking_connection_string
+        {
             config.insert(
-                "opa.policy.column-masking-uri".to_string(),
-                Some(column_masking_connection_string.clone()),
+                "opa.policy.batch-column-masking-uri".to_string(),
+                Some(batched_column_masking_connection_string.clone()),
             );
         }
         if self.allow_permission_management_operations {

rust/operator-binary/src/controller.rs

@@ -2070,8 +2070,8 @@ mod tests {
                 "http://simple-opa.default.svc.cluster.local:8081/v1/data/my-product/rowFilters"
                     .to_string(),
             ),
-            column_masking_connection_string: Some(
-                "http://simple-opa.default.svc.cluster.local:8081/v1/data/my-product/columnMask"
+            batched_column_masking_connection_string: Some(
+                "http://simple-opa.default.svc.cluster.local:8081/v1/data/my-product/batchColumnMasks"
                     .to_string(),
             ),
             allow_permission_management_operations: true,
@@ -2157,6 +2157,7 @@ mod tests {
                     hello-from-role-group: "true" # only defined here at group level
                     foo.bar: "true" # overrides role value
                     opa.policy.batched-uri: "http://simple-opa.default.svc.cluster.local:8081/v1/data/my-product/batch-new" # override value from config
+                    opa.policy.batch-column-masking-uri: "http://simple-opa.default.svc.cluster.local:8081/v1/data/my-product/batchColumnMasks-new" # override value from config
                 replicas: 1
           workers:
             roleGroups:
@@ -2173,7 +2174,7 @@ mod tests {
         assert!(access_control_config.contains("foo.bar=true"));
         assert!(access_control_config.contains("opa.allow-permission-management-operations=false"));
         assert!(access_control_config.contains(r#"opa.policy.batched-uri=http\://simple-opa.default.svc.cluster.local\:8081/v1/data/my-product/batch-new"#));
-        assert!(access_control_config.contains(r#"opa.policy.column-masking-uri=http\://simple-opa.default.svc.cluster.local\:8081/v1/data/my-product/columnMask"#));
+        assert!(access_control_config.contains(r#"opa.policy.batch-column-masking-uri=http\://simple-opa.default.svc.cluster.local\:8081/v1/data/my-product/batchColumnMasks-new"#));
         assert!(access_control_config.contains(r#"opa.policy.row-filters-uri=http\://simple-opa.default.svc.cluster.local\:8081/v1/data/my-product/rowFilters"#));
         assert!(access_control_config.contains(r#"opa.policy.uri=http\://simple-opa.default.svc.cluster.local\:8081/v1/data/my-product/allow"#));
     }

rust/operator-binary/src/crd/mod.rs

@@ -317,10 +317,24 @@ pub mod versioned {
 
     #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
     #[serde(rename_all = "camelCase")]
-    pub struct TrinoAuthorization {
+    pub enum TrinoAuthorization {
+        Opa {
+            // no doc - it's in the struct.
+            #[serde(default, flatten)]
+            config: TrinoAuthorizationOpaConfig,
+        },
+    }
+
+    #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct TrinoAuthorizationOpaConfig {
         // no doc - it's in the struct.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub opa: Option<OpaConfig>,
+        #[serde(flatten)]
+        pub opa: OpaConfig,
+
+        /// Whether to set the OPA batched column masking URI for Trino queries; defaults to true
+        #[serde(default = "TrinoAuthorizationOpaConfig::enabled_column_masking_default")]
+        pub enable_column_masking: bool,
     }
 
     #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
@@ -370,6 +384,12 @@ pub mod versioned {
     }
 }
 
+impl v1alpha1::TrinoAuthorizationOpaConfig {
+    pub fn enabled_column_masking_default() -> bool {
+        true
+    }
+}
+
 impl Default for v1alpha1::TrinoCoordinatorRoleConfig {
     fn default() -> Self {
         v1alpha1::TrinoCoordinatorRoleConfig {
@@ -882,12 +902,14 @@ impl v1alpha1::TrinoCluster {
         !spec.cluster_config.authentication.is_empty()
     }
 
-    pub fn get_opa_config(&self) -> Option<&OpaConfig> {
+    pub fn get_opa_config(&self) -> Option<&v1alpha1::TrinoAuthorizationOpaConfig> {
         self.spec
             .cluster_config
             .authorization
             .as_ref()
-            .and_then(|a| a.opa.as_ref())
+            .map(|a| match a {
+                v1alpha1::TrinoAuthorization::Opa { config } => config,
+            })
     }
 
     /// Return user provided server TLS settings