beryllw commented on code in PR #3420:
URL: https://github.com/apache/fluss/pull/3420#discussion_r3541159486
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonWriteResultSerializer.java:
##########
@@ -38,21 +42,58 @@ public int getVersion() {
@Override
public byte[] serialize(PaimonWriteResult paimonWriteResult) throws
IOException {
- CommitMessage commitMessage = paimonWriteResult.commitMessage();
- return messageSer.serialize(commitMessage);
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DataOutputViewStreamWrapper view = new
DataOutputViewStreamWrapper(out);
+
+ Long watermark = paimonWriteResult.getWatermark();
+ if (watermark == null) {
+ view.writeBoolean(false);
+ } else {
+ view.writeBoolean(true);
+ view.writeLong(watermark);
+ }
+
+ byte[] messageBytes =
messageSer.serialize(paimonWriteResult.commitMessage());
+ view.writeInt(messageBytes.length);
+ view.write(messageBytes);
+
+ return out.toByteArray();
}
@Override
public PaimonWriteResult deserialize(int version, byte[] serialized)
throws IOException {
- if (version != CURRENT_VERSION) {
- throw new UnsupportedOperationException(
- "Expecting PaimonWriteResult version to be "
- + CURRENT_VERSION
- + ", but found "
- + version
- + ".");
+ switch (version) {
+ case 1:
+ {
+ CommitMessage commitMessage =
+ messageSer.deserialize(messageSer.getVersion(),
serialized);
+ return new PaimonWriteResult(commitMessage, null);
+ }
+ case 2:
+ {
+ DataInputView view = new DataInputDeserializer(serialized);
+
+ Long watermark = null;
+ boolean watermarkNonNull = view.readBoolean();
+ if (watermarkNonNull) {
+ watermark = view.readLong();
+ }
+
+ int len = view.readInt();
+ byte[] messageBytes = new byte[len];
+ view.read(messageBytes);
Review Comment:
Might be slightly safer to use view.readFully(messageBytes) here —
read(byte[]) doesn't guarantee filling the whole array (and its return value is
unused). Not an issue today since messageBytes is the last field, just a nit
for consistency with TableBucketWriteResultSerializer.
--
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]
