add missing types to flink module

Author: polyzos

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/reader/LakeSnapshotAndLogSplitScanner.java

@@ -67,7 +67,7 @@ public class LakeSnapshotAndLogSplitScanner implements BatchScanner {
     // the sorted logs in memory, mapping from key -> value
     private Map<InternalRow, KeyValueRow> logRows;
 
-    private final LogScanner logScanner;
+    private final LogScanner<InternalRow> logScanner;
     private final long stoppingOffset;
     private boolean logScanFinished;
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java

@@ -38,6 +38,7 @@
 import org.apache.fluss.lake.source.LakeSplit;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.types.RowType;
 import org.apache.fluss.utils.CloseableIterator;
 import org.apache.fluss.utils.ExceptionUtils;
@@ -407,7 +408,7 @@ private void checkSnapshotSplitOrStartNext() {
         }
     }
 
-    private FlinkRecordsWithSplitIds forLogRecords(ScanRecords scanRecords) {
+    private FlinkRecordsWithSplitIds forLogRecords(ScanRecords<InternalRow> scanRecords) {
         // For calculating the currentFetchEventTimeLag
         long fetchTimestamp = System.currentTimeMillis();
         long maxConsumerRecordTimestampInFetch = -1;
@@ -426,7 +427,7 @@ private FlinkRecordsWithSplitIds forLogRecords(ScanRecords scanRecords) {
             }
             splitIdByTableBucket.put(scanBucket, splitId);
             tableScanBuckets.add(scanBucket);
-            List<ScanRecord> bucketScanRecords = scanRecords.records(scanBucket);
+            List<ScanRecord<InternalRow>> bucketScanRecords = scanRecords.records(scanBucket);
             if (!bucketScanRecords.isEmpty()) {
                 final ScanRecord lastRecord = bucketScanRecords.get(bucketScanRecords.size() - 1);
                 // We keep the maximum message timestamp in the fetch for calculating lags
@@ -478,7 +479,7 @@ public String next() {
     }
 
     private CloseableIterator<RecordAndPos> toRecordAndPos(
-            Iterator<ScanRecord> recordAndPosIterator) {
+            Iterator<ScanRecord<InternalRow>> recordAndPosIterator) {
         return new CloseableIterator<RecordAndPos>() {
 
             @Override

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

@@ -32,6 +32,9 @@
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.utils.CloseableIterator;
 
 import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
@@ -249,11 +252,12 @@ private void mayCreateLogScanner() {
     }
 
     private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
-            ScanRecords scanRecords) throws IOException {
+            ScanRecords<org.apache.fluss.row.InternalRow> scanRecords) throws IOException {
         Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
         Map<TableBucket, String> finishedSplitIds = new HashMap<>();
         for (TableBucket bucket : scanRecords.buckets()) {
-            List<ScanRecord> bucketScanRecords = scanRecords.records(bucket);
+            List<ScanRecord<org.apache.fluss.row.InternalRow>> bucketScanRecords =
+                    scanRecords.records(bucket);
             if (bucketScanRecords.isEmpty()) {
                 continue;
             }
@@ -265,13 +269,14 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
             LakeWriter<WriteResult> lakeWriter =
                     getOrCreateLakeWriter(
                             bucket, currentTableSplitsByBucket.get(bucket).getPartitionName());
-            for (ScanRecord record : bucketScanRecords) {
+            for (ScanRecord<InternalRow> record : bucketScanRecords) {
                 // if record is less than stopping offset
                 if (record.logOffset() < stoppingOffset) {
-                    lakeWriter.write(record);
+                    lakeWriter.write(new ScanRecordLogRecord(record));
                 }
             }
-            ScanRecord lastRecord = bucketScanRecords.get(bucketScanRecords.size() - 1);
+            ScanRecord<InternalRow> lastRecord =
+                    bucketScanRecords.get(bucketScanRecords.size() - 1);
             // has arrived into the end of the split,
             if (lastRecord.logOffset() >= stoppingOffset - 1) {
                 currentTableStoppingOffsets.remove(bucket);
@@ -390,8 +395,8 @@ private TableBucketWriteResultWithSplitIds forSnapshotSplitRecords(
                 getOrCreateLakeWriter(
                         bucket, checkNotNull(currentSnapshotSplit).getPartitionName());
         while (recordIterator.hasNext()) {
-            ScanRecord scanRecord = recordIterator.next().record();
-            lakeWriter.write(scanRecord);
+            ScanRecord<InternalRow> scanRecord = recordIterator.next().record();
+            lakeWriter.write(new ScanRecordLogRecord(scanRecord));
         }
         recordIterator.close();
         return emptyTableBucketWriteResultWithSplitIds();
@@ -460,6 +465,35 @@ public void close() throws Exception {
         // don't need to close connection, will be closed by TieringSourceReader
     }
 
+    /** Lightweight adapter to view a {@code ScanRecord<InternalRow>} as a {@link LogRecord}. */
+    private static final class ScanRecordLogRecord implements LogRecord {
+        private final ScanRecord<InternalRow> delegate;
+
+        private ScanRecordLogRecord(ScanRecord<InternalRow> delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public long logOffset() {
+            return delegate.logOffset();
+        }
+
+        @Override
+        public long timestamp() {
+            return delegate.timestamp();
+        }
+
+        @Override
+        public ChangeType getChangeType() {
+            return delegate.getChangeType();
+        }
+
+        @Override
+        public InternalRow getRow() {
+            return delegate.getRow();
+        }
+    }
+
     private void subscribeLog(TieringLogSplit logSplit) {
         // assign bucket offset dynamically
         TableBucket tableBucket = logSplit.getTableBucket();