[read] Complete Java scalar REST authorization casts

crates/integrations/datafusion/src/table/mod.rs

@@ -95,6 +95,24 @@ pub(crate) fn datafusion_arrow_schema(
     )))
 }
 
+/// The schema visible to SQL before authorization is fetched.
+fn query_auth_output_schema(table: &Table, schema: ArrowSchemaRef) -> ArrowSchemaRef {
+    if !table.schema().core_options().query_auth_enabled() {
+        return schema;
+    }
+    // Authorization is fetched while planning, after logical optimization.
+    // A mask can null a NOT NULL table column, so optimizers must not fold
+    // IS NULL beforehand. Keep physical read fields canonical and unchanged.
+    Arc::new(Schema::new_with_metadata(
+        schema
+            .fields()
+            .iter()
+            .map(|field| field.as_ref().clone().with_nullable(true))
+            .collect::<Vec<_>>(),
+        schema.metadata().clone(),
+    ))
+}
+
 /// Read-only table provider for a Paimon table.
 ///
 /// Supports full table scan, column projection, and predicate pushdown for
@@ -127,7 +145,7 @@ impl PaimonTableProvider {
         let identifier = table.identifier().clone();
         crate::table_loader::ensure_paimon_served(&table, &identifier)?;
         let fields = datafusion_read_fields(&table);
-        let schema = datafusion_arrow_schema(&fields, true)?;
+        let schema = query_auth_output_schema(&table, datafusion_arrow_schema(&fields, true)?);
         Ok(Self {
             table,
             schema,
@@ -166,7 +184,10 @@ impl PaimonTableProvider {
             return Ok(self);
         }
         let fields = datafusion_read_fields(&self.table);
-        self.schema = datafusion_arrow_schema(&fields, schema_force_view_types)?;
+        self.schema = query_auth_output_schema(
+            &self.table,
+            datafusion_arrow_schema(&fields, schema_force_view_types)?,
+        );
         Ok(self)
     }
 

crates/integrations/datafusion/tests/query_auth.rs

@@ -513,3 +513,48 @@ async fn test_query_auth_variant_get_runs_on_the_admitted_rows_only() {
         "7"
     );
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn not_null_masks_are_nullable_before_sql_optimization() {
+    let not_null_schema = |options: &[(&str, &str)]| {
+        let mut builder = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+            .column(
+                "secret",
+                DataType::VarChar(VarCharType::with_nullable(false, 255).unwrap()),
+            );
+        for (key, value) in options {
+            builder = builder.option(*key, *value);
+        }
+        builder.build().unwrap()
+    };
+    let (_tmp, _server, context) = restricted_people_with_schema(masks(), not_null_schema).await;
+    assert_eq!(
+        count(
+            &query(
+                &context,
+                &format!("SELECT COUNT(*) FROM {TABLE} WHERE secret IS NULL")
+            )
+            .await
+        ),
+        4
+    );
+    assert_eq!(
+        count(
+            &query(
+                &context,
+                &format!("SELECT COUNT(*) FROM {TABLE} WHERE secret IS NOT NULL")
+            )
+            .await
+        ),
+        0
+    );
+    assert_eq!(
+        count(&query(&context, &format!("SELECT COUNT(secret) FROM {TABLE}")).await),
+        0
+    );
+    let output = query(&context, &format!("SELECT secret FROM {TABLE} LIMIT 1")).await;
+    assert!(output[0].schema().field(0).is_nullable());
+    assert!(output[0].column(0).is_null(0));
+}