Skip to content

[spark] Support SQL filters in create_policy - #10950

Open
zhoulii wants to merge 3 commits into
apache:masterfrom
zhoulii:feature/spark-policy-filter-sql
Open

zhoulii wants to merge 3 commits into
apache:masterfrom
zhoulii:feature/spark-policy-filter-sql

Conversation

@zhoulii

@zhoulii zhoulii commented Nov 2, 2026

Copy link
Copy Markdown
Member

Purpose

close #10949

Tests

@JingsongLi JingsongLi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Requirement fit: SUPPORTED. Implementation: FINDINGS.

SQL policy authoring has clear end-to-end value. The 15 PermissionProcedureTest cases pass locally on JDK 8 / Spark 3.5.8 with the normal Maven quality checks, and this head's upstream CI is green. An additional real-table REST policy test demonstrates the authorization mismatch below. Temporary reviewer sources were removed.


val normalized = SimplifyCasts(ConstantFolding(filter)).asInstanceOf[Filter].condition
// A stored policy has no engine-side residual filter, so every conjunct must convert.
convertConditionToPaimonPredicate(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Require exact SQL semantics before storing a converted policy

ignorePartialFailure = false proves that every expression can be converted, but does not prove semantic equivalence. I reproduced an unauthorized row through the actual REST policy/read path on this head: create (id INT, v DOUBLE) with query-auth.enabled=true, insert (1, CAST('-0.0' AS DOUBLE)), (2, 0.0), (3, 1.0), and create a ROW_FILTER with filter_sql => 'v <> CAST(0 AS DOUBLE)'. Spark's evaluation over an independent VALUES relation returns only id 3; selecting from the policy-protected table returns ids 1 and 3. The generated NotEqual calls CompareUtils/Double.compareTo, which distinguishes the two signs of zero, whereas Spark SQL considers them equal. The persisted policy has no Spark residual to remove the extra row. FLOAT has the same comparison boundary. Please reject conversions without exact policy semantics or supply an exact conversion (including signed-zero comparisons), and add a real authorization regression test whose expected result is evaluated independently of Paimon pushdown.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed the signed-zero comparisons for FLOAT/DOUBLE. Timestamp precision is also normalized; legacy timestamp comparisons and nested timestamp comparisons with precision above 6 are rejected before storing a policy.

Added end-to-end REST authorization tests with expected results evaluated independently using Spark VALUES queries.

Normalize signed-zero comparisons and timestamp precision for SQL-authored policies while retaining native Predicate JSON semantics.

Reject timestamp value comparisons under legacy timestamp mapping, where JVM time-zone interpretation can change the matching rows. Verify real REST policy reads against independent Spark evaluation.

@JingsongLi JingsongLi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Requirement fit: SUPPORTED. Implementation: FINDINGS.

The signed-zero and timestamp-comparison issues from the earlier review are addressed. The normal JDK 8 / Spark 3.5.8 reactor build and all 20 PermissionProcedureTest cases pass, including the independent VALUES-based authorization checks. A separate actual REST policy/read regression still grants an unauthorized row when Spark's case conversion depends on the JVM locale, as detailed below. Adding that one case gives 20 passes and one failure; reviewer source changes were removed afterward. I am not claiming the same locale behavior for Spark 4.

relation.output,
rowType,
ignorePartialFailure = false)
.map(normalizePolicyPredicate)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Reject locale-sensitive case conversion without exact SQL semantics

Case-conversion transforms still pass through with native Paimon semantics. On JDK 8 / Spark 3.5.8, setting the JVM default locale to tr_TR, inserting (1, 'Ié') into (id INT, region STRING), and creating a ROW_FILTER with filter_sql => "lower(region) <> 'ıé'" reproduces an unauthorized row through the actual REST policy/read path. An independent Spark VALUES query returns no rows, while SELECT id from the protected table (without WHERE) returns id 1. For this non-ASCII input, Spark's UTF8String lower-case fallback uses the default locale and yields ıé; Paimon's BinaryString lower-case fallback uses Locale.ROOT and yields ié. There is no Spark residual to remove the extra row. Please reject this conversion when exact case/locale semantics cannot be represented, or provide an equivalent persisted transform, and retain this end-to-end locale regression.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Non-constant LOWER/UPPER expressions are now rejected after constant folding, before storing a policy. Constant calls remain supported.

Added a Turkish-locale REST regression verifying rejection without persisting a policy. Native Predicate JSON behavior is unchanged.

Reject non-constant LOWER/UPPER after constant folding, before a policy can be stored. Keep constant calls and native Predicate JSON supported.

Add end-to-end ROOT and Turkish locale regressions and document the support boundary.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Feature] Support SQL filters in Spark create_policy

2 participants