rebase branch

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/table/writer/AppendWriterImpl.java

@@ -90,7 +90,9 @@ public CompletableFuture<AppendResult> append(InternalRow row) {
             record = WriteRecord.forIndexedAppend(tableInfo, physicalPath, indexedRow, bucketKey);
         } else if (logFormat == LogFormat.COMPACTED) {
             CompactedRow compactedRow = encodeCompactedRow(row);
-            record = WriteRecord.forCompactedAppend(tableInfo, physicalPath, compactedRow, bucketKey);
+            record =
+                    WriteRecord.forCompactedAppend(
+                            tableInfo, physicalPath, compactedRow, bucketKey);
         } else {
             // ARROW format supports general internal row
             record = WriteRecord.forArrowAppend(tableInfo, physicalPath, row, bucketKey);

fluss-client/src/main/java/org/apache/fluss/client/write/IndexedLogWriteBatch.java

@@ -59,32 +59,35 @@ public IndexedLogWriteBatch(
                             MemoryLogRecordsIndexedBuilder.builder(
                                     schemaId, writeLimit, outputView, true);
 
-    @Override
-    public boolean isLogBatch() {
-        return true;
-    }
+                    @Override
+                    public boolean isLogBatch() {
+                        return true;
+                    }
+
+                    @Override
+                    public boolean tryAppend(WriteRecord writeRecord, WriteCallback callback)
+                            throws Exception {
+                        checkNotNull(callback, "write callback must be not null");
+                        checkNotNull(writeRecord.getRow(), "row must not be null for log record");
+                        checkArgument(
+                                writeRecord.getKey() == null, "key must be null for log record");
+                        checkArgument(
+                                writeRecord.getTargetColumns() == null,
+                                "target columns must be null for log record");
+                        checkArgument(
+                                writeRecord.getRow() instanceof IndexedRow,
+                                "row must not be IndexRow for indexed log table");
+                        IndexedRow row = (IndexedRow) writeRecord.getRow();
+                        if (!recordsBuilder.hasRoomFor(row) || isClosed()) {
+                            return false;
+                        } else {
+                            recordsBuilder.append(ChangeType.APPEND_ONLY, row);
+                            recordCount++;
+                            callbacks.add(callback);
+                            return true;
+                        }
+                    }
 
-    @Override
-    public boolean tryAppend(WriteRecord writeRecord, WriteCallback callback) throws Exception {
-        checkNotNull(callback, "write callback must be not null");
-        checkNotNull(writeRecord.getRow(), "row must not be null for log record");
-        checkArgument(writeRecord.getKey() == null, "key must be null for log record");
-        checkArgument(
-                writeRecord.getTargetColumns() == null,
-                "target columns must be null for log record");
-        checkArgument(
-                writeRecord.getRow() instanceof IndexedRow,
-                "row must not be IndexRow for indexed log table");
-        IndexedRow row = (IndexedRow) writeRecord.getRow();
-        if (!recordsBuilder.hasRoomFor(row) || isClosed()) {
-            return false;
-        } else {
-            recordsBuilder.append(ChangeType.APPEND_ONLY, row);
-            recordCount++;
-            callbacks.add(callback);
-            return true;
-        }
-    }
                     @Override
                     public boolean hasRoomFor(IndexedRow row) {
                         return delegate.hasRoomFor(row);

fluss-client/src/main/java/org/apache/fluss/client/write/WriteRecord.java

@@ -19,8 +19,8 @@
 
 import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.metadata.PhysicalTablePath;
-import org.apache.fluss.record.CompactedLogRecord;
 import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.record.CompactedLogRecord;
 import org.apache.fluss.record.DefaultKvRecord;
 import org.apache.fluss.record.IndexedLogRecord;
 import org.apache.fluss.row.BinaryRow;
@@ -53,8 +53,7 @@ public static WriteRecord forUpsert(
         checkNotNull(row, "row must not be null");
         checkNotNull(key, "key must not be null");
         checkNotNull(bucketKey, "bucketKey must not be null");
-        int estimatedSizeInBytes =
-                DefaultKvRecord.sizeOf(key, row) + RECORD_BATCH_HEADER_SIZE;
+        int estimatedSizeInBytes = DefaultKvRecord.sizeOf(key, row) + RECORD_BATCH_HEADER_SIZE;
         return new WriteRecord(
                 tableInfo,
                 tablePath,

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

@@ -233,7 +233,6 @@ public CloseableIterator<LogRecord> records(ReadContext context) {
             case INDEXED:
                 return rowRecordIterator(
                         rowType, context.getOutputProjectedRow(schemaId), timestamp);
-                return rowRecordIterator(rowType, timestamp);
             case COMPACTED:
                 return compactedRowRecordIterator(rowType, timestamp);
             default:

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

@@ -110,7 +110,7 @@ public static LogRecordReadContext createReadContext(
             int[] selectedFields = projection.getProjectionPositions();
             return createIndexedReadContext(rowType, schemaId, selectedFields, schemaGetter);
         } else if (logFormat == LogFormat.COMPACTED) {
-            int[] selectedFields = projection.getProjection();
+            int[] selectedFields = projection.getProjectionPositions();
             return createCompactedRowReadContext(rowType, schemaId, selectedFields);
         } else {
             throw new IllegalArgumentException("Unsupported log format: " + logFormat);
@@ -200,7 +200,7 @@ public static LogRecordReadContext createCompactedRowReadContext(
         FieldGetter[] fieldGetters = buildProjectedFieldGetters(rowType, selectedFields);
         // for COMPACTED log format, the projection is NEVER push downed to the server side
         return new LogRecordReadContext(
-                LogFormat.COMPACTED, rowType, schemaId, null, null, fieldGetters, false);
+                LogFormat.COMPACTED, rowType, schemaId, null, fieldGetters, false, null);
     }
 
     private LogRecordReadContext(