Copilot commented on code in PR #3500:
URL: https://github.com/apache/fluss/pull/3500#discussion_r3445290510
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializer.java:
##########
@@ -46,6 +48,11 @@ public byte[] serialize(PaimonSplit paimonSplit) throws
IOException {
DataSplit dataSplit = paimonSplit.dataSplit();
InstantiationUtil.serializeObject(view, dataSplit);
view.writeBoolean(paimonSplit.isBucketUnAware());
+ List<String> partition = paimonSplit.partition();
+ view.writeInt(partition.size());
+ for (String value : partition) {
+ view.writeUTF(value);
+ }
Review Comment:
The payload written in `serialize` now includes partition values, but
`getVersion()` still returns 1. This will make versioning incorrect and will
also break compatibility once a proper version bump is introduced (older v1
payloads don't contain the new fields). Please bump the serializer version
(e.g., to 2) and keep `deserialize` support for VERSION_1.
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplit.java:
##########
@@ -52,19 +53,7 @@ public int bucket() {
@Override
public List<String> partition() {
- BinaryRow partition = dataSplit.partition();
- if (partition.getFieldCount() == 0) {
- return Collections.emptyList();
- }
-
- List<String> partitions = new ArrayList<>();
- for (int i = 0; i < partition.getFieldCount(); i++) {
- // Todo Currently, partition column must be String datatype, so we
can always use
- // consider it as string. Revisit here when
- // #489 is finished.
- partitions.add(partition.getString(i).toString());
- }
- return partitions;
+ return partition;
}
Review Comment:
Because `serialVersionUID` stayed the same and a new field (`partition`) was
added, Java-deserializing older `PaimonSplit` instances will set `partition` to
`null` (constructors aren't called). That can lead to NPEs when `partition()`
is used. Consider adding a `readResolve()` (or bumping `serialVersionUID`) to
ensure `partition` is always non-null, and include `partition` in `toString()`
for easier debugging.
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonSplitSerializer.java:
##########
@@ -59,7 +66,12 @@ public PaimonSplit deserialize(int version, byte[]
serialized) throws IOExceptio
if (version == VERSION_1) {
DataInputStream dis = new DataInputStream(in);
boolean isBucketUnAware = dis.readBoolean();
- return new PaimonSplit(dataSplit, isBucketUnAware);
+ int size = dis.readInt();
+ List<String> partition = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ partition.add(dis.readUTF());
+ }
+ return new PaimonSplit(dataSplit, isBucketUnAware, partition);
} else {
throw new IOException("Unsupported PaimonSplit serialization
version: " + version);
}
Review Comment:
`deserialize` currently reads the appended partition fields under
`VERSION_1`. If this serializer version is bumped (as it should be), restoring
from older VERSION_1 bytes will hit EOF because those bytes only contain the
boolean after the DataSplit. Handle VERSION_1 (boolean only) separately and
read the partition list only for the new version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
