This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4220791657 [spark] Read split statistics through the query
authorization wrapper (#10051)
4220791657 is described below
commit 422079165759af3494a9ec49a68a27915b75eb35
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Sep 22 10:49:26 2026 +0800
[spark] Read split statistics through the query authorization wrapper
(#10051)
---
.../org/apache/paimon/spark/util/SplitUtils.scala | 11 +--
.../spark/util/SplitUtilsQueryAuthTest.scala | 83 ++++++++++++++++++++++
2 files changed, 90 insertions(+), 4 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
index c485fe4da3..c970db1297 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
@@ -19,7 +19,7 @@
package org.apache.paimon.spark.util
import org.apache.paimon.table.format.FormatDataSplit
-import org.apache.paimon.table.source.{DataSplit, Split}
+import org.apache.paimon.table.source.{DataSplit, Split, Splits}
import java.util.{Collections => JCollections}
@@ -27,8 +27,11 @@ import scala.collection.JavaConverters._
object SplitUtils {
+ /** Only metadata is read here, so peeling the authorization wrapper exposes
no rows. */
+ private def underlying(split: Split): Split = Splits.underlying(split)
+
def splitSize(split: Split): Long = {
- split match {
+ underlying(split) match {
case ds: DataSplit =>
ds.dataFiles().asScala.map(_.fileSize).sum
case fs: FormatDataSplit =>
@@ -40,7 +43,7 @@ object SplitUtils {
def fileCount(split: Split): Long = dataFileCount(split) +
deleteFileCount(split)
def dataFileCount(split: Split): Long = {
- split match {
+ underlying(split) match {
case ds: DataSplit => ds.dataFiles().size()
case fs: FormatDataSplit => fs.fileCount()
case _ => 0
@@ -48,7 +51,7 @@ object SplitUtils {
}
def deleteFileCount(split: Split): Long = {
- split match {
+ underlying(split) match {
case ds: DataSplit =>
ds.deletionFiles()
.orElse(JCollections.emptyList())
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/SplitUtilsQueryAuthTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/SplitUtilsQueryAuthTest.scala
new file mode 100644
index 0000000000..cada2b24a2
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/SplitUtilsQueryAuthTest.scala
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.util
+
+import org.apache.paimon.catalog.TableQueryAuthResult
+import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.io.DataFileMeta
+import org.apache.paimon.spark.PaimonRecordReaderIterator
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
+import org.apache.paimon.stats.SimpleStats
+import org.apache.paimon.table.source.{DataSplit, QueryAuthSplit, Split}
+
+import org.apache.spark.sql.types.BinaryType
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.Collections
+
+class SplitUtilsQueryAuthTest extends AnyFunSuite {
+
+ private def dataSplit(): DataSplit = {
+ val file = DataFileMeta.forAppend(
+ "data-0.parquet",
+ 1024L,
+ 10L,
+ SimpleStats.EMPTY_STATS,
+ 0L,
+ 9L,
+ 0L,
+ Collections.emptyList[String](),
+ null,
+ null,
+ null,
+ null,
+ null,
+ null)
+ DataSplit
+ .builder()
+ .withSnapshot(1L)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withDataFiles(Collections.singletonList(file))
+ .build()
+ }
+
+ private def authorized(split: Split): Split =
+ new QueryAuthSplit(split, new TableQueryAuthResult(null, null))
+
+ test("statistics survive the query-authorization wrapper") {
+ val split = dataSplit()
+ assert(SplitUtils.splitSize(split) == 1024L)
+ assert(SplitUtils.dataFileCount(split) == 1L)
+
+ val wrapped = authorized(split)
+ assert(SplitUtils.splitSize(wrapped) == 1024L, "split size collapsed under
query auth")
+ assert(SplitUtils.dataFileCount(wrapped) == 1L, "file count collapsed
under query auth")
+ }
+
+ test("partition metadata columns are refused on a query-authorized split") {
+ val partitionColumn =
+ PaimonMetadataColumn(-1, PaimonMetadataColumn.PARTITION_COLUMN,
BinaryType)
+ val error = intercept[RuntimeException] {
+ PaimonRecordReaderIterator(null, Seq(partitionColumn),
authorized(dataSplit()))
+ }
+ assert(error.getMessage.contains("DataSplit"))
+ }
+}
