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
commit 55a1f951859097b875586c0922f20b34e0f44f04 Author: JingsongLi <[email protected]> AuthorDate: Tue Oct 20 15:07:09 2026 +0800 Revert " [common] Support constructing BlobView by field name and unify fieldId resolution (#10606)" This reverts commit 474bc97b762217313fb9c7c7f675891366018d25. --- .../src/main/java/org/apache/paimon/data/Blob.java | 11 ------ .../org/apache/paimon/data/BlobViewStruct.java | 34 ----------------- .../org/apache/paimon/data/BlobViewStructTest.java | 30 --------------- .../org/apache/paimon/append/BlobTableTest.java | 44 +++++++++++++--------- .../org/apache/paimon/rest/RESTCatalogTest.java | 3 +- .../table/PrimaryKeyPartialUpdateBlobTest.java | 20 +++++----- .../paimon/flink/function/BlobViewFunction.java | 21 ++++++++++- .../catalyst/analysis/ReplacePaimonFunctions.scala | 16 ++++++-- 8 files changed, 71 insertions(+), 108 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/data/Blob.java b/paimon-common/src/main/java/org/apache/paimon/data/Blob.java index bb85e8af79..8f320e2369 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/Blob.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/Blob.java @@ -19,12 +19,10 @@ package org.apache.paimon.data; import org.apache.paimon.annotation.Public; -import org.apache.paimon.catalog.Identifier; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.local.LocalFileIO; -import org.apache.paimon.types.RowType; import org.apache.paimon.utils.UriReader; import org.apache.paimon.utils.UriReaderFactory; @@ -81,15 +79,6 @@ public interface Blob { return new BlobView(viewStruct); } - static BlobView fromView(Identifier identifier, int fieldId, long rowId) { - return fromView(new BlobViewStruct(identifier, fieldId, rowId)); - } - - static BlobView fromView( - Identifier identifier, RowType upstreamRowType, String fieldName, long rowId) { - return fromView(BlobViewStruct.of(identifier, upstreamRowType, fieldName, rowId)); - } - static Blob fromBytes( byte[] bytes, @Nullable UriReaderFactory uriReaderFactory, @Nullable FileIO fileIO) { return fromBytes(bytes, uriReaderFactory, fileIO, true); diff --git a/paimon-common/src/main/java/org/apache/paimon/data/BlobViewStruct.java b/paimon-common/src/main/java/org/apache/paimon/data/BlobViewStruct.java index 62533dcd1f..d53d49e446 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/BlobViewStruct.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/BlobViewStruct.java @@ -19,9 +19,6 @@ package org.apache.paimon.data; import org.apache.paimon.catalog.Identifier; -import org.apache.paimon.types.DataField; -import org.apache.paimon.types.DataTypeRoot; -import org.apache.paimon.types.RowType; import java.io.Serializable; import java.nio.ByteBuffer; @@ -55,37 +52,6 @@ public class BlobViewStruct implements Serializable { this.rowId = rowId; } - public static BlobViewStruct of( - Identifier identifier, RowType upstreamRowType, String fieldName, long rowId) { - return new BlobViewStruct( - identifier, resolveBlobFieldId(identifier, upstreamRowType, fieldName), rowId); - } - - public static int resolveBlobFieldId( - Identifier identifier, RowType upstreamRowType, String fieldName) { - Objects.requireNonNull(identifier, "identifier"); - Objects.requireNonNull(upstreamRowType, "upstreamRowType"); - Objects.requireNonNull(fieldName, "fieldName"); - if (!upstreamRowType.containsField(fieldName)) { - throw new IllegalArgumentException( - "Cannot find blob field " - + fieldName - + " in upstream table " - + identifier.getFullName() - + "."); - } - DataField field = upstreamRowType.getField(fieldName); - if (!field.type().is(DataTypeRoot.BLOB)) { - throw new IllegalArgumentException( - "Field " - + fieldName - + " in upstream table " - + identifier.getFullName() - + " is not a BLOB field."); - } - return field.id(); - } - public Identifier identifier() { return identifier; } diff --git a/paimon-common/src/test/java/org/apache/paimon/data/BlobViewStructTest.java b/paimon-common/src/test/java/org/apache/paimon/data/BlobViewStructTest.java index 952526b88e..de68d482c2 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/BlobViewStructTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/BlobViewStructTest.java @@ -19,9 +19,6 @@ package org.apache.paimon.data; import org.apache.paimon.catalog.Identifier; -import org.apache.paimon.types.DataField; -import org.apache.paimon.types.DataTypes; -import org.apache.paimon.types.RowType; import org.junit.jupiter.api.Test; @@ -47,33 +44,6 @@ public class BlobViewStructTest { assertThat(deserialized.rowId()).isEqualTo(5L); } - @Test - public void testConstructByFieldName() { - Identifier identifier = Identifier.fromString("default.source"); - RowType upstreamRowType = - new RowType( - Arrays.asList( - new DataField(0, "id", DataTypes.INT()), - new DataField(7, "picture", DataTypes.BLOB()))); - - BlobViewStruct viewStruct = BlobViewStruct.of(identifier, upstreamRowType, "picture", 5L); - assertThat(viewStruct.identifier()).isEqualTo(identifier); - assertThat(viewStruct.fieldId()).isEqualTo(7); - assertThat(viewStruct.rowId()).isEqualTo(5L); - - assertThat(Blob.fromView(identifier, upstreamRowType, "picture", 5L)) - .isEqualTo(Blob.fromView(identifier, 7, 5L)) - .isEqualTo(Blob.fromView(viewStruct)); - - assertThatThrownBy(() -> BlobViewStruct.of(identifier, upstreamRowType, "missing", 5L)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Cannot find blob field missing in upstream table default.source."); - - assertThatThrownBy(() -> BlobViewStruct.of(identifier, upstreamRowType, "id", 5L)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Field id in upstream table default.source is not a BLOB field."); - } - @Test public void testRejectUnknownDatabase() { BlobViewStruct viewStruct = diff --git a/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java b/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java index 71e03b953f..f67f3fc582 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java @@ -1142,6 +1142,13 @@ public class BlobTableTest extends TableTestBase { 2, BinaryString.fromString("row2"), new BlobData(imageBytes2)), GenericRow.of(3, BinaryString.fromString("row3"), null))); + int imageFieldId = + upstreamTable.rowType().getFields().stream() + .filter(f -> f.name().equals("image")) + .findFirst() + .orElseThrow(() -> new RuntimeException("image field not found")) + .id(); + RowTrackingTable upstreamRowTracking = new RowTrackingTable(upstreamTable); ReadBuilder rowIdReader = upstreamRowTracking.newReadBuilder().withProjection(new int[] {0, 3}); @@ -1173,7 +1180,7 @@ public class BlobTableTest extends TableTestBase { catalog.createTable(identifier(downstreamTableName), downstreamSchema.build(), true); FileStoreTable downstreamTable = getTable(identifier(downstreamTableName)); - Identifier upstreamIdentifier = Identifier.fromString(database + "." + upstreamTableName); + String upstreamFullName = database + "." + upstreamTableName; writeRows( downstreamTable, Arrays.asList( @@ -1181,26 +1188,26 @@ public class BlobTableTest extends TableTestBase { 1, BinaryString.fromString("label1"), Blob.fromView( - upstreamIdentifier, - upstreamTable.rowType(), - "image", - idToRowId.get(1))), + new BlobViewStruct( + Identifier.fromString(upstreamFullName), + imageFieldId, + idToRowId.get(1)))), GenericRow.of( 2, BinaryString.fromString("label2"), Blob.fromView( - upstreamIdentifier, - upstreamTable.rowType(), - "image", - idToRowId.get(2))), + new BlobViewStruct( + Identifier.fromString(upstreamFullName), + imageFieldId, + idToRowId.get(2)))), GenericRow.of( 3, BinaryString.fromString("label3"), Blob.fromView( - upstreamIdentifier, - upstreamTable.rowType(), - "image", - idToRowId.get(3))))); + new BlobViewStruct( + Identifier.fromString(upstreamFullName), + imageFieldId, + idToRowId.get(3)))))); ReadBuilder downstreamReadBuilder = downstreamTable.newReadBuilder(); downstreamReadBuilder @@ -1241,6 +1248,7 @@ public class BlobTableTest extends TableTestBase { GenericRow.of(1, new BlobData(imageBytes1)), GenericRow.of(2, new BlobData(imageBytes2)))); + int imageFieldId = upstreamTable.rowType().getField("image").id(); RowTrackingTable upstreamRowTracking = new RowTrackingTable(upstreamTable); ReadBuilder rowIdReader = upstreamRowTracking.newReadBuilder().withProjection(new int[] {0, 2}); @@ -1262,13 +1270,13 @@ public class BlobTableTest extends TableTestBase { firstViewSchema.option(CoreOptions.BLOB_VIEW_FIELD.key(), "image"); catalog.createTable(identifier(firstViewTableName), firstViewSchema.build(), true); - Identifier upstreamIdentifier = Identifier.fromString(database + "." + upstreamTableName); + String upstreamFullName = database + "." + upstreamTableName; BlobViewStruct viewStruct1 = - BlobViewStruct.of( - upstreamIdentifier, upstreamTable.rowType(), "image", idToRowId.get(1)); + new BlobViewStruct( + Identifier.fromString(upstreamFullName), imageFieldId, idToRowId.get(1)); BlobViewStruct viewStruct2 = - BlobViewStruct.of( - upstreamIdentifier, upstreamTable.rowType(), "image", idToRowId.get(2)); + new BlobViewStruct( + Identifier.fromString(upstreamFullName), imageFieldId, idToRowId.get(2)); FileStoreTable firstViewTable = getTable(identifier(firstViewTableName)); writeRows( firstViewTable, diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java index 74d898adf6..ad6f315261 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java @@ -32,6 +32,7 @@ import org.apache.paimon.consumer.ConsumerInfo; import org.apache.paimon.consumer.ConsumerManager; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Blob; +import org.apache.paimon.data.BlobViewStruct; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.serializer.InternalRowSerializer; @@ -4797,7 +4798,7 @@ public abstract class RESTCatalogTest extends CatalogTestBase { write.write( GenericRow.of( BinaryString.fromString("d1"), - Blob.fromView(identifier, table.rowType(), "image", 0L))); + Blob.fromView(new BlobViewStruct(identifier, 1, 0L)))); BatchTableCommit commit = writeBuilder.newCommit(); commit.commit(write.prepareCommit()); write.close(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java index 1484d1befc..c5ef165282 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyPartialUpdateBlobTest.java @@ -26,6 +26,7 @@ import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobData; import org.apache.paimon.data.BlobView; +import org.apache.paimon.data.BlobViewStruct; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; @@ -274,6 +275,7 @@ public class PrimaryKeyPartialUpdateBlobTest extends TableTestBase { GenericRow.of(1, new BlobData(imageBytes)), GenericRow.of(2, new BlobData(secondImageBytes))); + int imageFieldId = upstreamTable.rowType().getField("image").id(); RowTrackingTable upstreamRowTracking = new RowTrackingTable(upstreamTable); ReadBuilder rowIdReader = upstreamRowTracking.newReadBuilder().withProjection(new int[] {0, 2}); @@ -298,25 +300,25 @@ public class PrimaryKeyPartialUpdateBlobTest extends TableTestBase { catalog.createTable(identifier(downstreamName), downstreamSchema, true); FileStoreTable downstreamTable = getTable(identifier(downstreamName)); - Identifier upstreamIdentifier = Identifier.fromString(database + "." + upstreamName); + String upstreamFullName = database + "." + upstreamName; write( downstreamTable, GenericRow.of( 1, BinaryString.fromString("label1"), Blob.fromView( - upstreamIdentifier, - upstreamTable.rowType(), - "image", - idToRowId.get(1))), + new BlobViewStruct( + Identifier.fromString(upstreamFullName), + imageFieldId, + idToRowId.get(1)))), GenericRow.of( 2, BinaryString.fromString("label2"), Blob.fromView( - upstreamIdentifier, - upstreamTable.rowType(), - "image", - idToRowId.get(2)))); + new BlobViewStruct( + Identifier.fromString(upstreamFullName), + imageFieldId, + idToRowId.get(2))))); try (StreamTableWrite write = downstreamTable.newWrite(commitUser); StreamTableCommit commit = downstreamTable.newCommit(commitUser)) { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/function/BlobViewFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/function/BlobViewFunction.java index 68d490adb8..0f9b4b33ab 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/function/BlobViewFunction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/function/BlobViewFunction.java @@ -26,6 +26,8 @@ import org.apache.paimon.data.BlobViewStruct; import org.apache.paimon.flink.FlinkFileIOLoader; import org.apache.paimon.options.Options; import org.apache.paimon.table.Table; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypeRoot; import org.apache.flink.table.functions.FunctionContext; import org.apache.flink.table.functions.ScalarFunction; @@ -104,7 +106,24 @@ public class BlobViewFunction extends ScalarFunction implements CatalogAwareFunc private int fieldId(Identifier identifier, String fieldName) { try { Table table = catalog().getTable(identifier); - return BlobViewStruct.resolveBlobFieldId(identifier, table.rowType(), fieldName); + if (!table.rowType().containsField(fieldName)) { + throw new IllegalArgumentException( + "Cannot find blob field " + + fieldName + + " in upstream table " + + identifier.getFullName() + + "."); + } + DataField field = table.rowType().getField(fieldName); + if (!field.type().is(DataTypeRoot.BLOB)) { + throw new IllegalArgumentException( + "Field " + + fieldName + + " in upstream table " + + identifier.getFullName() + + " is not a BLOB field."); + } + return field.id(); } catch (Catalog.TableNotExistException e) { throw new IllegalArgumentException( "Cannot find upstream table " + identifier.getFullName() + ".", e); diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala index 39b98a48a4..3abe903768 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/ReplacePaimonFunctions.scala @@ -18,13 +18,13 @@ package org.apache.paimon.spark.catalyst.analysis -import org.apache.paimon.data.BlobViewStruct import org.apache.paimon.spark.{BaseTable, DataConverter, SparkTable, SparkTypeUtils, SparkUtils} import org.apache.paimon.spark.catalog.SparkBaseCatalog import org.apache.paimon.spark.catalog.functions.PaimonFunctions import org.apache.paimon.spark.function.{BlobViewFieldIdSparkFunction, BlobViewSparkFunction, DescriptorToPresignedUrlFunction, ResolvedDescriptorToPresignedUrlFunction} import org.apache.paimon.spark.utils.CatalogUtils import org.apache.paimon.table.DataTable +import org.apache.paimon.types.DataTypeRoot import org.apache.paimon.utils.{InternalRowUtils, TypeUtils} import org.apache.spark.sql.SparkSession @@ -107,12 +107,20 @@ object ReplacePaimonFunctions { CatalogUtils.toIdentifier( catalogAndIdentifier.identifier(), catalogAndIdentifier.catalog().name()) - val fieldId = - BlobViewStruct.resolveBlobFieldId(paimonIdentifier, sparkTable.table.rowType(), fieldName) + if (!sparkTable.table.rowType().containsField(fieldName)) { + throw new IllegalArgumentException( + s"Cannot find blob field $fieldName in upstream table ${paimonIdentifier.getFullName}.") + } + val field = sparkTable.table.rowType().getField(fieldName) + if (!field.`type`().is(DataTypeRoot.BLOB)) { + throw new IllegalArgumentException( + s"Field $fieldName in upstream table ${paimonIdentifier.getFullName} " + + "is not a BLOB field.") + } ApplyFunctionExpression( new BlobViewFieldIdSparkFunction, - Seq(Literal(paimonIdentifier.getFullName), Literal(fieldId), rowId)) + Seq(Literal(paimonIdentifier.getFullName), Literal(field.id()), rowId)) } } }
