This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 3ad516f276d0b3fc3d6dc1d4218fc8f210a1178a Author: Stefanietry <[email protected]> AuthorDate: Fri Sep 11 15:24:27 2026 +0800 [spark] Support source table fileio for blob descriptors on spark (#9598) --- .../paimon/utils}/BlobDescriptorResolvingRow.java | 171 +++++++++++++-- .../paimon/table/BlobDescriptorReadUtils.java | 67 ++++++ .../flink/sink/BlobDescriptorResolvingRow.java | 239 +-------------------- .../org/apache/paimon/spark/PaimonBatch.scala | 13 +- .../paimon/spark/PaimonPartitionReader.scala | 17 +- .../spark/PaimonPartitionReaderFactory.scala | 14 +- .../apache/paimon/spark/PostponeMergeOnRead.scala | 25 ++- .../spark/execution/PostponeMergeOnReadExec.scala | 37 +++- .../org/apache/paimon/spark/read/BaseScan.scala | 14 +- .../spark/sources/PaimonMicroBatchStream.scala | 15 +- .../apache/paimon/spark/SparkInternalRowTest.java | 80 +++++++ .../spark/sources/PaimonMicroBatchStreamTest.scala | 47 +++- .../org/apache/paimon/spark/sql/BlobTestBase.scala | 35 +++ 13 files changed, 491 insertions(+), 283 deletions(-) diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BlobDescriptorResolvingRow.java b/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorResolvingRow.java similarity index 58% copy from paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BlobDescriptorResolvingRow.java copy to paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorResolvingRow.java index 15b9752a41..7326b5d922 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BlobDescriptorResolvingRow.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorResolvingRow.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.paimon.flink.sink; +package org.apache.paimon.utils; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Blob; @@ -31,24 +31,43 @@ 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; +import org.apache.paimon.types.RowKind; -/** Reattaches a descriptor reader to BLOBs which passed through Flink row serialization. */ -final class BlobDescriptorResolvingRow extends PartialRow { +import javax.annotation.Nullable; - private final InternalRow wrapped; +import java.util.BitSet; + +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** Reattaches a descriptor reader to BLOBs in a row. */ +public class BlobDescriptorResolvingRow implements InternalRow { + + private InternalRow wrapped; private final UriReaderFactory uriReaderFactory; + @Nullable private final BitSet blobDescriptorFieldIndices; + + public BlobDescriptorResolvingRow( + UriReaderFactory uriReaderFactory, @Nullable int[] blobDescriptorFieldIndices) { + this(null, uriReaderFactory, blobDescriptorFieldIndices); + } - BlobDescriptorResolvingRow(InternalRow wrapped, UriReaderFactory uriReaderFactory) { - super(wrapped.getFieldCount(), wrapped); + public BlobDescriptorResolvingRow(InternalRow wrapped, UriReaderFactory uriReaderFactory) { + this(wrapped, uriReaderFactory, null); + } + + public BlobDescriptorResolvingRow( + InternalRow wrapped, + UriReaderFactory uriReaderFactory, + @Nullable int[] blobDescriptorFieldIndices) { this.wrapped = wrapped; - this.uriReaderFactory = uriReaderFactory; + this.uriReaderFactory = + checkNotNull(uriReaderFactory, "UriReaderFactory must not be null."); + this.blobDescriptorFieldIndices = toBitSet(blobDescriptorFieldIndices); } - @Override public BlobDescriptorResolvingRow replace(InternalRow row) { - throw new UnsupportedOperationException("Replacing the wrapped row is not supported."); + this.wrapped = row; + return this; } @Override @@ -56,12 +75,105 @@ final class BlobDescriptorResolvingRow extends PartialRow { 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(super.getBlob(pos), uriReaderFactory); + Blob blob = wrapped.getBlob(pos); + return shouldResolve(pos) ? withReader(blob, uriReaderFactory) : blob; + } + + private boolean shouldResolve(int pos) { + return blobDescriptorFieldIndices == null || blobDescriptorFieldIndices.get(pos); + } + + @Nullable + private static BitSet toBitSet(@Nullable int[] blobDescriptorFieldIndices) { + if (blobDescriptorFieldIndices == null) { + return null; + } + + BitSet bitSet = new BitSet(); + for (int blobDescriptorFieldIndex : blobDescriptorFieldIndices) { + bitSet.set(blobDescriptorFieldIndex); + } + return bitSet; } - private static Blob withReader(Blob blob, UriReaderFactory uriReaderFactory) { + private static Blob withReader(@Nullable Blob blob, UriReaderFactory uriReaderFactory) { if (!(blob instanceof BlobRef)) { return blob; } @@ -72,24 +184,40 @@ final class BlobDescriptorResolvingRow extends PartialRow { @Override public InternalArray getArray(int pos) { - return withReader(super.getArray(pos), uriReaderFactory); + InternalArray array = wrapped.getArray(pos); + return shouldResolve(pos) ? withReader(array, uriReaderFactory) : array; + } + + @Override + public InternalVector getVector(int pos) { + return wrapped.getVector(pos); } @Override public InternalMap getMap(int pos) { - return withReader(super.getMap(pos), uriReaderFactory); + InternalMap map = wrapped.getMap(pos); + return shouldResolve(pos) ? withReader(map, uriReaderFactory) : map; + } + + @Override + public InternalRow getRow(int pos, int numFields) { + InternalRow row = wrapped.getRow(pos, numFields); + return row == null || !shouldResolve(pos) + ? row + : new BlobDescriptorResolvingRow(row, uriReaderFactory); } private static InternalArray withReader( - InternalArray array, UriReaderFactory uriReaderFactory) { - if (array == BlobArrayPlaceholder.INSTANCE) { + @Nullable InternalArray array, UriReaderFactory uriReaderFactory) { + if (array == null || array == BlobArrayPlaceholder.INSTANCE) { return array; } return new BlobDescriptorResolvingArray(array, uriReaderFactory); } - private static InternalMap withReader(InternalMap map, UriReaderFactory uriReaderFactory) { - if (map == BlobMapPlaceholder.INSTANCE) { + private static InternalMap withReader( + @Nullable InternalMap map, UriReaderFactory uriReaderFactory) { + if (map == null || map == BlobMapPlaceholder.INSTANCE) { return map; } return new BlobDescriptorResolvingMap(map, uriReaderFactory); @@ -198,7 +326,8 @@ final class BlobDescriptorResolvingRow extends PartialRow { @Override public InternalRow getRow(int pos, int numFields) { - return wrapped.getRow(pos, numFields); + InternalRow row = wrapped.getRow(pos, numFields); + return row == null ? null : new BlobDescriptorResolvingRow(row, uriReaderFactory); } @Override @@ -254,7 +383,7 @@ final class BlobDescriptorResolvingRow extends PartialRow { @Override public InternalArray keyArray() { - return wrapped.keyArray(); + return withReader(wrapped.keyArray(), uriReaderFactory); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReadUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReadUtils.java new file mode 100644 index 0000000000..cebb504977 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/BlobDescriptorReadUtils.java @@ -0,0 +1,67 @@ +/* + * 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.table; + +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.UriReaderFactory; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +/** Utilities for resolving BLOB descriptors during reads. */ +public final class BlobDescriptorReadUtils { + + private BlobDescriptorReadUtils() {} + + public static UriReaderFactory createUriReaderFactory( + Table table, int[] blobDescriptorFieldIndices) { + if (blobDescriptorFieldIndices.length == 0) { + return null; + } + + return table instanceof FileStoreTable + ? BlobDescriptorReaderFactory.create((FileStoreTable) table) + : null; + } + + public static int[] blobDescriptorFieldIndices( + Table table, RowType readType, boolean blobAsDescriptor) { + if (blobAsDescriptor || !(table instanceof FileStoreTable)) { + return new int[0]; + } + + Set<String> blobDescriptorFields = + ((FileStoreTable) table).coreOptions().blobDescriptorField(); + List<Integer> indices = new ArrayList<>(); + List<DataField> fields = readType.getFields(); + for (int i = 0; i < fields.size(); i++) { + if (blobDescriptorFields.contains(fields.get(i).name())) { + indices.add(i); + } + } + + int[] result = new int[indices.size()]; + for (int i = 0; i < indices.size(); i++) { + result[i] = indices.get(i); + } + return result; + } +} 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 15b9752a41..28ac48b4c9 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 bd479d4fbf..f778971440 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,20 @@ case class PaimonBatch( inputPartitions: Seq[PaimonInputPartition], readBuilder: ReadBuilder, blobAsDescriptor: Boolean, - metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty) + metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty)( + uriReaderFactory: UriReaderFactory = null, + blobDescriptorFieldIndices: Array[Int] = Array.empty[Int]) extends Batch { override def planInputPartitions(): Array[InputPartition] = inputPartitions.map(_.asInstanceOf[InputPartition]).toArray override def createReaderFactory(): PartitionReaderFactory = - PaimonPartitionReaderFactory(readBuilder, metadataColumns, blobAsDescriptor) + PaimonPartitionReaderFactory( + readBuilder = readBuilder, + metadataColumns = metadataColumns, + blobAsDescriptor = blobAsDescriptor, + uriReaderFactory = uriReaderFactory, + blobDescriptorFieldIndices = blobDescriptorFieldIndices + ) } 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 fd8a6178c4..02ac56f578 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 @@ -26,6 +26,7 @@ import org.apache.paimon.spark.schema.PaimonMetadataColumn import org.apache.paimon.spark.util.SplitUtils import org.apache.paimon.table.source.{ReadBuilder, Split} import org.apache.paimon.types.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 +42,9 @@ case class PaimonPartitionReader( readBuilder: ReadBuilder, partition: PaimonInputPartition, metadataColumns: Seq[PaimonMetadataColumn], - blobAsDescriptor: Boolean + blobAsDescriptor: Boolean, + uriReaderFactory: UriReaderFactory, + blobDescriptorFieldIndices: Array[Int] ) 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(uriReaderFactory, blobDescriptorFieldIndices) 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) } } 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 64630574a9..ebe483d692 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,22 @@ import java.util.Objects case class PaimonPartitionReaderFactory( readBuilder: ReadBuilder, metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty, - blobAsDescriptor: Boolean) + blobAsDescriptor: Boolean, + uriReaderFactory: UriReaderFactory = null, + blobDescriptorFieldIndices: Array[Int] = Array.empty[Int]) extends PartitionReaderFactory { override def createReader(partition: InputPartition): PartitionReader[InternalRow] = { partition match { case paimonInputPartition: PaimonInputPartition => - PaimonPartitionReader(readBuilder, paimonInputPartition, metadataColumns, blobAsDescriptor) + PaimonPartitionReader( + readBuilder = readBuilder, + partition = paimonInputPartition, + metadataColumns = metadataColumns, + blobAsDescriptor = blobAsDescriptor, + uriReaderFactory = uriReaderFactory, + blobDescriptorFieldIndices = blobDescriptorFieldIndices + ) 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 e340be2882..5000a7c593 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 @@ -22,8 +22,9 @@ import org.apache.paimon.CoreOptions import org.apache.paimon.partition.PartitionPredicate 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.{BlobDescriptorReadUtils, BucketMode, FileStoreTable, Table} import org.apache.paimon.table.source.{PostponeMergePlan, PostponeMergeReadBuilder} +import org.apache.paimon.utils.UriReaderFactory import scala.collection.JavaConverters._ @@ -81,8 +82,22 @@ 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 blobDescriptorFieldIndices = + BlobDescriptorReadUtils.blobDescriptorFieldIndices( + scan.table, + corePlan.resultReadType(), + blobAsDescriptor) + val uriReaderFactory = + BlobDescriptorReadUtils.createUriReaderFactory(scan.table, blobDescriptorFieldIndices) + mergePlan = MergePlan( + readBuilder = builder, + corePlan = corePlan, + blobAsDescriptor = blobAsDescriptor, + realScanInfo = realScanInfo, + uriReaderFactory = uriReaderFactory, + blobDescriptorFieldIndices = blobDescriptorFieldIndices + ) } mergePlan } @@ -131,7 +146,9 @@ private[spark] object PostponeMergeOnRead { readBuilder: PostponeMergeReadBuilder, corePlan: PostponeMergePlan, blobAsDescriptor: Boolean, - realScanInfo: RealScanInfo) + realScanInfo: RealScanInfo, + uriReaderFactory: UriReaderFactory, + blobDescriptorFieldIndices: Array[Int]) 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 ac160233c5..55194f2c07 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,8 @@ private[spark] case class PostponeMergeOnReadExec( val readBuilder = mergePlan.readBuilder val resultRowType = mergePlan.corePlan.resultReadType() val blobAsDescriptor = mergePlan.blobAsDescriptor + val blobDescriptorFieldIndices = mergePlan.blobDescriptorFieldIndices + val uriReaderFactory = mergePlan.uriReaderFactory val outputAttributes = output val numOutputRows = longMetric(NUM_OUTPUT_ROWS) val numSplits = longMetric(NUM_SPLITS) @@ -129,13 +131,16 @@ private[spark] case class PostponeMergeOnReadExec( rows => val unsafeProjection = UnsafeProjection.create(outputAttributes, outputAttributes) new PostponeMergeOnReadExec.SortedBucketMergeIterator( - rows, - readBuilder, - resultRowType, - blobAsDescriptor, - numSplits, - partitionSize, - readBatchTime) + rows = rows, + readBuilder = readBuilder, + resultRowType = resultRowType, + blobAsDescriptor = blobAsDescriptor, + numSplits = numSplits, + partitionSize = partitionSize, + readBatchTime = readBatchTime, + uriReaderFactory = uriReaderFactory, + blobDescriptorFieldIndices = blobDescriptorFieldIndices + ) .map { row => numOutputRows += 1L @@ -199,7 +204,9 @@ private[spark] object PostponeMergeOnReadExec { blobAsDescriptor: Boolean, numSplits: SQLMetric, partitionSize: SQLMetric, - readBatchTime: SQLMetric) + readBatchTime: SQLMetric, + uriReaderFactory: UriReaderFactory, + blobDescriptorFieldIndices: Array[Int]) extends Iterator[InternalRow] with AutoCloseable { @@ -207,6 +214,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(uriReaderFactory, blobDescriptorFieldIndices) private var currentReader: RecordReaderIterator[PaimonInternalRow] = _ private var currentTimedReader: TimedRecordReader[PaimonInternalRow] = _ private var nextRow: InternalRow = _ @@ -235,13 +244,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 088b785c7b..b7854ea4e2 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 @@ -26,6 +26,7 @@ import org.apache.paimon.spark.schema.PaimonMetadataColumn import org.apache.paimon.spark.schema.PaimonMetadataColumn._ import org.apache.paimon.spark.util.{OptionUtils, SplitUtils} import org.apache.paimon.table.{SpecialFields, Table} +import org.apache.paimon.table.BlobDescriptorReadUtils import org.apache.paimon.table.source.{ReadBuilder, Split} import org.apache.paimon.types.RowType @@ -144,7 +145,18 @@ 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 blobDescriptorFieldIndices = + BlobDescriptorReadUtils.blobDescriptorFieldIndices(table, readTableRowType, blobAsDescriptor) + val uriReaderFactory = + BlobDescriptorReadUtils.createUriReaderFactory(table, blobDescriptorFieldIndices) + PaimonBatch( + inputPartitions = inputPartitions, + readBuilder = readBuilder, + blobAsDescriptor = blobAsDescriptor, + metadataColumns = metadataColumns)( + uriReaderFactory = uriReaderFactory, + blobDescriptorFieldIndices = blobDescriptorFieldIndices) } 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 1730cb6d08..8ebc613dde 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 @@ -22,7 +22,7 @@ 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.table.DataTable +import org.apache.paimon.table.{BlobDescriptorReadUtils, DataTable} import org.apache.paimon.table.source.{DataSplit, OutOfRangeException, ReadBuilder} import org.apache.paimon.utils.DataEvolutionUtils @@ -104,6 +104,13 @@ class PaimonMicroBatchStream( } private lazy val blobAsDescriptor: Boolean = options.get(CoreOptions.BLOB_AS_DESCRIPTOR) + private lazy val blobDescriptorFieldIndices = + BlobDescriptorReadUtils.blobDescriptorFieldIndices( + table, + readBuilder.readType(), + blobAsDescriptor) + private lazy val uriReaderFactory = + BlobDescriptorReadUtils.createUriReaderFactory(table, blobDescriptorFieldIndices) private[spark] lazy val schemaLoader: Function[JLong, TableSchema] = { val schemaManager = table.schemaManager() @@ -210,7 +217,11 @@ class PaimonMicroBatchStream( } override def createReaderFactory(): PartitionReaderFactory = { - PaimonPartitionReaderFactory(readBuilder, blobAsDescriptor = blobAsDescriptor) + PaimonPartitionReaderFactory( + readBuilder, + blobAsDescriptor = blobAsDescriptor, + uriReaderFactory = uriReaderFactory, + blobDescriptorFieldIndices = blobDescriptorFieldIndices) } 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 e56a07a527..bde723cdd0 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.GenericMap; 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,7 +44,10 @@ import org.apache.spark.sql.types.StructType; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.ByteArrayInputStream; +import java.io.IOException; import java.math.BigDecimal; +import java.net.URI; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.time.LocalDate; @@ -164,6 +170,80 @@ public class SparkInternalRowTest { 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 testBlobDescriptorResolvingRowScopesReattachedReader() { + byte[] descriptorBytes = new byte[] {1, 2, 3}; + byte[] managedBytes = new byte[] {4, 5, 6}; + BlobDescriptor descriptor = new BlobDescriptor("scoped://descriptor", 0, -1); + BlobDescriptor managedDescriptor = new BlobDescriptor("scoped://managed", 0, -1); + InternalRow row = + GenericRow.of( + Blob.fromDescriptor(new ByteArrayUriReader(descriptorBytes), descriptor), + Blob.fromDescriptor( + new ByteArrayUriReader(managedBytes), managedDescriptor)); + UriReaderFactory failingReaderFactory = + new UriReaderFactory(null) { + @Override + protected UriReader newReader(URI uri) { + return new UriReader() { + @Override + public SeekableInputStream newInputStream(String uri) + throws IOException { + throw new IOException("Should not reattach managed reader."); + } + }; + } + }; + + BlobDescriptorResolvingRow resolvingRow = + new BlobDescriptorResolvingRow(row, failingReaderFactory, new int[] {0}); + + assertThatThrownBy(() -> resolvingRow.getBlob(0).toData()) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Should not reattach managed reader."); + assertThat(resolvingRow.getBlob(1).toData()).isEqualTo(managedBytes); + } + + private static class ByteArrayUriReader implements UriReader { + + private final byte[] bytes; + + private ByteArrayUriReader(byte[] bytes) { + this.bytes = bytes; + } + + @Override + public SeekableInputStream newInputStream(String uri) { + return SeekableInputStream.wrap(new ByteArrayInputStream(bytes)); + } + } + @Test public void testMapBlobRejectsNullKey() { Map<Object, Object> 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 d4c6965a22..5963e3e63d 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,40 @@ 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.singletonMap(CoreOptions.BLOB_DESCRIPTOR_FIELD.key(), "picture") + 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) + assert(readerFactory.blobDescriptorFieldIndices.sameElements(Array(1))) + } + private def consumerOffset(index: Long, totalSplits: Long): PaimonSourceOffset = { PaimonSourceOffset.withTotalSplits( snapshotId = 5L, @@ -105,4 +143,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 b09a243587..59e028cf54 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( @@ -400,6 +426,15 @@ class BlobTestBase extends PaimonSparkTestBase { Row(2, Array[Byte](89, 69)) ) ) + + sql("DROP TABLE blob_source") + checkAnswer( + sql("SELECT id, picture FROM blob_target ORDER BY id"), + Seq( + Row(1, Array[Byte](72, 101, 108, 108, 111)), + Row(2, Array[Byte](89, 69)) + ) + ) } }
