fix failed cases

Author: wuchong

fluss-common/src/main/java/org/apache/fluss/row/BinaryWriter.java

@@ -114,7 +114,7 @@ static BinaryWriter.ValueWriter createValueWriter(
      *
      * @param elementType the element type
      */
-    private static BinaryWriter.ValueWriter createNotNullValueWriter(
+    static BinaryWriter.ValueWriter createNotNullValueWriter(
             DataType elementType, @Nullable BinaryRowFormat rowFormat) {
         switch (elementType.getTypeRoot()) {
             case CHAR:

fluss-common/src/main/java/org/apache/fluss/row/serializer/ArraySerializer.java

@@ -26,10 +26,15 @@
 import org.apache.fluss.row.array.AlignedArray;
 import org.apache.fluss.row.array.CompactedArray;
 import org.apache.fluss.row.array.IndexedArray;
+import org.apache.fluss.row.array.PrimitiveBinaryArray;
 import org.apache.fluss.types.DataType;
 
 import java.io.Serializable;
 
+import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.ALIGNED;
+import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.COMPACTED;
+import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.INDEXED;
+
 /** Serializer for {@link InternalArray} to {@link BinaryArray} and {@code CompactedArray}. */
 public class ArraySerializer implements Serializable {
     private static final long serialVersionUID = 1L;
@@ -76,7 +81,13 @@ private class BinaryArraySerializer {
 
         public BinaryArray toAlignedArray(InternalArray from) {
             if (from instanceof BinaryArray) {
-                return (BinaryArray) from;
+                if (from instanceof PrimitiveBinaryArray
+                        || rowFormat == INDEXED && from instanceof IndexedArray
+                        || rowFormat == COMPACTED && from instanceof CompactedArray
+                        || rowFormat == ALIGNED && from instanceof AlignedArray) {
+                    // directly return the original array iff the array is in the expected format
+                    return (BinaryArray) from;
+                }
             }
 
             if (from instanceof GenericArray) {

fluss-common/src/main/java/org/apache/fluss/row/serializer/RowSerializer.java

@@ -20,14 +20,21 @@
 import org.apache.fluss.row.BinaryRow;
 import org.apache.fluss.row.BinaryRow.BinaryRowFormat;
 import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.aligned.AlignedRow;
+import org.apache.fluss.row.compacted.CompactedRow;
 import org.apache.fluss.row.encode.AlignedRowEncoder;
 import org.apache.fluss.row.encode.CompactedRowEncoder;
 import org.apache.fluss.row.encode.IndexedRowEncoder;
 import org.apache.fluss.row.encode.RowEncoder;
+import org.apache.fluss.row.indexed.IndexedRow;
 import org.apache.fluss.types.DataType;
 
 import java.io.Serializable;
 
+import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.ALIGNED;
+import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.COMPACTED;
+import static org.apache.fluss.row.BinaryRow.BinaryRowFormat.INDEXED;
+
 /** Serializer for {@link InternalRow} to {@link BinaryRow}. */
 public class RowSerializer implements Serializable {
     private static final long serialVersionUID = 1L;
@@ -53,7 +60,12 @@ public RowSerializer(DataType[] fieldTypes, BinaryRowFormat format) {
      */
     public BinaryRow toBinaryRow(InternalRow from) {
         if (from instanceof BinaryRow) {
-            return (BinaryRow) from;
+            if (format == INDEXED && from instanceof IndexedRow
+                    || format == COMPACTED && from instanceof CompactedRow
+                    || format == ALIGNED && from instanceof AlignedRow) {
+                // directly return the original row iff the row is in the expected format
+                return (BinaryRow) from;
+            }
         }
 
         if (serializer == null) {