fix(flink): improve read path coverage and CDC reliability

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcImageManager.java

@@ -39,6 +39,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
 import java.util.Map;
 import java.util.TreeMap;
@@ -170,6 +171,9 @@ static final class BytesArrayInputView extends DataInputStream implements DataIn
     public void skipBytesToRead(int numBytes) throws IOException {
       while (numBytes > 0) {
         int skipped = skipBytes(numBytes);
+        if (skipped == 0) {
+          throw new EOFException("Could not skip " + numBytes + " remaining bytes");
+        }
         numBytes -= skipped;
       }
     }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcIterators.java

@@ -124,19 +124,19 @@ public CdcFileSplitsIterator(
 
     @Override
     public boolean hasNext() {
-      if (recordIterator != null) {
-        if (recordIterator.hasNext()) {
-          return true;
-        } else {
+      while (true) {
+        if (recordIterator != null) {
+          if (recordIterator.hasNext()) {
+            return true;
+          }
           recordIterator.close();
           recordIterator = null;
         }
-      }
-      if (fileSplitIterator.hasNext()) {
+        if (!fileSplitIterator.hasNext()) {
+          return false;
+        }
         recordIterator = recordIteratorFunc.apply(fileSplitIterator.next());
-        return recordIterator.hasNext();
       }
-      return false;
     }
 
     @Override

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestExpressionEvaluators.java

@@ -28,6 +28,7 @@
 import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.expressions.CallExpression;
 import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
 import org.apache.flink.table.expressions.ValueLiteralExpression;
 import org.apache.flink.table.functions.BuiltInFunctionDefinition;
 import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
@@ -45,7 +46,9 @@
 
 import static org.apache.hudi.source.ExpressionEvaluators.fromExpression;
 import static org.apache.hudi.source.prune.ColumnStatsProbe.convertColumnStats;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
@@ -393,6 +396,55 @@ void testAlwaysFalse() {
     }
   }
 
+  @Test
+  void testFromExpressionBuildsCompositeEvaluators() {
+    FieldReferenceExpression ref = new FieldReferenceExpression("f_int", DataTypes.INT(), 2, 2);
+    ValueLiteralExpression twelve = new ValueLiteralExpression(12);
+    ValueLiteralExpression thirteen = new ValueLiteralExpression(13);
+    CallExpression equals = call(BuiltInFunctionDefinitions.EQUALS, ref, twelve);
+    CallExpression greaterThan = call(BuiltInFunctionDefinitions.GREATER_THAN, ref, thirteen);
+    Map<String, ColumnStats> stats = convertColumnStats(intIndexRow(12, 12, 0L), queryFields(2));
+
+    ExpressionEvaluators.Evaluator not = fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.NOT, List.of(equals), DataTypes.BOOLEAN()));
+    ExpressionEvaluators.Evaluator and = fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.AND, Arrays.asList(equals, greaterThan), DataTypes.BOOLEAN()));
+    ExpressionEvaluators.Evaluator or = fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.OR, Arrays.asList(equals, greaterThan), DataTypes.BOOLEAN()));
+
+    assertFalse(not.eval(stats));
+    assertFalse(and.eval(stats));
+    assertTrue(or.eval(stats));
+    assertEquals(2, ((ExpressionEvaluators.Or) or).getEvaluators().length);
+    assertEquals(2, fromExpression(Arrays.<ResolvedExpression>asList(equals, greaterThan)).size());
+  }
+
+  @Test
+  void testFromExpressionHandlesUnaryInAndReversedComparison() {
+    FieldReferenceExpression ref = new FieldReferenceExpression("f_int", DataTypes.INT(), 2, 2);
+    ValueLiteralExpression eleven = new ValueLiteralExpression(11);
+    ValueLiteralExpression twelve = new ValueLiteralExpression(12);
+    Map<String, ColumnStats> stats = convertColumnStats(intIndexRow(12, 13), queryFields(2));
+
+    assertInstanceOf(ExpressionEvaluators.IsNull.class, fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.IS_NULL, List.of(ref), DataTypes.BOOLEAN())));
+    assertInstanceOf(ExpressionEvaluators.IsNotNull.class, fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.IS_NOT_NULL, List.of(ref), DataTypes.BOOLEAN())));
+
+    ExpressionEvaluators.Evaluator in = fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.IN, Arrays.asList(ref, eleven, twelve), DataTypes.BOOLEAN()));
+    assertTrue(in.eval(stats));
+
+    assertInstanceOf(ExpressionEvaluators.GreaterThan.class, fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.LESS_THAN, Arrays.asList(eleven, ref), DataTypes.BOOLEAN())));
+    assertInstanceOf(ExpressionEvaluators.LessThan.class, fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.GREATER_THAN, Arrays.asList(eleven, ref), DataTypes.BOOLEAN())));
+    assertInstanceOf(ExpressionEvaluators.GreaterThanOrEqual.class, fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL, Arrays.asList(eleven, ref), DataTypes.BOOLEAN())));
+    assertInstanceOf(ExpressionEvaluators.LessThanOrEqual.class, fromExpression(CallExpression.permanent(
+        BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL, Arrays.asList(eleven, ref), DataTypes.BOOLEAN())));
+  }
+
   @ParameterizedTest
   @MethodSource("twelveObjects")
   void testAllNumericDataTypes(Object twelve) {
@@ -410,6 +462,11 @@ public static Stream<Object> twelveObjects() {
     return Stream.of((byte) 12, (short) 12, 12, 12L, new BigDecimal(12), 12f, 12d);
   }
 
+  private static CallExpression call(
+      BuiltInFunctionDefinition definition, ResolvedExpression left, ResolvedExpression right) {
+    return CallExpression.permanent(definition, Arrays.asList(left, right), DataTypes.BOOLEAN());
+  }
+
   private static RowData intIndexRow(Integer minVal, Integer maxVal) {
     return intIndexRow(minVal, maxVal, 2L);
   }

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieSource.java

@@ -19,21 +19,28 @@
 package org.apache.hudi.source;
 
 import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.function.SerializableSupplier;
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableConfig;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.PartitionPathEncodeUtils;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.configuration.HadoopConfigurations;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.index.bucket.BucketIdentifier;
+import org.apache.hudi.source.enumerator.HoodieSplitEnumeratorState;
+import org.apache.hudi.source.enumerator.HoodieStaticSplitEnumerator;
 import org.apache.hudi.source.prune.ColumnStatsProbe;
 import org.apache.hudi.source.prune.PartitionPruners;
 import org.apache.hudi.source.reader.HoodieRecordEmitter;
 import org.apache.hudi.source.reader.function.HoodieSplitReaderFunction;
+import org.apache.hudi.source.reader.function.SplitReaderFunction;
 import org.apache.hudi.source.split.HoodieSourceSplit;
 import org.apache.hudi.source.split.HoodieSourceSplitComparator;
+import org.apache.hudi.source.split.SerializableComparator;
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
 import org.apache.hudi.table.format.InternalSchemaManager;
@@ -43,6 +50,8 @@
 import org.apache.hudi.utils.TestData;
 
 import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.data.RowData;
@@ -66,8 +75,12 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * Test cases for {@link HoodieSource}.
@@ -418,6 +431,50 @@ public void testIncrementalQueryWithPartitionPruner() throws Exception {
     assertNotNull(splits, "Incremental splits with pruner should not be null");
   }
 
+  @Test
+  @SuppressWarnings("unchecked")
+  public void testConstructorRejectsNullCollaborators() {
+    HoodieScanContext scanContext = mock(HoodieScanContext.class);
+    SerializableSupplier<SplitReaderFunction<RowData>> readerSupplier = mock(SerializableSupplier.class);
+    SerializableComparator<HoodieSourceSplit> comparator = mock(SerializableComparator.class);
+    HoodieTableMetaClient client = mock(HoodieTableMetaClient.class);
+    HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+    HoodieRecordEmitter<RowData> emitter = mock(HoodieRecordEmitter.class);
+    when(client.getTableConfig()).thenReturn(tableConfig);
+    when(tableConfig.getTableName()).thenReturn("test_table");
+
+    assertThrows(IllegalArgumentException.class,
+        () -> new HoodieSource<>(null, readerSupplier, comparator, client, emitter));
+    assertThrows(IllegalArgumentException.class,
+        () -> new HoodieSource<>(scanContext, null, comparator, client, emitter));
+    assertThrows(IllegalArgumentException.class,
+        () -> new HoodieSource<>(scanContext, readerSupplier, null, client, emitter));
+    assertThrows(IllegalArgumentException.class,
+        () -> new HoodieSource<>(scanContext, readerSupplier, comparator, null, emitter));
+    assertThrows(IllegalArgumentException.class,
+        () -> new HoodieSource<>(scanContext, readerSupplier, comparator, client, null));
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void testCreateAndRestoreStaticEnumerator() throws Exception {
+    metaClient = HoodieTestUtils.init(tempDir.getAbsolutePath(), HoodieTableType.COPY_ON_WRITE);
+    conf.set(FlinkOptions.TABLE_TYPE, HoodieTableType.COPY_ON_WRITE.name());
+    HoodieSource<RowData> source = createHoodieSource(conf, metaClient);
+    SplitEnumeratorContext<HoodieSourceSplit> context = mock(SplitEnumeratorContext.class);
+    when(context.currentParallelism()).thenReturn(1);
+
+    SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> created =
+        source.createEnumerator(context);
+    HoodieSplitEnumeratorState state = new HoodieSplitEnumeratorState(
+        Collections.emptyList(), Option.empty(), Option.empty());
+    SplitEnumerator<HoodieSourceSplit, HoodieSplitEnumeratorState> restored =
+        source.restoreEnumerator(context, state);
+
+    assertInstanceOf(HoodieStaticSplitEnumerator.class, created);
+    assertInstanceOf(HoodieStaticSplitEnumerator.class, restored);
+  }
+
   // Helper methods
 
   private HoodieSource<RowData> createHoodieSource(Configuration conf, HoodieTableMetaClient metaClient) {

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java

@@ -66,6 +66,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -811,4 +812,15 @@ void testBatchHoodieSourceSplitsWithDifferentTableTypes(HoodieTableType tableTyp
     assertNotNull(result.getSplits(), "Batch splits should not be null for table type: " + tableType);
     assertFalse(result.getSplits().isEmpty(), "Batch splits should not be empty for table type: " + tableType);
   }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  void testMergeListHandlesEmptyAndPopulatedInputs() throws Exception {
+    Method mergeList = IncrementalInputSplits.class.getDeclaredMethod("mergeList", List.class, List.class);
+    mergeList.setAccessible(true);
+
+    assertEquals(List.of(1), mergeList.invoke(null, Collections.emptyList(), List.of(1)));
+    assertEquals(List.of(1), mergeList.invoke(null, List.of(1), Collections.emptyList()));
+    assertEquals(List.of(1, 2), mergeList.invoke(null, List.of(1), List.of(2)));
+  }
 }