[common] Change class name DefaultLogRecord to IndexedLogRecord (#394)

MehulBatra · web-flow · commit 14ae7d4a8363 · 2025-02-22T10:05:38.000+08:00
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/write/WriteRecord.java b/fluss-client/src/main/java/com/alibaba/fluss/client/write/WriteRecord.java
@@ -20,8 +20,8 @@
 import com.alibaba.fluss.metadata.PhysicalTablePath;
 import com.alibaba.fluss.record.DefaultKvRecord;
 import com.alibaba.fluss.record.DefaultKvRecordBatch;
-import com.alibaba.fluss.record.DefaultLogRecord;
 import com.alibaba.fluss.record.DefaultLogRecordBatch;
+import com.alibaba.fluss.record.IndexedLogRecord;
 import com.alibaba.fluss.row.BinaryRow;
 import com.alibaba.fluss.row.InternalRow;
 import com.alibaba.fluss.row.indexed.IndexedRow;
@@ -84,7 +84,7 @@ public static WriteRecord forIndexedAppend(
             PhysicalTablePath tablePath, IndexedRow row, @Nullable byte[] bucketKey) {
         checkNotNull(row);
         int estimatedSizeInBytes =
-                DefaultLogRecord.sizeOf(row) + DefaultLogRecordBatch.RECORD_BATCH_HEADER_SIZE;
+                IndexedLogRecord.sizeOf(row) + DefaultLogRecordBatch.RECORD_BATCH_HEADER_SIZE;
         return new WriteRecord(
                 tablePath,
                 null,
diff --git a/fluss-client/src/test/java/com/alibaba/fluss/client/write/IndexedLogWriteBatchTest.java b/fluss-client/src/test/java/com/alibaba/fluss/client/write/IndexedLogWriteBatchTest.java
@@ -19,8 +19,8 @@
 import com.alibaba.fluss.memory.MemorySegment;
 import com.alibaba.fluss.memory.PreAllocatedPagedOutputView;
 import com.alibaba.fluss.metadata.TableBucket;
-import com.alibaba.fluss.record.DefaultLogRecord;
 import com.alibaba.fluss.record.DefaultLogRecordBatch;
+import com.alibaba.fluss.record.IndexedLogRecord;
 import com.alibaba.fluss.record.LogRecord;
 import com.alibaba.fluss.record.LogRecordBatch;
 import com.alibaba.fluss.record.LogRecordReadContext;
@@ -52,7 +52,7 @@ public class IndexedLogWriteBatchTest {
     @BeforeEach
     void setup() {
         row = indexedRow(DATA1_ROW_TYPE, new Object[] {1, "a"});
-        estimatedSizeInBytes = DefaultLogRecord.sizeOf(row);
+        estimatedSizeInBytes = IndexedLogRecord.sizeOf(row);
     }
 
     @Test
diff --git a/fluss-client/src/test/java/com/alibaba/fluss/client/write/RecordAccumulatorTest.java b/fluss-client/src/test/java/com/alibaba/fluss/client/write/RecordAccumulatorTest.java
@@ -31,7 +31,7 @@
 import com.alibaba.fluss.metadata.TableInfo;
 import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.record.DefaultKvRecord;
-import com.alibaba.fluss.record.DefaultLogRecord;
+import com.alibaba.fluss.record.IndexedLogRecord;
 import com.alibaba.fluss.record.LogRecord;
 import com.alibaba.fluss.record.LogRecordBatch;
 import com.alibaba.fluss.record.LogRecordReadContext;
@@ -391,7 +391,7 @@ void testAppendWithStickyBucketAssigner() throws Exception {
     void testPartialDrain() throws Exception {
         IndexedRow row = indexedRow(DATA1_ROW_TYPE, new Object[] {1, "a"});
         RecordAccumulator accum = createTestRecordAccumulator(1024, 10L * 1024);
-        int appends = 1024 / DefaultLogRecord.sizeOf(row) + 1;
+        int appends = 1024 / IndexedLogRecord.sizeOf(row) + 1;
         List<TableBucket> buckets = Arrays.asList(tb1, tb2);
         for (TableBucket tb : buckets) {
             for (int i = 0; i < appends; i++) {
@@ -597,7 +597,7 @@ private int expectedNumAppends(IndexedRow row, int batchSize) {
         int size = RECORD_BATCH_HEADER_SIZE;
         int offsetDelta = 0;
         while (true) {
-            int recordSize = DefaultLogRecord.sizeOf(row);
+            int recordSize = IndexedLogRecord.sizeOf(row);
             if (size + recordSize > batchSize) {
                 return offsetDelta;
             }
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/record/DefaultKvRecord.java b/fluss-common/src/main/java/com/alibaba/fluss/record/DefaultKvRecord.java
@@ -36,7 +36,7 @@
 import java.nio.ByteBuffer;
 
 /**
- * This class is an immutable kv record. Different from {@link DefaultLogRecord}, it isn't designed
+ * This class is an immutable kv record. Different from {@link IndexedLogRecord}, it isn't designed
  * for persistence. The schema is as follows:
  *
  * <ul>
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/record/DefaultLogRecordBatch.java b/fluss-common/src/main/java/com/alibaba/fluss/record/DefaultLogRecordBatch.java
@@ -280,8 +280,8 @@ private CloseableIterator<LogRecord> rowRecordIterator(RowType rowType, long tim
 
             @Override
             protected LogRecord readNext(long baseOffset) {
-                DefaultLogRecord logRecord =
-                        DefaultLogRecord.readFrom(
+                IndexedLogRecord logRecord =
+                        IndexedLogRecord.readFrom(
                                 segment, position, baseOffset + rowId, timestamp, fieldTypes);
                 rowId++;
                 position += logRecord.getSizeInBytes();
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/record/IndexedLogRecord.java b/fluss-common/src/main/java/com/alibaba/fluss/record/IndexedLogRecord.java
@@ -54,8 +54,7 @@
  * @since 0.1
  */
 @PublicEvolving
-// TODO: should rename to IndexedLogRecord as it only indexed row?
-public class DefaultLogRecord implements LogRecord {
+public class IndexedLogRecord implements LogRecord {
 
     private static final int ATTRIBUTES_LENGTH = 1;
 
@@ -67,7 +66,7 @@ public class DefaultLogRecord implements LogRecord {
     private int offset;
     private int sizeInBytes;
 
-    DefaultLogRecord(long logOffset, long timestamp, DataType[] fieldTypes) {
+    IndexedLogRecord(long logOffset, long timestamp, DataType[] fieldTypes) {
         this.logOffset = logOffset;
         this.fieldTypes = fieldTypes;
         this.timestamp = timestamp;
@@ -93,7 +92,7 @@ public boolean equals(Object o) {
             return false;
         }
 
-        DefaultLogRecord that = (DefaultLogRecord) o;
+        IndexedLogRecord that = (IndexedLogRecord) o;
         return sizeInBytes == that.sizeInBytes
                 && segment.equalTo(that.segment, offset, that.offset, sizeInBytes);
     }
@@ -149,14 +148,14 @@ public static int writeTo(OutputView outputView, RowKind rowKind, IndexedRow row
         return sizeInBytes + LENGTH_LENGTH;
     }
 
-    public static DefaultLogRecord readFrom(
+    public static IndexedLogRecord readFrom(
             MemorySegment segment,
             int position,
             long logOffset,
             long logTimestamp,
             DataType[] colTypes) {
         int sizeInBytes = segment.getInt(position);
-        DefaultLogRecord logRecord = new DefaultLogRecord(logOffset, logTimestamp, colTypes);
+        IndexedLogRecord logRecord = new IndexedLogRecord(logOffset, logTimestamp, colTypes);
         logRecord.pointTo(segment, position, sizeInBytes + LENGTH_LENGTH);
         return logRecord;
     }
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/record/MemoryLogRecordsIndexedBuilder.java b/fluss-common/src/main/java/com/alibaba/fluss/record/MemoryLogRecordsIndexedBuilder.java
@@ -109,7 +109,7 @@ public static MemoryLogRecordsIndexedBuilder builder(
      * appended, then this returns true.
      */
     public boolean hasRoomFor(IndexedRow row) {
-        return sizeInBytes + DefaultLogRecord.sizeOf(row) <= writeLimit;
+        return sizeInBytes + IndexedLogRecord.sizeOf(row) <= writeLimit;
     }
 
     public void append(RowKind rowKind, IndexedRow row) throws Exception {
@@ -122,7 +122,7 @@ private void appendRecord(RowKind rowKind, IndexedRow row) throws IOException {
                     "Tried to append a record, but MemoryLogRecordsBuilder is closed for record appends");
         }
 
-        int recordByteSizes = DefaultLogRecord.writeTo(pagedOutputView, rowKind, row);
+        int recordByteSizes = IndexedLogRecord.writeTo(pagedOutputView, rowKind, row);
         currentRecordNumber++;
         sizeInBytes += recordByteSizes;
     }
diff --git a/fluss-common/src/test/java/com/alibaba/fluss/record/IndexedLogRecordTest.java b/fluss-common/src/test/java/com/alibaba/fluss/record/IndexedLogRecordTest.java
@@ -29,8 +29,8 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-/** Test for {@link DefaultLogRecord}. */
-class DefaultLogRecordTest extends LogTestBase {
+/** Test for {@link IndexedLogRecord}. */
+class IndexedLogRecordTest extends LogTestBase {
 
     @Test
     void testBase() throws IOException {
@@ -43,10 +43,10 @@ void testBase() throws IOException {
         writer.writeString(BinaryString.fromString("abc"));
         row.pointTo(writer.segment(), 0, writer.position());
 
-        DefaultLogRecord.writeTo(outputView, RowKind.APPEND_ONLY, row);
+        IndexedLogRecord.writeTo(outputView, RowKind.APPEND_ONLY, row);
         // Test read from.
-        DefaultLogRecord defaultLogRecord =
-                DefaultLogRecord.readFrom(
+        IndexedLogRecord defaultLogRecord =
+                IndexedLogRecord.readFrom(
                         MemorySegment.wrap(outputView.getCopyOfBuffer()),
                         0,
                         1000,
@@ -64,13 +64,13 @@ void testBase() throws IOException {
     void testWriteToAndReadFromWithRandomData() throws IOException {
         // Test write to.
         IndexedRow row = TestInternalRowGenerator.genIndexedRowForAllType();
-        DefaultLogRecord.writeTo(outputView, RowKind.APPEND_ONLY, row);
+        IndexedLogRecord.writeTo(outputView, RowKind.APPEND_ONLY, row);
         DataType[] allColTypes =
                 TestInternalRowGenerator.createAllRowType().getChildren().toArray(new DataType[0]);
 
         // Test read from.
         LogRecord defaultLogRecord =
-                DefaultLogRecord.readFrom(
+                IndexedLogRecord.readFrom(
                         MemorySegment.wrap(outputView.getCopyOfBuffer()),
                         0,
                         1000,
diff --git a/fluss-common/src/test/java/com/alibaba/fluss/record/LogTestBase.java b/fluss-common/src/test/java/com/alibaba/fluss/record/LogTestBase.java
@@ -83,8 +83,8 @@ protected void assertIndexedLogRecordBatchAndRowEquals(
                 CloseableIterator<LogRecord> expectIter = expected.records(readContext)) {
             int i = 0;
             while (actualIter.hasNext() && expectIter.hasNext()) {
-                DefaultLogRecord actualRecord = (DefaultLogRecord) actualIter.next();
-                DefaultLogRecord expectedRecord = (DefaultLogRecord) expectIter.next();
+                IndexedLogRecord actualRecord = (IndexedLogRecord) actualIter.next();
+                IndexedLogRecord expectedRecord = (IndexedLogRecord) expectIter.next();
                 assertIndexedRecordEquals(actualRecord, expectedRecord, rows.get(i), i);
                 i++;
             }
diff --git a/fluss-connectors/fluss-connector-flink/src/test/java/com/alibaba/fluss/connector/flink/catalog/FlinkCatalogTest.java b/fluss-connectors/fluss-connector-flink/src/test/java/com/alibaba/fluss/connector/flink/catalog/FlinkCatalogTest.java
@@ -124,6 +124,10 @@ static void afterAll() {
 
     @BeforeEach
     void beforeEach() throws Exception {
+        // First check if database exists, and drop it if it does
+        if (catalog.databaseExists(DEFAULT_DB)) {
+            catalog.dropDatabase(DEFAULT_DB, true, true);
+        }
         try {
             catalog.createDatabase(
                     DEFAULT_DB, new CatalogDatabaseImpl(Collections.emptyMap(), null), true);

Original file line number	Diff line number	Diff line change
`@@ -36,7 +36,7 @@`
`36`	`36`	`import java.nio.ByteBuffer;`
`37`	`37`
`38`	`38`	`/**`
`39`		`- * This class is an immutable kv record. Different from {@link DefaultLogRecord}, it isn't designed`
	`39`	`+ * This class is an immutable kv record. Different from {@link IndexedLogRecord}, it isn't designed`
`40`	`40`	`* for persistence. The schema is as follows:`
`41`	`41`	`*`
`42`	`42`	`* <ul>`
Original file line number	Diff line number	Diff line change
`@@ -109,7 +109,7 @@ public static MemoryLogRecordsIndexedBuilder builder(`
`109`	`109`	`* appended, then this returns true.`
`110`	`110`	`*/`
`111`	`111`	`public boolean hasRoomFor(IndexedRow row) {`
`112`		`- return sizeInBytes + DefaultLogRecord.sizeOf(row) <= writeLimit;`
	`112`	`+ return sizeInBytes + IndexedLogRecord.sizeOf(row) <= writeLimit;`
`113`	`113`	`}`
`114`	`114`
`115`	`115`	`public void append(RowKind rowKind, IndexedRow row) throws Exception {`
`@@ -122,7 +122,7 @@ private void appendRecord(RowKind rowKind, IndexedRow row) throws IOException {`
`122`	`122`	`"Tried to append a record, but MemoryLogRecordsBuilder is closed for record appends");`
`123`	`123`	`}`
`124`	`124`
`125`		`- int recordByteSizes = DefaultLogRecord.writeTo(pagedOutputView, rowKind, row);`
	`125`	`+ int recordByteSizes = IndexedLogRecord.writeTo(pagedOutputView, rowKind, row);`
`126`	`126`	`currentRecordNumber++;`
`127`	`127`	`sizeInBytes += recordByteSizes;`
`128`	`128`	`}`