[FLINK-38930][checkpoint] Filtering record before processing without spilling strategy

Core filtering mechanism for recovered channel state buffers:
- ChannelStateFilteringHandler with per-gate GateFilterHandler
- RecordFilterContext with VirtualChannelRecordFilterFactory
- Partial data check in SequentialChannelStateReaderImpl
- Fix RecordFilterContext for Union downscale scenario

Author: 1996fanrui

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java

@@ -31,6 +31,7 @@
 import org.apache.flink.runtime.io.network.partition.consumer.RecoveredInputChannel;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -63,7 +64,7 @@ public void close() {
      * case of an error.
      */
     void recover(Info info, int oldSubtaskIndex, BufferWithContext<Context> bufferWithContext)
-            throws IOException;
+            throws IOException, InterruptedException;
 }
 
 class InputChannelRecoveredStateHandler
@@ -75,10 +76,19 @@ class InputChannelRecoveredStateHandler
     private final Map<InputChannelInfo, RecoveredInputChannel> rescaledChannels = new HashMap<>();
     private final Map<Integer, RescaleMappings> oldToNewMappings = new HashMap<>();
 
+    /**
+     * Optional filtering handler for filtering recovered buffers. When non-null, filtering is
+     * performed during recovery in the channel-state-unspilling thread.
+     */
+    @Nullable private final ChannelStateFilteringHandler filteringHandler;
+
     InputChannelRecoveredStateHandler(
-            InputGate[] inputGates, InflightDataRescalingDescriptor channelMapping) {
+            InputGate[] inputGates,
+            InflightDataRescalingDescriptor channelMapping,
+            @Nullable ChannelStateFilteringHandler filteringHandler) {
         this.inputGates = inputGates;
         this.channelMapping = channelMapping;
+        this.filteringHandler = filteringHandler;
     }
 
     @Override
@@ -95,23 +105,60 @@ public void recover(
             InputChannelInfo channelInfo,
             int oldSubtaskIndex,
             BufferWithContext<Buffer> bufferWithContext)
-            throws IOException {
+            throws IOException, InterruptedException {
         Buffer buffer = bufferWithContext.context;
         try {
             if (buffer.readableBytes() > 0) {
                 RecoveredInputChannel channel = getMappedChannels(channelInfo);
-                channel.onRecoveredStateBuffer(
-                        EventSerializer.toBuffer(
-                                new SubtaskConnectionDescriptor(
-                                        oldSubtaskIndex, channelInfo.getInputChannelIdx()),
-                                false));
-                channel.onRecoveredStateBuffer(buffer.retainBuffer());
+
+                if (filteringHandler != null) {
+                    // Filtering mode: filter records and rewrite to new buffers
+                    recoverWithFiltering(channel, channelInfo, oldSubtaskIndex, buffer);
+                } else {
+                    // Non-filtering mode: pass through original buffer with descriptor
+                    channel.onRecoveredStateBuffer(
+                            EventSerializer.toBuffer(
+                                    new SubtaskConnectionDescriptor(
+                                            oldSubtaskIndex, channelInfo.getInputChannelIdx()),
+                                    false));
+                    channel.onRecoveredStateBuffer(buffer.retainBuffer());
+                }
             }
         } finally {
             buffer.recycleBuffer();
         }
     }
 
+    private void recoverWithFiltering(
+            RecoveredInputChannel channel,
+            InputChannelInfo channelInfo,
+            int oldSubtaskIndex,
+            Buffer buffer)
+            throws IOException, InterruptedException {
+        checkState(filteringHandler != null, "filtering handler not set.");
+        // Extra retain: filterAndRewrite consumes one ref, caller's finally releases another.
+        buffer.retainBuffer();
+
+        List<Buffer> filteredBuffers;
+        try {
+            filteredBuffers =
+                    filteringHandler.filterAndRewrite(
+                            channelInfo.getGateIdx(),
+                            oldSubtaskIndex,
+                            channelInfo.getInputChannelIdx(),
+                            buffer,
+                            channel::requestBufferBlocking);
+        } catch (Throwable t) {
+            // filterAndRewrite didn't consume the buffer, release the extra ref.
+            buffer.recycleBuffer();
+            throw t;
+        }
+
+        for (Buffer filteredBuffer : filteredBuffers) {
+            channel.onRecoveredStateBuffer(filteredBuffer);
+        }
+    }
+
     @Override
     public void close() throws IOException {
         // note that we need to finish all RecoveredInputChannels, not just those with state
@@ -191,7 +238,7 @@ public void recover(
             ResultSubpartitionInfo subpartitionInfo,
             int oldSubtaskIndex,
             BufferWithContext<BufferBuilder> bufferWithContext)
-            throws IOException {
+            throws IOException, InterruptedException {
         try (BufferBuilder bufferBuilder = bufferWithContext.context;
                 BufferConsumer bufferConsumer = bufferBuilder.createBufferConsumerFromBeginning()) {
             bufferBuilder.finish();

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java

@@ -20,14 +20,22 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
 import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
+import org.apache.flink.streaming.runtime.io.recovery.RecordFilterContext;
 
 import java.io.IOException;
 
 /** Reads channel state saved during checkpoint/savepoint. */
 @Internal
 public interface SequentialChannelStateReader extends AutoCloseable {
 
-    void readInputData(InputGate[] inputGates) throws IOException, InterruptedException;
+    /**
+     * Reads input channel state with filtering support.
+     *
+     * @param inputGates The input gates to recover state for.
+     * @param filterContext The filter context containing input configs and rescaling info.
+     */
+    void readInputData(InputGate[] inputGates, RecordFilterContext filterContext)
+            throws IOException, InterruptedException;
 
     void readOutputData(ResultPartitionWriter[] writers, boolean notifyAndBlockOnCompletion)
             throws IOException, InterruptedException;
@@ -39,7 +47,8 @@ void readOutputData(ResultPartitionWriter[] writers, boolean notifyAndBlockOnCom
             new SequentialChannelStateReader() {
 
                 @Override
-                public void readInputData(InputGate[] inputGates) {}
+                public void readInputData(
+                        InputGate[] inputGates, RecordFilterContext filterContext) {}
 
                 @Override
                 public void readOutputData(

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java

@@ -28,6 +28,7 @@
 import org.apache.flink.runtime.state.AbstractChannelStateHandle;
 import org.apache.flink.runtime.state.ChannelStateHelper;
 import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.streaming.runtime.io.recovery.RecordFilterContext;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -43,6 +44,7 @@
 import static java.util.Comparator.comparingLong;
 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.toList;
+import static org.apache.flink.util.Preconditions.checkState;
 
 /** {@link SequentialChannelStateReader} implementation. */
 public class SequentialChannelStateReaderImpl implements SequentialChannelStateReader {
@@ -58,10 +60,21 @@ public SequentialChannelStateReaderImpl(TaskStateSnapshot taskStateSnapshot) {
     }
 
     @Override
-    public void readInputData(InputGate[] inputGates) throws IOException, InterruptedException {
+    public void readInputData(InputGate[] inputGates, RecordFilterContext filterContext)
+            throws IOException, InterruptedException {
+
+        // Create filtering handler if filtering is needed
+        ChannelStateFilteringHandler filteringHandler = null;
+        if (filterContext.isUnalignedDuringRecoveryEnabled()) {
+            filteringHandler =
+                    ChannelStateFilteringHandler.createFromContext(filterContext, inputGates);
+        }
+
         try (InputChannelRecoveredStateHandler stateHandler =
                 new InputChannelRecoveredStateHandler(
-                        inputGates, taskStateSnapshot.getInputRescalingDescriptor())) {
+                        inputGates,
+                        taskStateSnapshot.getInputRescalingDescriptor(),
+                        filteringHandler)) {
             read(
                     stateHandler,
                     groupByDelegate(
@@ -72,6 +85,18 @@ public void readInputData(InputGate[] inputGates) throws IOException, Interrupte
                     groupByDelegate(
                             streamSubtaskStates(),
                             OperatorSubtaskState::getUpstreamOutputBufferState));
+
+            if (filteringHandler != null) {
+                checkState(
+                        !filteringHandler.hasPartialData(),
+                        "Not all data has been fully consumed during filtering");
+            }
+        } finally {
+            // Clean up filtering handler resources (e.g., temp files from
+            // SpillingAdaptiveSpanningRecordDeserializer) on both success and error paths
+            if (filteringHandler != null) {
+                filteringHandler.clear();
+            }
         }
     }
 

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java

@@ -23,6 +23,7 @@
 import org.apache.flink.api.common.TaskInfo;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.CheckpointingOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.memory.ManagedMemoryUseCase;
 import org.apache.flink.metrics.Counter;
@@ -103,6 +104,10 @@ public static StreamMultipleInputProcessor create(
                 "Number of configured inputs in StreamConfig [%s] doesn't match the main operator's number of inputs [%s]",
                 configuredInputs.length,
                 inputsCount);
+
+        boolean unalignedDuringRecoveryEnabled =
+                CheckpointingOptions.isUnalignedDuringRecoveryEnabled(jobConfig);
+
         StreamTaskInput[] inputs = new StreamTaskInput[inputsCount];
         for (int i = 0; i < inputsCount; i++) {
             StreamConfig.InputConfig configuredInput = configuredInputs[i];
@@ -121,7 +126,8 @@ public static StreamMultipleInputProcessor create(
                                 gatePartitioners,
                                 taskInfo,
                                 canEmitBatchOfRecords,
-                                streamConfig.getWatermarkDeclarations(userClassloader));
+                                streamConfig.getWatermarkDeclarations(userClassloader),
+                                unalignedDuringRecoveryEnabled);
             } else if (configuredInput instanceof StreamConfig.SourceInputConfig) {
                 StreamConfig.SourceInputConfig sourceInput =
                         (StreamConfig.SourceInputConfig) configuredInput;

flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java

@@ -47,9 +47,14 @@ public static <T> StreamTaskInput<T> create(
             Function<Integer, StreamPartitioner<?>> gatePartitioners,
             TaskInfo taskInfo,
             CanEmitBatchOfRecordsChecker canEmitBatchOfRecords,
-            Set<AbstractInternalWatermarkDeclaration<?>> watermarkDeclarationSet) {
+            Set<AbstractInternalWatermarkDeclaration<?>> watermarkDeclarationSet,
+            boolean unalignedDuringRecoveryEnabled) {
         return rescalingDescriptorinflightDataRescalingDescriptor.equals(
-                        InflightDataRescalingDescriptor.NO_RESCALE)
+                                InflightDataRescalingDescriptor.NO_RESCALE)
+                        // When filter during recovery is enabled, records are already filtered in
+                        // the channel-state-unspilling thread. Use StreamTaskNetworkInput to avoid
+                        // redundant demultiplexing/filtering in the Task thread.
+                        || unalignedDuringRecoveryEnabled
                 ? new StreamTaskNetworkInput<>(
                         checkpointedInputGate,
                         inputSerializer,

flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/StreamTaskNetworkInputFactory.java

@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.TaskInfo;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.CheckpointingOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.memory.ManagedMemoryUseCase;
 import org.apache.flink.metrics.Counter;
@@ -84,6 +85,10 @@ public static <IN1, IN2> StreamMultipleInputProcessor create(
         checkNotNull(operatorChain);
 
         taskIOMetricGroup.reuseRecordsInputCounter(numRecordsIn);
+
+        boolean unalignedDuringRecoveryEnabled =
+                CheckpointingOptions.isUnalignedDuringRecoveryEnabled(jobConfig);
+
         TypeSerializer<IN1> typeSerializer1 = streamConfig.getTypeSerializerIn(0, userClassloader);
         StreamTaskInput<IN1> input1 =
                 StreamTaskNetworkInputFactory.create(
@@ -96,7 +101,8 @@ public static <IN1, IN2> StreamMultipleInputProcessor create(
                         gatePartitioners,
                         taskInfo,
                         canEmitBatchOfRecords,
-                        streamConfig.getWatermarkDeclarations(userClassloader));
+                        streamConfig.getWatermarkDeclarations(userClassloader),
+                        unalignedDuringRecoveryEnabled);
         TypeSerializer<IN2> typeSerializer2 = streamConfig.getTypeSerializerIn(1, userClassloader);
         StreamTaskInput<IN2> input2 =
                 StreamTaskNetworkInputFactory.create(
@@ -109,7 +115,8 @@ public static <IN1, IN2> StreamMultipleInputProcessor create(
                         gatePartitioners,
                         taskInfo,
                         canEmitBatchOfRecords,
-                        streamConfig.getWatermarkDeclarations(userClassloader));
+                        streamConfig.getWatermarkDeclarations(userClassloader),
+                        unalignedDuringRecoveryEnabled);
 
         InputSelectable inputSelectable =
                 streamOperator instanceof InputSelectable ? (InputSelectable) streamOperator : null;