[GLUTEN-12206][FLINK]Support filesystem writer bucket state snapshot and restore

gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java

@@ -331,20 +331,27 @@ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
 
   @Override
   public void snapshotState(StateSnapshotContext context) throws Exception {
-    // TODO: implement it
-    task.snapshotState(context.getCheckpointId());
+    snapshotNativeState(context);
     super.snapshotState(context);
   }
 
+  protected void snapshotNativeState(StateSnapshotContext context) throws Exception {
+    task.snapshotState(context.getCheckpointId());
+  }
+
   @Override
   public void initializeState(StateInitializationContext context) throws Exception {
+    initializeNativeState(context);
+    super.initializeState(context);
+  }
+
+  protected void initializeNativeState(StateInitializationContext context) throws Exception {
     if (task == null) {
       initSession();
     }
     if (!(getKeyedStateBackend() instanceof RocksDBKeyedStateBackend)) {
       task.initializeState(0, null);
     }
-    super.initializeState(context);
   }
 
   @Override

gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenStreamingFileWriterOperator.java

@@ -22,15 +22,31 @@
 import io.github.zhztheplayer.velox4j.type.RowType;
 
 import org.apache.flink.api.common.TaskInfo;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
 import org.apache.flink.connector.file.table.stream.PartitionCommitInfo;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 
 public class GlutenStreamingFileWriterOperator<IN>
     extends GlutenOneInputOperator<IN, PartitionCommitInfo> {
 
+  private static final Logger LOG =
+      LoggerFactory.getLogger(GlutenStreamingFileWriterOperator.class);
+
+  private transient ListState<String> checkpointState;
+  private transient String[] restoredCheckpointRecords = new String[0];
+
   public GlutenStreamingFileWriterOperator(
       StatefulPlanNode plan,
       String id,
@@ -56,6 +72,44 @@ public void processWatermark(Watermark mark) throws Exception {
     output.emitWatermark(mark);
   }
 
+  @Override
+  protected void snapshotNativeState(StateSnapshotContext context) throws Exception {
+    checkpointState.clear();
+    String[] checkpointRecords = task.snapshotState(context.getCheckpointId());
+    for (String checkpointRecord : checkpointRecords) {
+      checkpointState.add(checkpointRecord);
+    }
+  }
+
+  @Override
+  protected void initializeNativeState(StateInitializationContext context) throws Exception {
+    checkpointState =
+        context
+            .getOperatorStateStore()
+            .getListState(
+                new ListStateDescriptor<>("gluten-native-file-writer-checkpoint", String.class));
+    if (context.isRestored()) {
+      List<String> records = new ArrayList<>();
+      for (String checkpointRecord : checkpointState.get()) {
+        records.add(checkpointRecord);
+      }
+      restoredCheckpointRecords = records.toArray(new String[0]);
+    }
+    if (restoredCheckpointRecords != null) {
+      LOG.info(
+          "Restore native file writer state for operator {}, restored {}, records {}, contents {}",
+          getDescription(),
+          context.isRestored(),
+          restoredCheckpointRecords.length,
+          Arrays.toString(
+              restoredCheckpointRecords != null ? restoredCheckpointRecords : new String[0]));
+    }
+    if (task == null) {
+      initSession();
+    }
+    task.initializeState(0, null, restoredCheckpointRecords);
+  }
+
   @Override
   public void notifyCheckpointComplete(long checkpointId) throws Exception {
     String[] committed = task.notifyCheckpointComplete(checkpointId);