leaves12138 commented on code in PR #8711:
URL: https://github.com/apache/paimon/pull/8711#discussion_r3613372247
##########
paimon-core/src/main/java/org/apache/paimon/table/source/QueryAuthSplit.java:
##########
@@ -49,6 +49,24 @@ public Split split() {
return split;
}
+ /** Unwraps one query authorization layer to expose the wrapped split. */
+ public static Split unwrap(Split split) {
+ return split instanceof QueryAuthSplit ? ((QueryAuthSplit)
split).split() : split;
Review Comment:
[P1] Unwrap nested query-auth layers before casting
When the lateral primary-key vector query has an authorized match,
`scan.plan()` returns `QueryAuthSplit(QueryAuthSplit(IndexedSplit))`. This
method removes only the outer wrapper, so `PaimonStrategy.scala:570` casts the
remaining `QueryAuthSplit` to `IndexedSplit` and throws `ClassCastException`. I
reproduced this by making row 1's query vector match row 1 in the new REST
test. Please unwrap repeatedly and add a nested-wrapper regression test.
##########
paimon-core/src/main/java/org/apache/paimon/table/source/QueryAuthSplit.java:
##########
@@ -49,6 +49,24 @@ public Split split() {
return split;
}
+ /** Unwraps one query authorization layer to expose the wrapped split. */
+ public static Split unwrap(Split split) {
+ return split instanceof QueryAuthSplit ? ((QueryAuthSplit)
split).split() : split;
+ }
+
+ /** Unwraps one query authorization layer and returns the wrapped data
split. */
+ public static DataSplit unwrapDataSplit(Split split) {
+ return (DataSplit) unwrap(split);
Review Comment:
[P1] Handle fallback wrappers before returning a data split
`FallbackReadFileStoreTable.toFallbackSplit` wraps non-`DataSplit` inputs as
`FallbackSplitImpl`, so fallback plus query auth produces
`FallbackSplitImpl(QueryAuthSplit(DataSplit))`. Callers such as
`PreAssignSplitAssigner` pass that object here; `unwrap` leaves it unchanged
and this cast throws `ClassCastException`. A focused Flink DPP regression fails
at this line with `FallbackSplitImpl cannot be cast to DataSplit`. Please use
wrapper-aware extraction (or keep the fallback wrapper `DataSplit`-compatible)
and cover this composition.
##########
paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java:
##########
@@ -733,6 +734,117 @@ public void testRowFilterBucketKeyTable() {
.isEqualTo("[[3,c]]");
}
+ @Test
+ public void testRowFilterWithPartitionAndBucketMetadata() {
+ spark.sql(
+ "CREATE TABLE t_auth_metadata (id INT, name STRING, pt STRING)
"
+ + "PARTITIONED BY (pt) TBLPROPERTIES "
+ + "('bucket'='2', 'bucket-key'='id',
'query-auth.enabled'='true')");
+ spark.sql(
+ "INSERT INTO t_auth_metadata VALUES "
+ + "(1, 'blocked', 'p1'), (2, 'allowed', 'p2')");
+
+ Predicate idEq2Predicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(0, "id",
DataTypes.INT())),
+ Equal.INSTANCE,
+ Collections.singletonList(2));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_auth_metadata"),
+ Collections.singletonList(idEq2Predicate));
+
+ List<Row> rows =
+ spark.sql(
+ "SELECT id, __paimon_partition,
__paimon_bucket "
+ + "FROM t_auth_metadata ORDER BY id")
+ .collectAsList();
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getInt(0)).isEqualTo(2);
+ assertThat(rows.get(0).getStruct(1).getString(0)).isEqualTo("p2");
+ assertThat(rows.get(0).getInt(2)).isBetween(0, 1);
+ }
+
+ @Test
+ public void testStreamingRowFilter() throws Exception {
+ spark.sql(
+ "CREATE TABLE t_stream_auth (id INT, name STRING)
TBLPROPERTIES "
+ + "('query-auth.enabled'='true')");
+ spark.sql("INSERT INTO t_stream_auth VALUES (1, 'blocked'), (2,
'allowed')");
+
+ Predicate idEq2Predicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(0, "id",
DataTypes.INT())),
+ Equal.INSTANCE,
+ Collections.singletonList(2));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_stream_auth"),
+ Collections.singletonList(idEq2Predicate));
+
+ StreamingQuery query =
+ spark.readStream()
+ .format("paimon")
+ .table("t_stream_auth")
+ .writeStream()
+ .format("memory")
+ .option(
+ "checkpointLocation",
+
tempFile.resolve("stream-auth-checkpoint").toString())
+ .queryName("stream_auth_result")
+ .outputMode("append")
+ .start();
+ try {
+ query.processAllAvailable();
+ assertThat(
+ spark.sql("SELECT * FROM stream_auth_result ORDER
BY id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[2,allowed]]");
+ } finally {
+ query.stop();
+ }
+ }
+
+ @Test
+ public void testLateralPrimaryKeyVectorSearchWithRowFilter() {
+ spark.sql(
+ "CREATE TABLE t_auth_vector ("
+ + "id INT, embedding ARRAY<FLOAT>, query_embedding
ARRAY<FLOAT>) "
+ + "TBLPROPERTIES ("
+ + "'primary-key'='id', "
+ + "'bucket'='1', "
+ + "'deletion-vectors.enabled'='true', "
+ + "'vector-field'='embedding', "
+ + "'field.embedding.vector-dim'='2', "
+ + "'pk-vector.index.columns'='embedding', "
+ +
"'fields.embedding.pk-vector.index.type'='test-vector-ann', "
+ + "'fields.embedding.pk-vector.distance.metric'='l2', "
+ + "'test.vector.dimension'='2', "
+ + "'test.vector.metric'='l2', "
+ + "'query-auth.enabled'='true')");
+ spark.sql(
+ "INSERT INTO t_auth_vector VALUES "
+ + "(1, array(5.0f, 0.0f), array(0.0f, 0.0f)), "
+ + "(2, array(1.0f, 0.0f), array(0.0f, 0.0f))");
+
+ Predicate idEq1Predicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(0, "id",
DataTypes.INT())),
+ Equal.INSTANCE,
+ Collections.singletonList(1));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_auth_vector"),
+ Collections.singletonList(idEq1Predicate));
+
+ List<Row> rows =
+ spark.sql(
+ "SELECT q.id AS query_id, r.id AS result_id "
+ + "FROM t_auth_vector AS q, "
+ + "LATERAL (SELECT id FROM
vector_search("
+ + "'t_auth_vector', 'embedding',
q.query_embedding, 1)) AS r")
+ .collectAsList();
+ assertThat(rows).isEmpty();
Review Comment:
[P1] Preserve score-reader metadata for authorized matches
This test only exercises the case where the nearest vector candidate is
rejected, so no scored row reaches `PaimonRecordReaderIterator`. With an
allowed match (row 1 query vector `[5, 0]`, expecting `[[1,1]]`), after fixing
the nested unwrapping issue, `TableQueryAuthResult.doAuth` calls
`RecordReader.filter` and returns a plain `RecordReader`.
`needVectorSearchMetadata` then becomes false and
`PaimonRecordReaderIterator.scala:139` casts a `ScoreRecordIterator` to
`FileRecordIterator`, causing `ClassCastException`. Please preserve the
`ScoreRecordReader` contract through auth filter/transform and add a
positive-result assertion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
