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 3d49b5044d [spark] Cache blob field lookup in SparkInternalRow (#8943)
3d49b5044d is described below
commit 3d49b5044d31c07107e6a9cc2405464e08a6e2f3
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Thu Aug 6 17:44:01 2026 +0800
[spark] Cache blob field lookup in SparkInternalRow (#8943)
---
.../main/java/org/apache/paimon/types/RowType.java | 18 +++++
.../paimon/spark/data/SparkInternalRow.scala | 27 +++----
.../paimon/spark/data/SparkInternalRowTest.scala | 83 ++++++++++++++++++++++
3 files changed, 110 insertions(+), 18 deletions(-)
diff --git a/paimon-api/src/main/java/org/apache/paimon/types/RowType.java
b/paimon-api/src/main/java/org/apache/paimon/types/RowType.java
index 30b357f5b9..cadb88f317 100644
--- a/paimon-api/src/main/java/org/apache/paimon/types/RowType.java
+++ b/paimon-api/src/main/java/org/apache/paimon/types/RowType.java
@@ -70,6 +70,8 @@ public final class RowType extends DataType {
private transient volatile Map<Integer, DataField> laziedFieldIdToField;
private transient volatile Map<Integer, Integer> laziedFieldIdToIndex;
+ private transient volatile Set<Integer> laziedBlobFieldIndices;
+
public RowType(boolean isNullable, List<DataField> fields) {
super(isNullable, DataTypeRoot.ROW);
this.fields =
@@ -121,6 +123,22 @@ public final class RowType extends DataType {
return projection;
}
+ /** Returns the indices of top-level BLOB fields. */
+ public Set<Integer> getBlobFieldIndices() {
+ Set<Integer> blobFieldIndices = this.laziedBlobFieldIndices;
+ if (blobFieldIndices == null) {
+ Set<Integer> indices = new HashSet<>();
+ for (int i = 0; i < fields.size(); i++) {
+ if (fields.get(i).type().getTypeRoot() == DataTypeRoot.BLOB) {
+ indices.add(i);
+ }
+ }
+ blobFieldIndices = Collections.unmodifiableSet(indices);
+ this.laziedBlobFieldIndices = blobFieldIndices;
+ }
+ return blobFieldIndices;
+ }
+
public boolean containsField(String fieldName) {
return nameToField().containsKey(fieldName);
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/data/SparkInternalRow.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/data/SparkInternalRow.scala
index 71d1361d0c..e0a5f5ac04 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/data/SparkInternalRow.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/data/SparkInternalRow.scala
@@ -18,12 +18,12 @@
package org.apache.paimon.spark.data
-import org.apache.paimon.types.{DataTypeRoot, RowType}
+import org.apache.paimon.types.RowType
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.paimon.shims.SparkShimLoader
-import scala.collection.mutable
+import scala.collection.JavaConverters._
abstract class SparkInternalRow extends InternalRow {
def replace(row: org.apache.paimon.data.InternalRow): SparkInternalRow
@@ -38,24 +38,15 @@ object SparkInternalRow {
}
def create(rowType: RowType, blobAsDescriptor: Boolean): SparkInternalRow = {
- val blobs = blobFields(rowType)
- if (blobs.nonEmpty) {
- SparkShimLoader.shim.createSparkInternalRowWithBlob(rowType, blobs,
blobAsDescriptor)
- } else {
+ val blobFieldIndices = rowType.getBlobFieldIndices
+ if (blobFieldIndices.isEmpty) {
SparkShimLoader.shim.createSparkInternalRow(rowType).withBlobAsDescriptor(blobAsDescriptor)
+ } else {
+ SparkShimLoader.shim.createSparkInternalRowWithBlob(
+ rowType,
+ blobFieldIndices.asScala.map(_.intValue()).toSet,
+ blobAsDescriptor)
}
}
- private def blobFields(rowType: RowType): Set[Int] = {
- var i: Int = 0
- val blobFields = new mutable.HashSet[Int]()
- while (i < rowType.getFieldCount) {
- if (rowType.getTypeAt(i).getTypeRoot.equals(DataTypeRoot.BLOB)) {
- blobFields.add(i)
- }
- i += 1
- }
- blobFields.toSet
- }
-
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/data/SparkInternalRowTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/data/SparkInternalRowTest.scala
new file mode 100644
index 0000000000..b6bd9a77d6
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/data/SparkInternalRowTest.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.data
+
+import org.apache.paimon.types.{BinaryType, DataField, DataTypeRoot, RowType}
+
+import org.apache.spark.SparkFunSuite
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.collection.JavaConverters._
+
+class SparkInternalRowTest extends SparkFunSuite {
+
+ test("cache blob field lookup for reused row type") {
+ val getTypeRootCount = new AtomicInteger()
+ val rowType = newRowType(getTypeRootCount)
+
+ SparkInternalRow.create(rowType)
+ SparkInternalRow.create(rowType)
+
+ assert(getTypeRootCount.get() == 10)
+ assert(rowType.getBlobFieldIndices.asScala == Set(3, 7))
+ }
+
+ test("cache blob field lookup per row type instance") {
+ val getTypeRootCount = new AtomicInteger()
+ val first = newRowType(getTypeRootCount)
+ val second = newRowType(getTypeRootCount)
+
+ assert(first == second)
+ SparkInternalRow.create(first)
+ SparkInternalRow.create(second)
+
+ assert(getTypeRootCount.get() == 20)
+ }
+
+ test("blob field indices are immutable") {
+ val rowType = newRowType(new AtomicInteger())
+
+ intercept[UnsupportedOperationException] {
+ rowType.getBlobFieldIndices.add(1)
+ }
+ }
+
+ private def newRowType(getTypeRootCount: AtomicInteger): RowType = {
+ new RowType((0 until 10).map {
+ i =>
+ val dataType =
+ if (i == 3 || i == 7) {
+ new CountingBinaryType(getTypeRootCount, DataTypeRoot.BLOB)
+ } else {
+ new CountingBinaryType(getTypeRootCount, DataTypeRoot.BINARY)
+ }
+ new DataField(i, s"f$i", dataType)
+ }.asJava)
+ }
+
+ private class CountingBinaryType(getTypeRootCount: AtomicInteger, typeRoot:
DataTypeRoot)
+ extends BinaryType {
+
+ override def getTypeRoot: DataTypeRoot = {
+ getTypeRootCount.incrementAndGet()
+ typeRoot
+ }
+ }
+}
