fix rebase issues

Author: polyzos

fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java

@@ -1565,10 +1565,27 @@ void testPkCompactedProject() throws Exception {
             }
             upsert.flush();
 
-            // Creating a projected log scanner for COMPACTED should fail
-            assertThatThrownBy(() -> createLogScanner(table, new int[] {0, 2}))
-                    .isInstanceOf(IllegalArgumentException.class)
-                    .hasMessageContaining("Projection is not supported for COMPACTED log format");
+            // Creating a projected log scanner for COMPACTED should work
+            try (LogScanner scanner = createLogScanner(table, new int[] {0, 2})) {
+                subscribeFromBeginning(scanner, table);
+                int seen = 0;
+                while (seen < 10) {
+                    ScanRecords records = scanner.poll(Duration.ofSeconds(5));
+                    for (ScanRecord record : records) {
+                        InternalRow row = record.getRow();
+                        assertThat(row.getFieldCount()).isEqualTo(2);
+                        int a = row.getInt(0);
+                        assertThat(a).isEqualTo(seen);
+                        if (a % 2 == 0) {
+                            assertThat(row.getString(1).toString()).isEqualTo("v" + a);
+                        } else {
+                            assertThat(row.isNullAt(1)).isTrue();
+                        }
+                        seen++;
+                    }
+                }
+                assertThat(seen).isEqualTo(10);
+            }
         }
     }
 

fluss-common/src/main/java/org/apache/fluss/record/CompactedLogRecord.java

@@ -32,14 +32,15 @@
 import java.io.IOException;
 
 import static org.apache.fluss.record.LogRecordBatchFormat.LENGTH_LENGTH;
+
 /**
- * An immutable log record for @CompactedRow which can be directly persisted. The on-wire schema is
- * identical to IndexedLogRecord but the row payload uses the CompactedRow binary format:
+ * An immutable log record for {@link CompactedRow} which can be directly persisted. The on-wire
+ * schema is identical to IndexedLogRecord but the row payload uses the CompactedRow binary format:
  *
  * <ul>
- *   <li>Length => int32 (total number of bytes following this length field)</li>
- *   <li>Attributes => int8 (low 4 bits encode {@link ChangeType})</li>
- *   <li>Value => {@link CompactedRow} (bytes in compacted row format)</li>
+ *   <li>Length => int32 (total number of bytes following this length field)
+ *   <li>Attributes => int8 (low 4 bits encode {@link ChangeType})
+ *   <li>Value => {@link CompactedRow} (bytes in compacted row format)
  * </ul>
  *
  * <p>Differences vs {@link IndexedLogRecord}: - Uses CompactedRow encoding which is space-optimized
@@ -48,12 +49,6 @@
  * in a CompactedRow with a {@link CompactedRowDeserializer} and only decode to object values when a
  * field is accessed. - The record header (Length + Attributes) layout and attribute semantics are
  * the same.
- * Differences vs {@link IndexedLogRecord}:
- * - Uses CompactedRow encoding which is space-optimized (VLQ for ints/longs, per-row null bitset) and
- *   trades CPU for smaller storage; random access to fields is not supported without decoding.
- * - Deserialization is lazy: we wrap the underlying bytes in a CompactedRow with a
- *   {@link CompactedRowDeserializer} and only decode to object values when a field is accessed.
- * - The record header (Length + Attributes) layout and attribute semantics are the same.
  *
  * <p>The offset computes the difference relative to the base offset of the batch containing this
  * record.
@@ -170,24 +165,22 @@ private static int calculateSizeInBytes(BinaryRow row) {
         int size = 1; // one byte for attributes
         size += row.getSizeInBytes();
         return size;
-        int rowOffset = LENGTH_LENGTH + ATTRIBUTES_LENGTH;
-        return deserializeCompactedRow(
-                sizeInBytes - rowOffset, segment, offset + rowOffset, fieldTypes, LogFormat.COMPACTED);
     }
 
-
     private static InternalRow deserializeCompactedRow(
             int length,
             MemorySegment segment,
             int position,
             DataType[] fieldTypes,
             LogFormat logFormat) {
         if (logFormat == LogFormat.COMPACTED) {
-            CompactedRow compactedRow = new CompactedRow(fieldTypes.length, compactedDeserializer);
+            CompactedRow compactedRow =
+                    new CompactedRow(fieldTypes.length, new CompactedRowDeserializer(fieldTypes));
             compactedRow.pointTo(segment, position, length);
             return compactedRow;
         } else {
-            throw new IllegalArgumentException("No such compacted row deserializer for: " + logFormat);
+            throw new IllegalArgumentException(
+                    "No such compacted row deserializer for: " + logFormat);
         }
     }
 }

fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java

@@ -79,12 +79,6 @@ public static LogRecordReadContext createReadContext(
                     Projection.of(
                             IntStream.range(0, rowType.getFieldCount()).toArray(),
                             tableInfo.getSchema());
-        } else {
-            // Explicitly forbid any projection for COMPACTED log format
-            if (logFormat == LogFormat.COMPACTED) {
-                throw new IllegalArgumentException(
-                        "Projection is not supported for COMPACTED log format. Please remove projection.");
-            }
         }
 
         if (logFormat == LogFormat.ARROW) {