From 916acc463c577a0f31b8850fedcd1937ee96df60 Mon Sep 17 00:00:00 2001 From: Stefanietry Date: Thu, 3 Sep 2026 17:45:47 +0800 Subject: [PATCH] [spark] Support source table fileio for blob descriptors on spark --- .../utils/BlobDescriptorResolvingRow.java | 353 ++++++++++++++++++ .../sink/BlobDescriptorResolvingRow.java | 239 +----------- .../org/apache/paimon/spark/PaimonBatch.scala | 6 +- .../paimon/spark/PaimonPartitionReader.scala | 40 +- .../spark/PaimonPartitionReaderFactory.scala | 11 +- .../paimon/spark/PostponeMergeOnRead.scala | 13 +- .../execution/PostponeMergeOnReadExec.scala | 21 +- .../apache/paimon/spark/read/BaseScan.scala | 7 +- .../sources/PaimonMicroBatchStream.scala | 9 +- .../paimon/spark/SparkInternalRowTest.java | 30 ++ .../sources/PaimonMicroBatchStreamTest.scala | 46 ++- .../paimon/spark/sql/BlobTestBase.scala | 26 ++ 12 files changed, 544 insertions(+), 257 deletions(-) create mode 100644 paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorResolvingRow.java diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorResolvingRow.java b/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorResolvingRow.java new file mode 100644 index 000000000000..d04bf082ba12 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorResolvingRow.java @@ -0,0 +1,353 @@ +/* + * 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.utils; + +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobArrayPlaceholder; +import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.data.BlobMapPlaceholder; +import org.apache.paimon.data.BlobRef; +import org.apache.paimon.data.Decimal; +import org.apache.paimon.data.InternalArray; +import org.apache.paimon.data.InternalMap; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.InternalVector; +import org.apache.paimon.data.Timestamp; +import org.apache.paimon.data.variant.Variant; +import org.apache.paimon.types.RowKind; + +import javax.annotation.Nullable; + +/** Reattaches a descriptor reader to BLOBs in a row. */ +public class BlobDescriptorResolvingRow implements InternalRow { + + private InternalRow wrapped; + private final UriReaderFactory uriReaderFactory; + + public BlobDescriptorResolvingRow(InternalRow wrapped, UriReaderFactory uriReaderFactory) { + this.wrapped = wrapped; + this.uriReaderFactory = uriReaderFactory; + } + + public BlobDescriptorResolvingRow replace(InternalRow row) { + this.wrapped = row; + return this; + } + + @Override + public int getFieldCount() { + return wrapped.getFieldCount(); + } + + @Override + public RowKind getRowKind() { + return wrapped.getRowKind(); + } + + @Override + public void setRowKind(RowKind kind) { + wrapped.setRowKind(kind); + } + + @Override + public boolean isNullAt(int pos) { + return wrapped.isNullAt(pos); + } + + @Override + public boolean getBoolean(int pos) { + return wrapped.getBoolean(pos); + } + + @Override + public byte getByte(int pos) { + return wrapped.getByte(pos); + } + + @Override + public short getShort(int pos) { + return wrapped.getShort(pos); + } + + @Override + public int getInt(int pos) { + return wrapped.getInt(pos); + } + + @Override + public long getLong(int pos) { + return wrapped.getLong(pos); + } + + @Override + public float getFloat(int pos) { + return wrapped.getFloat(pos); + } + + @Override + public double getDouble(int pos) { + return wrapped.getDouble(pos); + } + + @Override + public BinaryString getString(int pos) { + return wrapped.getString(pos); + } + + @Override + public Decimal getDecimal(int pos, int precision, int scale) { + return wrapped.getDecimal(pos, precision, scale); + } + + @Override + public Timestamp getTimestamp(int pos, int precision) { + return wrapped.getTimestamp(pos, precision); + } + + @Override + public byte[] getBinary(int pos) { + return wrapped.getBinary(pos); + } + + @Override + public Variant getVariant(int pos) { + return wrapped.getVariant(pos); + } + + @Override + public Blob getBlob(int pos) { + return withReader(wrapped.getBlob(pos), uriReaderFactory); + } + + private static Blob withReader(@Nullable Blob blob, UriReaderFactory uriReaderFactory) { + if (!(blob instanceof BlobRef)) { + return blob; + } + + BlobDescriptor descriptor = blob.toDescriptor(); + return Blob.fromDescriptor(uriReaderFactory.create(descriptor.uri()), descriptor); + } + + @Override + public InternalArray getArray(int pos) { + return withReader(wrapped.getArray(pos), uriReaderFactory); + } + + @Override + public InternalVector getVector(int pos) { + return wrapped.getVector(pos); + } + + @Override + public InternalMap getMap(int pos) { + return withReader(wrapped.getMap(pos), uriReaderFactory); + } + + @Override + public InternalRow getRow(int pos, int numFields) { + InternalRow row = wrapped.getRow(pos, numFields); + return row == null ? null : new BlobDescriptorResolvingRow(row, uriReaderFactory); + } + + private static InternalArray withReader( + @Nullable InternalArray array, UriReaderFactory uriReaderFactory) { + if (array == null || array == BlobArrayPlaceholder.INSTANCE) { + return array; + } + return new BlobDescriptorResolvingArray(array, uriReaderFactory); + } + + private static InternalMap withReader( + @Nullable InternalMap map, UriReaderFactory uriReaderFactory) { + if (map == null || map == BlobMapPlaceholder.INSTANCE) { + return map; + } + return new BlobDescriptorResolvingMap(map, uriReaderFactory); + } + + private static final class BlobDescriptorResolvingArray implements InternalArray { + + private final InternalArray wrapped; + private final UriReaderFactory uriReaderFactory; + + private BlobDescriptorResolvingArray( + InternalArray wrapped, UriReaderFactory uriReaderFactory) { + this.wrapped = wrapped; + this.uriReaderFactory = uriReaderFactory; + } + + @Override + public int size() { + return wrapped.size(); + } + + @Override + public boolean isNullAt(int pos) { + return wrapped.isNullAt(pos); + } + + @Override + public boolean getBoolean(int pos) { + return wrapped.getBoolean(pos); + } + + @Override + public byte getByte(int pos) { + return wrapped.getByte(pos); + } + + @Override + public short getShort(int pos) { + return wrapped.getShort(pos); + } + + @Override + public int getInt(int pos) { + return wrapped.getInt(pos); + } + + @Override + public long getLong(int pos) { + return wrapped.getLong(pos); + } + + @Override + public float getFloat(int pos) { + return wrapped.getFloat(pos); + } + + @Override + public double getDouble(int pos) { + return wrapped.getDouble(pos); + } + + @Override + public BinaryString getString(int pos) { + return wrapped.getString(pos); + } + + @Override + public Decimal getDecimal(int pos, int precision, int scale) { + return wrapped.getDecimal(pos, precision, scale); + } + + @Override + public Timestamp getTimestamp(int pos, int precision) { + return wrapped.getTimestamp(pos, precision); + } + + @Override + public byte[] getBinary(int pos) { + return wrapped.getBinary(pos); + } + + @Override + public Variant getVariant(int pos) { + return wrapped.getVariant(pos); + } + + @Override + public Blob getBlob(int pos) { + return withReader(wrapped.getBlob(pos), uriReaderFactory); + } + + @Override + public InternalArray getArray(int pos) { + return withReader(wrapped.getArray(pos), uriReaderFactory); + } + + @Override + public InternalVector getVector(int pos) { + return wrapped.getVector(pos); + } + + @Override + public InternalMap getMap(int pos) { + return withReader(wrapped.getMap(pos), uriReaderFactory); + } + + @Override + public InternalRow getRow(int pos, int numFields) { + InternalRow row = wrapped.getRow(pos, numFields); + return row == null ? null : new BlobDescriptorResolvingRow(row, uriReaderFactory); + } + + @Override + public boolean[] toBooleanArray() { + return wrapped.toBooleanArray(); + } + + @Override + public byte[] toByteArray() { + return wrapped.toByteArray(); + } + + @Override + public short[] toShortArray() { + return wrapped.toShortArray(); + } + + @Override + public int[] toIntArray() { + return wrapped.toIntArray(); + } + + @Override + public long[] toLongArray() { + return wrapped.toLongArray(); + } + + @Override + public float[] toFloatArray() { + return wrapped.toFloatArray(); + } + + @Override + public double[] toDoubleArray() { + return wrapped.toDoubleArray(); + } + } + + private static final class BlobDescriptorResolvingMap implements InternalMap { + + private final InternalMap wrapped; + private final UriReaderFactory uriReaderFactory; + + private BlobDescriptorResolvingMap(InternalMap wrapped, UriReaderFactory uriReaderFactory) { + this.wrapped = wrapped; + this.uriReaderFactory = uriReaderFactory; + } + + @Override + public int size() { + return wrapped.size(); + } + + @Override + public InternalArray keyArray() { + return withReader(wrapped.keyArray(), uriReaderFactory); + } + + @Override + public InternalArray valueArray() { + return withReader(wrapped.valueArray(), uriReaderFactory); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BlobDescriptorResolvingRow.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BlobDescriptorResolvingRow.java index 15b9752a41a3..28ac48b4c972 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BlobDescriptorResolvingRow.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BlobDescriptorResolvingRow.java @@ -18,248 +18,13 @@ package org.apache.paimon.flink.sink; -import org.apache.paimon.data.BinaryString; -import org.apache.paimon.data.Blob; -import org.apache.paimon.data.BlobArrayPlaceholder; -import org.apache.paimon.data.BlobDescriptor; -import org.apache.paimon.data.BlobMapPlaceholder; -import org.apache.paimon.data.BlobRef; -import org.apache.paimon.data.Decimal; -import org.apache.paimon.data.InternalArray; -import org.apache.paimon.data.InternalMap; import org.apache.paimon.data.InternalRow; -import org.apache.paimon.data.InternalVector; -import org.apache.paimon.data.Timestamp; -import org.apache.paimon.data.variant.Variant; -import org.apache.paimon.utils.PartialRow; import org.apache.paimon.utils.UriReaderFactory; /** Reattaches a descriptor reader to BLOBs which passed through Flink row serialization. */ -final class BlobDescriptorResolvingRow extends PartialRow { - - private final InternalRow wrapped; - private final UriReaderFactory uriReaderFactory; +final class BlobDescriptorResolvingRow extends org.apache.paimon.utils.BlobDescriptorResolvingRow { BlobDescriptorResolvingRow(InternalRow wrapped, UriReaderFactory uriReaderFactory) { - super(wrapped.getFieldCount(), wrapped); - this.wrapped = wrapped; - this.uriReaderFactory = uriReaderFactory; - } - - @Override - public BlobDescriptorResolvingRow replace(InternalRow row) { - throw new UnsupportedOperationException("Replacing the wrapped row is not supported."); - } - - @Override - public int getFieldCount() { - return wrapped.getFieldCount(); - } - - @Override - public Blob getBlob(int pos) { - return withReader(super.getBlob(pos), uriReaderFactory); - } - - private static Blob withReader(Blob blob, UriReaderFactory uriReaderFactory) { - if (!(blob instanceof BlobRef)) { - return blob; - } - - BlobDescriptor descriptor = blob.toDescriptor(); - return Blob.fromDescriptor(uriReaderFactory.create(descriptor.uri()), descriptor); - } - - @Override - public InternalArray getArray(int pos) { - return withReader(super.getArray(pos), uriReaderFactory); - } - - @Override - public InternalMap getMap(int pos) { - return withReader(super.getMap(pos), uriReaderFactory); - } - - private static InternalArray withReader( - InternalArray array, UriReaderFactory uriReaderFactory) { - if (array == BlobArrayPlaceholder.INSTANCE) { - return array; - } - return new BlobDescriptorResolvingArray(array, uriReaderFactory); - } - - private static InternalMap withReader(InternalMap map, UriReaderFactory uriReaderFactory) { - if (map == BlobMapPlaceholder.INSTANCE) { - return map; - } - return new BlobDescriptorResolvingMap(map, uriReaderFactory); - } - - private static final class BlobDescriptorResolvingArray implements InternalArray { - - private final InternalArray wrapped; - private final UriReaderFactory uriReaderFactory; - - private BlobDescriptorResolvingArray( - InternalArray wrapped, UriReaderFactory uriReaderFactory) { - this.wrapped = wrapped; - this.uriReaderFactory = uriReaderFactory; - } - - @Override - public int size() { - return wrapped.size(); - } - - @Override - public boolean isNullAt(int pos) { - return wrapped.isNullAt(pos); - } - - @Override - public boolean getBoolean(int pos) { - return wrapped.getBoolean(pos); - } - - @Override - public byte getByte(int pos) { - return wrapped.getByte(pos); - } - - @Override - public short getShort(int pos) { - return wrapped.getShort(pos); - } - - @Override - public int getInt(int pos) { - return wrapped.getInt(pos); - } - - @Override - public long getLong(int pos) { - return wrapped.getLong(pos); - } - - @Override - public float getFloat(int pos) { - return wrapped.getFloat(pos); - } - - @Override - public double getDouble(int pos) { - return wrapped.getDouble(pos); - } - - @Override - public BinaryString getString(int pos) { - return wrapped.getString(pos); - } - - @Override - public Decimal getDecimal(int pos, int precision, int scale) { - return wrapped.getDecimal(pos, precision, scale); - } - - @Override - public Timestamp getTimestamp(int pos, int precision) { - return wrapped.getTimestamp(pos, precision); - } - - @Override - public byte[] getBinary(int pos) { - return wrapped.getBinary(pos); - } - - @Override - public Variant getVariant(int pos) { - return wrapped.getVariant(pos); - } - - @Override - public Blob getBlob(int pos) { - return withReader(wrapped.getBlob(pos), uriReaderFactory); - } - - @Override - public InternalArray getArray(int pos) { - return withReader(wrapped.getArray(pos), uriReaderFactory); - } - - @Override - public InternalVector getVector(int pos) { - return wrapped.getVector(pos); - } - - @Override - public InternalMap getMap(int pos) { - return withReader(wrapped.getMap(pos), uriReaderFactory); - } - - @Override - public InternalRow getRow(int pos, int numFields) { - return wrapped.getRow(pos, numFields); - } - - @Override - public boolean[] toBooleanArray() { - return wrapped.toBooleanArray(); - } - - @Override - public byte[] toByteArray() { - return wrapped.toByteArray(); - } - - @Override - public short[] toShortArray() { - return wrapped.toShortArray(); - } - - @Override - public int[] toIntArray() { - return wrapped.toIntArray(); - } - - @Override - public long[] toLongArray() { - return wrapped.toLongArray(); - } - - @Override - public float[] toFloatArray() { - return wrapped.toFloatArray(); - } - - @Override - public double[] toDoubleArray() { - return wrapped.toDoubleArray(); - } - } - - private static final class BlobDescriptorResolvingMap implements InternalMap { - - private final InternalMap wrapped; - private final UriReaderFactory uriReaderFactory; - - private BlobDescriptorResolvingMap(InternalMap wrapped, UriReaderFactory uriReaderFactory) { - this.wrapped = wrapped; - this.uriReaderFactory = uriReaderFactory; - } - - @Override - public int size() { - return wrapped.size(); - } - - @Override - public InternalArray keyArray() { - return wrapped.keyArray(); - } - - @Override - public InternalArray valueArray() { - return withReader(wrapped.valueArray(), uriReaderFactory); - } + super(wrapped, uriReaderFactory); } } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala index bd479d4fbf87..cff07c7cbbca 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala @@ -20,6 +20,7 @@ package org.apache.paimon.spark import org.apache.paimon.spark.schema.PaimonMetadataColumn import org.apache.paimon.table.source.ReadBuilder +import org.apache.paimon.utils.UriReaderFactory import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionReaderFactory} @@ -28,12 +29,13 @@ case class PaimonBatch( inputPartitions: Seq[PaimonInputPartition], readBuilder: ReadBuilder, blobAsDescriptor: Boolean, - metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty) + metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty)( + uriReaderFactory: UriReaderFactory = null) extends Batch { override def planInputPartitions(): Array[InputPartition] = inputPartitions.map(_.asInstanceOf[InputPartition]).toArray override def createReaderFactory(): PartitionReaderFactory = - PaimonPartitionReaderFactory(readBuilder, metadataColumns, blobAsDescriptor) + PaimonPartitionReaderFactory(readBuilder, metadataColumns, blobAsDescriptor, uriReaderFactory) } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala index fd8a6178c476..71a0dd459304 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReader.scala @@ -24,8 +24,10 @@ import org.apache.paimon.spark.SparkUtils.createIOManager import org.apache.paimon.spark.data.SparkInternalRow import org.apache.paimon.spark.schema.PaimonMetadataColumn import org.apache.paimon.spark.util.SplitUtils +import org.apache.paimon.table.{BlobDescriptorReaderFactory, FileStoreTable, Table} import org.apache.paimon.table.source.{ReadBuilder, Split} -import org.apache.paimon.types.RowType +import org.apache.paimon.types.{BlobType, RowType} +import org.apache.paimon.utils.{BlobDescriptorResolvingRow, UriReaderFactory} import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.connector.metric.CustomTaskMetric @@ -41,7 +43,8 @@ case class PaimonPartitionReader( readBuilder: ReadBuilder, partition: PaimonInputPartition, metadataColumns: Seq[PaimonMetadataColumn], - blobAsDescriptor: Boolean + blobAsDescriptor: Boolean, + uriReaderFactory: UriReaderFactory ) extends PartitionReader[InternalRow] { private val splits: Iterator[Split] = partition.splits.toIterator @@ -58,6 +61,8 @@ case class PaimonPartitionReader( private var totalReadBatchTimeMs: Long = 0L private lazy val read = readBuilder.newRead().withIOManager(ioManager) + private lazy val blobDescriptorResolvingRow = + new BlobDescriptorResolvingRow(currentRow, uriReaderFactory) override def next(): Boolean = { if (currentRecordReader == null) { @@ -73,7 +78,15 @@ case class PaimonPartitionReader( null } else { advanced = false - sparkRow.replace(currentRow) + sparkRow.replace(resolveBlobDescriptors(currentRow)) + } + } + + private def resolveBlobDescriptors(row: PaimonInternalRow): PaimonInternalRow = { + if (blobAsDescriptor || uriReaderFactory == null) { + row + } else { + blobDescriptorResolvingRow.replace(row) } } @@ -136,3 +149,24 @@ case class PaimonPartitionReader( } } } + +private[spark] object BlobDescriptorUtils { + + def createUriReaderFactory( + table: Table, + readType: RowType, + blobAsDescriptor: Boolean): UriReaderFactory = { + if (blobAsDescriptor || !hasBlobFileFields(readType)) { + null + } else { + table match { + case fileStoreTable: FileStoreTable => BlobDescriptorReaderFactory.create(fileStoreTable) + case _ => null + } + } + } + + private def hasBlobFileFields(rowType: RowType): Boolean = { + rowType.getFieldTypes.asScala.exists(BlobType.isBlobFileField) + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala index 64630574a916..d279379c4ae2 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionReaderFactory.scala @@ -20,6 +20,7 @@ package org.apache.paimon.spark import org.apache.paimon.spark.schema.PaimonMetadataColumn import org.apache.paimon.table.source.ReadBuilder +import org.apache.paimon.utils.UriReaderFactory import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader, PartitionReaderFactory} @@ -29,13 +30,19 @@ import java.util.Objects case class PaimonPartitionReaderFactory( readBuilder: ReadBuilder, metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty, - blobAsDescriptor: Boolean) + blobAsDescriptor: Boolean, + uriReaderFactory: UriReaderFactory = null) extends PartitionReaderFactory { override def createReader(partition: InputPartition): PartitionReader[InternalRow] = { partition match { case paimonInputPartition: PaimonInputPartition => - PaimonPartitionReader(readBuilder, paimonInputPartition, metadataColumns, blobAsDescriptor) + PaimonPartitionReader( + readBuilder, + paimonInputPartition, + metadataColumns, + blobAsDescriptor, + uriReaderFactory) case _ => throw new RuntimeException(s"It's not a Paimon input partition, $partition") } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala index e340be28826a..2f9aa684d6d2 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala @@ -24,6 +24,7 @@ import org.apache.paimon.predicate.PredicateBuilder import org.apache.paimon.spark.PostponeMergeOnRead.{MergePlan, RealScanInfo} import org.apache.paimon.table.{BucketMode, FileStoreTable, Table} import org.apache.paimon.table.source.{PostponeMergePlan, PostponeMergeReadBuilder} +import org.apache.paimon.utils.UriReaderFactory import scala.collection.JavaConverters._ @@ -81,8 +82,13 @@ final private[spark] class PostponeMergeOnRead(scan: PaimonBaseScan) { .reportDriverMetrics() .map(metric => metric.name() -> metric.value()) .toMap) - mergePlan = - MergePlan(builder, corePlan, scan.coreOptions.blobAsDescriptor(), realScanInfo) + val blobAsDescriptor = scan.coreOptions.blobAsDescriptor() + val uriReaderFactory = + BlobDescriptorUtils.createUriReaderFactory( + scan.table, + corePlan.resultReadType(), + blobAsDescriptor) + mergePlan = MergePlan(builder, corePlan, blobAsDescriptor, realScanInfo, uriReaderFactory) } mergePlan } @@ -131,7 +137,8 @@ private[spark] object PostponeMergeOnRead { readBuilder: PostponeMergeReadBuilder, corePlan: PostponeMergePlan, blobAsDescriptor: Boolean, - realScanInfo: RealScanInfo) + realScanInfo: RealScanInfo, + uriReaderFactory: UriReaderFactory) private[spark] case class RealScanInfo( tableName: String, diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PostponeMergeOnReadExec.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PostponeMergeOnReadExec.scala index ac160233c540..ed806abeb02b 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PostponeMergeOnReadExec.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PostponeMergeOnReadExec.scala @@ -30,7 +30,7 @@ import org.apache.paimon.spark.read.BinPackingSplits import org.apache.paimon.spark.util.SplitUtils import org.apache.paimon.table.source.{DataSplit, PostponeMergePlan, PostponeMergeReadBuilder, SplitSerializer} import org.apache.paimon.types.{RowKind, RowType} -import org.apache.paimon.utils.{IteratorRecordReader, SerializationUtils} +import org.apache.paimon.utils.{BlobDescriptorResolvingRow, IteratorRecordReader, SerializationUtils, UriReaderFactory} import org.apache.spark.TaskContext import org.apache.spark.rdd.RDD @@ -119,6 +119,7 @@ private[spark] case class PostponeMergeOnReadExec( val readBuilder = mergePlan.readBuilder val resultRowType = mergePlan.corePlan.resultReadType() val blobAsDescriptor = mergePlan.blobAsDescriptor + val uriReaderFactory = mergePlan.uriReaderFactory val outputAttributes = output val numOutputRows = longMetric(NUM_OUTPUT_ROWS) val numSplits = longMetric(NUM_SPLITS) @@ -135,7 +136,8 @@ private[spark] case class PostponeMergeOnReadExec( blobAsDescriptor, numSplits, partitionSize, - readBatchTime) + readBatchTime, + uriReaderFactory) .map { row => numOutputRows += 1L @@ -199,7 +201,8 @@ private[spark] object PostponeMergeOnReadExec { blobAsDescriptor: Boolean, numSplits: SQLMetric, partitionSize: SQLMetric, - readBatchTime: SQLMetric) + readBatchTime: SQLMetric, + uriReaderFactory: UriReaderFactory) extends Iterator[InternalRow] with AutoCloseable { @@ -207,6 +210,8 @@ private[spark] object PostponeMergeOnReadExec { private val ioManager = SparkUtils.createIOManager() private val read = readBuilder.newRead().withIOManager(ioManager) private val sparkRow = SparkInternalRow.create(resultRowType, blobAsDescriptor) + private lazy val blobDescriptorResolvingRow = + new BlobDescriptorResolvingRow(null, uriReaderFactory) private var currentReader: RecordReaderIterator[PaimonInternalRow] = _ private var currentTimedReader: TimedRecordReader[PaimonInternalRow] = _ private var nextRow: InternalRow = _ @@ -235,13 +240,21 @@ private[spark] object PostponeMergeOnReadExec { return } if (currentReader.hasNext) { - nextRow = sparkRow.replace(currentReader.next()) + nextRow = sparkRow.replace(resolveBlobDescriptors(currentReader.next())) } else { closeCurrentReader() } } } + private def resolveBlobDescriptors(row: PaimonInternalRow): PaimonInternalRow = { + if (blobAsDescriptor || uriReaderFactory == null) { + row + } else { + blobDescriptorResolvingRow.replace(row) + } + } + private def openNextReader(): Boolean = { if (!bufferedRows.hasNext) { false diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala index 088b785c7b2e..0e9e4b0a7e41 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala @@ -21,7 +21,7 @@ package org.apache.paimon.spark.read import org.apache.paimon.CoreOptions import org.apache.paimon.partition.PartitionPredicate import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, Predicate, TopN, VectorSearch} -import org.apache.paimon.spark.{PaimonBatch, PaimonInputPartition, PaimonNumSplitMetric, PaimonPartitionSizeMetric, PaimonReadBatchTimeMetric, PaimonResultedTableFilesMetric, PaimonResultedTableFilesTaskMetric, SparkTypeUtils} +import org.apache.paimon.spark.{BlobDescriptorUtils, PaimonBatch, PaimonInputPartition, PaimonNumSplitMetric, PaimonPartitionSizeMetric, PaimonReadBatchTimeMetric, PaimonResultedTableFilesMetric, PaimonResultedTableFilesTaskMetric, SparkTypeUtils} import org.apache.paimon.spark.schema.PaimonMetadataColumn import org.apache.paimon.spark.schema.PaimonMetadataColumn._ import org.apache.paimon.spark.util.{OptionUtils, SplitUtils} @@ -144,7 +144,10 @@ trait BaseScan extends Scan with SupportsReportStatistics with Logging { override def toBatch: Batch = { val metadataColumns = metadataFields.map( field => PaimonMetadataColumn.get(field.name, SparkTypeUtils.toSparkPartitionType(table))) - PaimonBatch(inputPartitions, readBuilder, coreOptions.blobAsDescriptor(), metadataColumns) + val blobAsDescriptor = coreOptions.blobAsDescriptor() + val uriReaderFactory = + BlobDescriptorUtils.createUriReaderFactory(table, readTableRowType, blobAsDescriptor) + PaimonBatch(inputPartitions, readBuilder, blobAsDescriptor, metadataColumns)(uriReaderFactory) } def estimateStatistics: Statistics = { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala index 1730cb6d08e5..9f699abc191f 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStream.scala @@ -21,7 +21,7 @@ package org.apache.paimon.spark.sources import org.apache.paimon.CoreOptions import org.apache.paimon.options.Options import org.apache.paimon.schema.TableSchema -import org.apache.paimon.spark.{PaimonImplicits, PaimonMicroBatchInputPartition, PaimonMicroBatchMetadata, PaimonPartitionReaderFactory, SparkConnectorOptions} +import org.apache.paimon.spark.{BlobDescriptorUtils, PaimonImplicits, PaimonMicroBatchInputPartition, PaimonMicroBatchMetadata, PaimonPartitionReaderFactory, SparkConnectorOptions} import org.apache.paimon.table.DataTable import org.apache.paimon.table.source.{DataSplit, OutOfRangeException, ReadBuilder} import org.apache.paimon.utils.DataEvolutionUtils @@ -104,6 +104,8 @@ class PaimonMicroBatchStream( } private lazy val blobAsDescriptor: Boolean = options.get(CoreOptions.BLOB_AS_DESCRIPTOR) + private lazy val uriReaderFactory = + BlobDescriptorUtils.createUriReaderFactory(table, readBuilder.readType(), blobAsDescriptor) private[spark] lazy val schemaLoader: Function[JLong, TableSchema] = { val schemaManager = table.schemaManager() @@ -210,7 +212,10 @@ class PaimonMicroBatchStream( } override def createReaderFactory(): PartitionReaderFactory = { - PaimonPartitionReaderFactory(readBuilder, blobAsDescriptor = blobAsDescriptor) + PaimonPartitionReaderFactory( + readBuilder, + blobAsDescriptor = blobAsDescriptor, + uriReaderFactory = uriReaderFactory) } override def initialOffset(): Offset = { diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java index e56a07a5274b..98791e7b9e1c 100644 --- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java +++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java @@ -27,11 +27,14 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.Timestamp; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.spark.data.SparkInternalRow; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.BlobDescriptorResolvingRow; import org.apache.paimon.utils.DateTimeUtils; +import org.apache.paimon.utils.UriReader; import org.apache.paimon.utils.UriReaderFactory; import org.apache.spark.sql.catalyst.CatalystTypeConverters; @@ -41,6 +44,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.IOException; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.nio.file.Files; @@ -164,6 +168,32 @@ public void testReadBlobWithProvidedUriReaderFactory() throws Exception { assertThat(wrapper.getBlob(0).toData()).isEqualTo(bytes); } + @Test + public void testBlobDescriptorResolvingRowReattachesReader() throws Exception { + byte[] bytes = new byte[] {1, 2, 3}; + java.nio.file.Path blobFile = tempPath.resolve("resolved-blob"); + Files.write(blobFile, bytes); + BlobDescriptor descriptor = + new BlobDescriptor(blobFile.toUri().toString(), 0, bytes.length); + UriReader failingReader = + new UriReader() { + @Override + public SeekableInputStream newInputStream(String uri) throws IOException { + throw new IOException("Should use reattached reader."); + } + }; + + InternalRow row = GenericRow.of(Blob.fromDescriptor(failingReader, descriptor)); + BlobDescriptorResolvingRow resolvingRow = + new BlobDescriptorResolvingRow( + row, UriReaderFactory.fromFileIO(LocalFileIO.create())); + + assertThatThrownBy(() -> row.getBlob(0).toData()) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Should use reattached reader."); + assertThat(resolvingRow.getBlob(0).toData()).isEqualTo(bytes); + } + @Test public void testMapBlobRejectsNullKey() { Map map = new LinkedHashMap<>(); diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala index d4c6965a2268..8a3a780d4909 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sources/PaimonMicroBatchStreamTest.scala @@ -19,15 +19,19 @@ package org.apache.paimon.spark.sources import org.apache.paimon.CoreOptions -import org.apache.paimon.table.DataTable +import org.apache.paimon.catalog.CatalogContext +import org.apache.paimon.options.Options +import org.apache.paimon.spark.PaimonPartitionReaderFactory +import org.apache.paimon.table.{CatalogEnvironment, DataTable, FileStoreTable} import org.apache.paimon.table.source.{ReadBuilder, StreamDataTableScan} +import org.apache.paimon.types.{BlobType, DataField, IntType, RowType} import org.mockito.ArgumentMatchers.anyLong import org.mockito.Mockito.{doNothing, doThrow, mock, never, times, verify, when} import org.scalatest.funsuite.AnyFunSuite import java.io.{IOException, UncheckedIOException} -import java.util.Collections +import java.util.{Arrays, Collections, HashMap} class PaimonMicroBatchStreamTest extends AnyFunSuite { @@ -87,6 +91,39 @@ class PaimonMicroBatchStreamTest extends AnyFunSuite { verify(scan, times(2)).notifyCheckpointComplete(6L) } + test("skip blob descriptor source table loading when descriptor output is requested") { + val table = mock(classOf[FileStoreTable]) + val readBuilder = mock(classOf[ReadBuilder]) + val options = new HashMap[String, String] + options.put(CoreOptions.BLOB_AS_DESCRIPTOR.key(), "true") + options.put(CoreOptions.BLOB_DESCRIPTOR_SOURCE_TABLE.key(), "db.missing_source") + when(table.options()).thenReturn(options) + when(readBuilder.readType()).thenReturn(blobRowType) + + val stream = new PaimonMicroBatchStream(table, readBuilder, "unused") + val readerFactory = stream.createReaderFactory().asInstanceOf[PaimonPartitionReaderFactory] + + assert(readerFactory.uriReaderFactory == null) + verify(table, never()).catalogEnvironment() + } + + test("propagate blob descriptor reader to streaming reader factory") { + val table = mock(classOf[FileStoreTable]) + val readBuilder = mock(classOf[ReadBuilder]) + val catalogEnvironment = mock(classOf[CatalogEnvironment]) + val options = Collections.emptyMap[String, String]() + when(table.options()).thenReturn(options) + when(table.coreOptions()).thenReturn(CoreOptions.fromMap(options)) + when(table.catalogEnvironment()).thenReturn(catalogEnvironment) + when(catalogEnvironment.catalogContext()).thenReturn(CatalogContext.create(new Options)) + when(readBuilder.readType()).thenReturn(blobRowType) + + val stream = new PaimonMicroBatchStream(table, readBuilder, "unused") + val readerFactory = stream.createReaderFactory().asInstanceOf[PaimonPartitionReaderFactory] + + assert(readerFactory.uriReaderFactory != null) + } + private def consumerOffset(index: Long, totalSplits: Long): PaimonSourceOffset = { PaimonSourceOffset.withTotalSplits( snapshotId = 5L, @@ -105,4 +142,9 @@ class PaimonMicroBatchStreamTest extends AnyFunSuite { when(scan.dropStats()).thenReturn(scan) (new PaimonMicroBatchStream(table, readBuilder, "unused"), scan) } + + private def blobRowType: RowType = { + new RowType( + Arrays.asList(new DataField(0, "id", new IntType), new DataField(1, "picture", new BlobType))) + } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala index b09a243587bd..1f86fa53c02a 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala @@ -373,6 +373,32 @@ class BlobTestBase extends PaimonSparkTestBase { } } + test("Blob: read blob descriptor field data") { + withTable("t") { + val blobData = new Array[Byte](1024 * 1024) + RANDOM.nextBytes(blobData) + val fileIO = new LocalFileIO + val uri = "file://" + tempDBDir.toString + "/external_descriptor_field_blob" + writeFile(fileIO, uri, blobData) + val blobDescriptor = new BlobDescriptor(uri, 0, blobData.length) + + sql( + "CREATE TABLE t (id INT, picture BINARY) TBLPROPERTIES (" + + "'row-tracking.enabled'='true', " + + "'data-evolution.enabled'='true', " + + "'blob-descriptor-field'='picture')") + sql(s"INSERT INTO t VALUES (1, X'${bytesToHex(blobDescriptor.serialize())}')") + + checkAnswer(sql("SELECT picture FROM t WHERE id = 1"), Seq(Row(blobData))) + + sql("ALTER TABLE t SET TBLPROPERTIES ('blob-as-descriptor'='true')") + val descriptorBytes = + sql("SELECT picture FROM t WHERE id = 1").collect()(0).get(0).asInstanceOf[Array[Byte]] + val actualDescriptor = BlobDescriptor.deserialize(descriptorBytes) + assert(actualDescriptor.equals(blobDescriptor)) + } + } + test("Blob: materialize descriptor with source table FileIO") { withTable("blob_source", "blob_target") { sql(