[core] Encapsulate split serialization protocols

paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java

@@ -308,11 +308,15 @@ public OptionalLong mergedRowCount() {
  }
 
  private void writeObject(ObjectOutputStream out) throws IOException {
- serialize(new DataOutputViewStreamWrapper(out));
+SplitSerializer.serialize(this, new DataOutputViewStreamWrapper(out));
  }
 
  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
- assign(deserialize(new DataInputViewStreamWrapper(in)));
+ Split split = SplitSerializer.deserialize(new DataInputViewStreamWrapper(in));
+ if (!(split instanceof FallbackSplitImpl)) {
+ throw new IOException("Deserialized split is not a FallbackSplitImpl: " + split);
+ }
+ assign((FallbackSplitImpl) split);
  }
 
  private void assign(FallbackSplitImpl other) {
@@ -321,15 +325,14 @@ private void assign(FallbackSplitImpl other) {
  }
 
  public void serialize(DataOutputView out) throws IOException {
- SplitSerializer.serialize(this, out);
+ out.writeBoolean(isFallback);
+ SplitSerializer.serialize(split, out);
  }
 
  public static FallbackSplitImpl deserialize(DataInputView in) throws IOException {
+ boolean isFallback = in.readBoolean();
  Split split = SplitSerializer.deserialize(in);
- if (!(split instanceof FallbackSplitImpl)) {
- throw new IOException("Deserialized split is not a FallbackSplitImpl: " + split);
- }
- return (FallbackSplitImpl) split;
+ return new FallbackSplitImpl(split, isFallback);
  }
  }
 

paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.io.DataFileMetaSerializer;
 import org.apache.paimon.io.DataInputView;
 import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputView;
 import org.apache.paimon.io.DataOutputViewStreamWrapper;
 import org.apache.paimon.utils.FunctionWithIOException;
 
@@ -191,7 +192,27 @@ public String toString() {
  }
 
  private void writeObject(ObjectOutputStream objectOutputStream) throws IOException {
- DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(objectOutputStream);
+ serialize(new DataOutputViewStreamWrapper(objectOutputStream));
+ }
+
+ private void readObject(ObjectInputStream objectInputStream)
+ throws IOException, ClassNotFoundException {
+ assign(deserialize(new DataInputViewStreamWrapper(objectInputStream)));
+ }
+
+ protected void assign(IncrementalSplit other) {
+ snapshotId = other.snapshotId;
+ partition = other.partition;
+ bucket = other.bucket;
+ totalBuckets = other.totalBuckets;
+ beforeFiles = other.beforeFiles;
+ beforeDeletionFiles = other.beforeDeletionFiles;
+ afterFiles = other.afterFiles;
+ afterDeletionFiles = other.afterDeletionFiles;
+ isStreaming = other.isStreaming;
+ }
+
+ public void serialize(DataOutputView out) throws IOException {
  out.writeInt(VERSION);
  out.writeLong(snapshotId);
  serializeBinaryRow(partition, out);
@@ -216,39 +237,49 @@ private void writeObject(ObjectOutputStream objectOutputStream) throws IOExcepti
  out.writeBoolean(isStreaming);
  }
 
- private void readObject(ObjectInputStream objectInputStream)
- throws IOException, ClassNotFoundException {
- DataInputViewStreamWrapper in = new DataInputViewStreamWrapper(objectInputStream);
+ public static IncrementalSplit deserialize(DataInputView in) throws IOException {
  int version = in.readInt();
  if (version != VERSION) {
  throw new UnsupportedOperationException("Unsupported version: " + version);
  }
 
- snapshotId = in.readLong();
- partition = deserializeBinaryRow(in);
- bucket = in.readInt();
- totalBuckets = in.readInt();
+long snapshotId = in.readLong();
+BinaryRow partition = deserializeBinaryRow(in);
+int bucket = in.readInt();
+int totalBuckets = in.readInt();
 
  DataFileMetaSerializer dataFileMetaSerializer = new DataFileMetaSerializer();
  FunctionWithIOException<DataInputView, DeletionFile> deletionFileSerializer =
  DeletionFile::deserialize;
 
  int beforeNumber = in.readInt();
- beforeFiles = new ArrayList<>(beforeNumber);
+List<DataFileMeta> beforeFiles = new ArrayList<>(beforeNumber);
  for (int i = 0; i < beforeNumber; i++) {
  beforeFiles.add(dataFileMetaSerializer.deserialize(in));
  }
 
- beforeDeletionFiles = DeletionFile.deserializeList(in, deletionFileSerializer);
+ List<DeletionFile> beforeDeletionFiles =
+ DeletionFile.deserializeList(in, deletionFileSerializer);
 
  int fileNumber = in.readInt();
- afterFiles = new ArrayList<>(fileNumber);
+List<DataFileMeta> afterFiles = new ArrayList<>(fileNumber);
  for (int i = 0; i < fileNumber; i++) {
  afterFiles.add(dataFileMetaSerializer.deserialize(in));
  }
 
- afterDeletionFiles = DeletionFile.deserializeList(in, deletionFileSerializer);
+ List<DeletionFile> afterDeletionFiles =
+ DeletionFile.deserializeList(in, deletionFileSerializer);
 
- isStreaming = in.readBoolean();
+ boolean isStreaming = in.readBoolean();
+ return new IncrementalSplit(
+ snapshotId,
+ partition,
+ bucket,
+ totalBuckets,
+ beforeFiles,
+ beforeDeletionFiles,
+ afterFiles,
+ afterDeletionFiles,
+ isStreaming);
  }
 }

paimon-core/src/main/java/org/apache/paimon/table/source/QueryAuthSplit.java

@@ -29,8 +29,13 @@
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.OptionalLong;
+import java.util.TreeMap;
 
 /** A wrapper class for {@link Split} that adds query authorization information. */
 public class QueryAuthSplit implements Split {
@@ -71,11 +76,15 @@ public OptionalLong mergedRowCount() {
  }
 
  private void writeObject(ObjectOutputStream out) throws IOException {
- serialize(new DataOutputViewStreamWrapper(out));
+SplitSerializer.serialize(this, new DataOutputViewStreamWrapper(out));
  }
 
  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
- assign(deserialize(new DataInputViewStreamWrapper(in)));
+ Split split = SplitSerializer.deserialize(new DataInputViewStreamWrapper(in));
+ if (!(split instanceof QueryAuthSplit)) {
+ throw new IOException("Deserialized split is not a QueryAuthSplit: " + split);
+ }
+ assign((QueryAuthSplit) split);
  }
 
  private void assign(QueryAuthSplit other) {
@@ -84,14 +93,113 @@ private void assign(QueryAuthSplit other) {
  }
 
  public void serialize(DataOutputView out) throws IOException {
- SplitSerializer.serialize(this, out);
+ SplitSerializer.serialize(split, out);
+ writeAuthResult(out, authResult);
  }
 
  public static QueryAuthSplit deserialize(DataInputView in) throws IOException {
  Split split = SplitSerializer.deserialize(in);
- if (!(split instanceof QueryAuthSplit)) {
- throw new IOException("Deserialized split is not a QueryAuthSplit: " + split);
+ return new QueryAuthSplit(split, readAuthResult(in));
+ }
+
+ private static void writeAuthResult(
+ DataOutputView out, @Nullable TableQueryAuthResult authResult) throws IOException {
+ if (authResult == null) {
+ out.writeBoolean(false);
+ return;
+ }
+
+ out.writeBoolean(true);
+ writeStringList(out, authResult.filter());
+ writeStringMap(out, authResult.columnMasking());
+ }
+
+ @Nullable
+ private static TableQueryAuthResult readAuthResult(DataInputView in) throws IOException {
+ if (!in.readBoolean()) {
+ return null;
+ }
+ return new TableQueryAuthResult(readStringList(in), readNullableStringMap(in));
+ }
+
+ private static void writeStringList(DataOutputView out, @Nullable List<String> strings)
+ throws IOException {
+ if (strings == null) {
+ out.writeBoolean(false);
+ return;
+ }
+
+ out.writeBoolean(true);
+ out.writeInt(strings.size());
+ for (String string : strings) {
+ writeString(out, string);
  }
- return (QueryAuthSplit) split;
+ }
+
+ @Nullable
+ private static List<String> readStringList(DataInputView in) throws IOException {
+ if (!in.readBoolean()) {
+ return null;
+ }
+
+ int size = in.readInt();
+ List<String> strings = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ strings.add(readString(in));
+ }
+ return strings;
+ }
+
+ private static void writeStringMap(DataOutputView out, @Nullable Map<String, String> map)
+ throws IOException {
+ if (map == null) {
+ out.writeBoolean(false);
+ return;
+ }
+
+ out.writeBoolean(true);
+ out.writeInt(map.size());
+ for (Map.Entry<String, String> entry : new TreeMap<>(map).entrySet()) {
+ writeString(out, entry.getKey());
+ writeString(out, entry.getValue());
+ }
+ }
+
+ @Nullable
+ private static Map<String, String> readNullableStringMap(DataInputView in) throws IOException {
+ if (!in.readBoolean()) {
+ return null;
+ }
+
+ int size = in.readInt();
+ Map<String, String> map = new HashMap<>(size);
+ for (int i = 0; i < size; i++) {
+ map.put(readString(in), readString(in));
+ }
+ return map;
+ }
+
+ private static void writeString(DataOutputView out, @Nullable String string)
+ throws IOException {
+ if (string == null) {
+ out.writeInt(-1);
+ return;
+ }
+
+ byte[] bytes = string.getBytes(StandardCharsets.UTF_8);
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
+
+ @Nullable
+ private static String readString(DataInputView in) throws IOException {
+ int length = in.readInt();
+ if (length < 0) {
+ return null;
+ }
+
+ byte[] bytes = new byte[length];
+ in.readFully(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
  }
 }