This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 474bc97b76 [common] Support constructing BlobView by field name and
unify fieldId resolution (#10606)
474bc97b76 is described below
commit 474bc97b762217313fb9c7c7f675891366018d25
Author: zhuanshenbsj1 <[email protected]>
AuthorDate: Fri Oct 16 21:56:35 2026 +0800
[common] Support constructing BlobView by field name and unify fieldId
resolution (#10606)
---
.../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, 108 insertions(+), 71 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 8f320e2369..bb85e8af79 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,10 +19,12 @@
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;
@@ -79,6 +81,15 @@ 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 d53d49e446..62533dcd1f 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,6 +19,9 @@
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;
@@ -52,6 +55,37 @@ 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 de68d482c2..952526b88e 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,6 +19,9 @@
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;
@@ -44,6 +47,33 @@ 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 f67f3fc582..71e03b953f 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,13 +1142,6 @@ 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});
@@ -1180,7 +1173,7 @@ public class BlobTableTest extends TableTestBase {
catalog.createTable(identifier(downstreamTableName),
downstreamSchema.build(), true);
FileStoreTable downstreamTable =
getTable(identifier(downstreamTableName));
- String upstreamFullName = database + "." + upstreamTableName;
+ Identifier upstreamIdentifier = Identifier.fromString(database + "." +
upstreamTableName);
writeRows(
downstreamTable,
Arrays.asList(
@@ -1188,26 +1181,26 @@ public class BlobTableTest extends TableTestBase {
1,
BinaryString.fromString("label1"),
Blob.fromView(
- new BlobViewStruct(
-
Identifier.fromString(upstreamFullName),
- imageFieldId,
- idToRowId.get(1)))),
+ upstreamIdentifier,
+ upstreamTable.rowType(),
+ "image",
+ idToRowId.get(1))),
GenericRow.of(
2,
BinaryString.fromString("label2"),
Blob.fromView(
- new BlobViewStruct(
-
Identifier.fromString(upstreamFullName),
- imageFieldId,
- idToRowId.get(2)))),
+ upstreamIdentifier,
+ upstreamTable.rowType(),
+ "image",
+ idToRowId.get(2))),
GenericRow.of(
3,
BinaryString.fromString("label3"),
Blob.fromView(
- new BlobViewStruct(
-
Identifier.fromString(upstreamFullName),
- imageFieldId,
- idToRowId.get(3))))));
+ upstreamIdentifier,
+ upstreamTable.rowType(),
+ "image",
+ idToRowId.get(3)))));
ReadBuilder downstreamReadBuilder = downstreamTable.newReadBuilder();
downstreamReadBuilder
@@ -1248,7 +1241,6 @@ 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});
@@ -1270,13 +1262,13 @@ public class BlobTableTest extends TableTestBase {
firstViewSchema.option(CoreOptions.BLOB_VIEW_FIELD.key(), "image");
catalog.createTable(identifier(firstViewTableName),
firstViewSchema.build(), true);
- String upstreamFullName = database + "." + upstreamTableName;
+ Identifier upstreamIdentifier = Identifier.fromString(database + "." +
upstreamTableName);
BlobViewStruct viewStruct1 =
- new BlobViewStruct(
- Identifier.fromString(upstreamFullName), imageFieldId,
idToRowId.get(1));
+ BlobViewStruct.of(
+ upstreamIdentifier, upstreamTable.rowType(), "image",
idToRowId.get(1));
BlobViewStruct viewStruct2 =
- new BlobViewStruct(
- Identifier.fromString(upstreamFullName), imageFieldId,
idToRowId.get(2));
+ BlobViewStruct.of(
+ upstreamIdentifier, upstreamTable.rowType(), "image",
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 ad6f315261..74d898adf6 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,7 +32,6 @@ 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;
@@ -4798,7 +4797,7 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
write.write(
GenericRow.of(
BinaryString.fromString("d1"),
- Blob.fromView(new BlobViewStruct(identifier, 1, 0L))));
+ Blob.fromView(identifier, table.rowType(), "image",
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 c5ef165282..1484d1befc 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,7 +26,6 @@ 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;
@@ -275,7 +274,6 @@ 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});
@@ -300,25 +298,25 @@ public class PrimaryKeyPartialUpdateBlobTest extends
TableTestBase {
catalog.createTable(identifier(downstreamName), downstreamSchema,
true);
FileStoreTable downstreamTable = getTable(identifier(downstreamName));
- String upstreamFullName = database + "." + upstreamName;
+ Identifier upstreamIdentifier = Identifier.fromString(database + "." +
upstreamName);
write(
downstreamTable,
GenericRow.of(
1,
BinaryString.fromString("label1"),
Blob.fromView(
- new BlobViewStruct(
-
Identifier.fromString(upstreamFullName),
- imageFieldId,
- idToRowId.get(1)))),
+ upstreamIdentifier,
+ upstreamTable.rowType(),
+ "image",
+ idToRowId.get(1))),
GenericRow.of(
2,
BinaryString.fromString("label2"),
Blob.fromView(
- new BlobViewStruct(
-
Identifier.fromString(upstreamFullName),
- imageFieldId,
- idToRowId.get(2)))));
+ upstreamIdentifier,
+ upstreamTable.rowType(),
+ "image",
+ 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 0f9b4b33ab..68d490adb8 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,8 +26,6 @@ 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;
@@ -106,24 +104,7 @@ public class BlobViewFunction extends ScalarFunction
implements CatalogAwareFunc
private int fieldId(Identifier identifier, String fieldName) {
try {
Table table = catalog().getTable(identifier);
- 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();
+ return BlobViewStruct.resolveBlobFieldId(identifier,
table.rowType(), fieldName);
} 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 3abe903768..39b98a48a4 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,20 +107,12 @@ object ReplacePaimonFunctions {
CatalogUtils.toIdentifier(
catalogAndIdentifier.identifier(),
catalogAndIdentifier.catalog().name())
- 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.")
- }
+ val fieldId =
+ BlobViewStruct.resolveBlobFieldId(paimonIdentifier,
sparkTable.table.rowType(), fieldName)
ApplyFunctionExpression(
new BlobViewFieldIdSparkFunction,
- Seq(Literal(paimonIdentifier.getFullName), Literal(field.id()), rowId))
+ Seq(Literal(paimonIdentifier.getFullName), Literal(fieldId), rowId))
}
}
}
